forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 02:42:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

swarm/storage/localstore: rename ChunkInfo to ChunkDescriptor

Browse source
This commit is contained in:
Janos Guljas 2019-01-11 09:27:27 +01:00
parent c5a645637c
commit 33726a4ba3
2 changed files with 24 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
swarm/storage/localstore/subscription_pull.go
View file

@ -35,8 +35,8 @@ import (
// function will terminate current and further iterations without errors, and also close the returned channel. // function will terminate current and further iterations without errors, and also close the returned channel.
// Make sure that you check the second returned parameter from the channel to stop iteration when its value // Make sure that you check the second returned parameter from the channel to stop iteration when its value
// is false. // is false.
func (db *DB) SubscribePull(ctx context.Context, bin uint8, since, until *ChunkInfo) (c <-chan ChunkInfo, stop func()) { func (db *DB) SubscribePull(ctx context.Context, bin uint8, since, until *ChunkDescriptor) (c <-chan ChunkDescriptor, stop func()) {
chunkInfos := make(chan ChunkInfo) chunkDescriptors := make(chan ChunkDescriptor)
trigger := make(chan struct{}, 1) trigger := make(chan struct{}, 1)
db.pullTriggersMu.Lock() db.pullTriggersMu.Lock()
@ -53,13 +53,13 @@ func (db *DB) SubscribePull(ctx context.Context, bin uint8, since, until *ChunkI
var stopChanOnce sync.Once var stopChanOnce sync.Once
// used to provide information from the iterator to // used to provide information from the iterator to
// stop subscription when until chunk info is reached // stop subscription when until chunk descriptor is reached
var errStopSubscription = errors.New("stop subscription") var errStopSubscription = errors.New("stop subscription")
go func() { go func() {
// close the returned chunkInfo channel at the end to // close the returned ChunkDescriptor channel at the end to
// signal that the subscription is done // signal that the subscription is done
defer close(chunkInfos) defer close(chunkDescriptors)
// sinceItem is the Item from which the next iteration // sinceItem is the Item from which the next iteration
// should start. The first iteration starts from the first Item. // should start. The first iteration starts from the first Item.
var sinceItem *shed.Item var sinceItem *shed.Item
@ -78,11 +78,11 @@ func (db *DB) SubscribePull(ctx context.Context, bin uint8, since, until *ChunkI
// - context is done // - context is done
err := db.pullIndex.Iterate(func(item shed.Item) (stop bool, err error) { err := db.pullIndex.Iterate(func(item shed.Item) (stop bool, err error) {
select { select {
case chunkInfos <- ChunkInfo{ case chunkDescriptors <- ChunkDescriptor{
Address: item.Address, Address: item.Address,
StoreTimestamp: item.StoreTimestamp, StoreTimestamp: item.StoreTimestamp,
}: }:
// until chunk info is sent // until chunk descriptor is sent
// break the iteration // break the iteration
if until != nil && if until != nil &&
(item.StoreTimestamp >= until.StoreTimestamp || (item.StoreTimestamp >= until.StoreTimestamp ||
@ -154,12 +154,12 @@ func (db *DB) SubscribePull(ctx context.Context, bin uint8, since, until *ChunkI
} }
} }
return chunkInfos, stop return chunkDescriptors, stop
} }
// ChunkInfo holds information required for Pull syncing. This struct // ChunkDescriptor holds information required for Pull syncing. This struct
// is provided by subscribing to pull index. // is provided by subscribing to pull index.
type ChunkInfo struct { type ChunkDescriptor struct {
Address storage.Address Address storage.Address
StoreTimestamp int64 StoreTimestamp int64
} }

28
swarm/storage/localstore/subscription_pull_test.go
View file

@ -147,8 +147,8 @@ func TestDB_SubscribePull_since(t *testing.T) {
return atomic.AddInt64(&lastTimestamp, 1) return atomic.AddInt64(&lastTimestamp, 1)
})() })()
uploadRandomChunks := func(count int, wanted bool) (last map[uint8]ChunkInfo) { uploadRandomChunks := func(count int, wanted bool) (last map[uint8]ChunkDescriptor) {
last = make(map[uint8]ChunkInfo) last = make(map[uint8]ChunkDescriptor)
for i := 0; i < count; i++ { for i := 0; i < count; i++ {
chunk := generateRandomChunk() chunk := generateRandomChunk()
@ -167,7 +167,7 @@ func TestDB_SubscribePull_since(t *testing.T) {
wantedChunksCount++ wantedChunksCount++
} }
last[bin] = ChunkInfo{ last[bin] = ChunkDescriptor{
Address: chunk.Address(), Address: chunk.Address(),
StoreTimestamp: atomic.LoadInt64(&lastTimestamp), StoreTimestamp: atomic.LoadInt64(&lastTimestamp),
} }
@ -190,7 +190,7 @@ func TestDB_SubscribePull_since(t *testing.T) {
errChan := make(chan error) errChan := make(chan error)
for bin := uint8(0); bin <= uint8(storage.MaxPO); bin++ { for bin := uint8(0); bin <= uint8(storage.MaxPO); bin++ {
var since *ChunkInfo var since *ChunkDescriptor
if c, ok := last[bin]; ok { if c, ok := last[bin]; ok {
since = &c since = &c
} }
@ -226,8 +226,8 @@ func TestDB_SubscribePull_until(t *testing.T) {
return atomic.AddInt64(&lastTimestamp, 1) return atomic.AddInt64(&lastTimestamp, 1)
})() })()
uploadRandomChunks := func(count int, wanted bool) (last map[uint8]ChunkInfo) { uploadRandomChunks := func(count int, wanted bool) (last map[uint8]ChunkDescriptor) {
last = make(map[uint8]ChunkInfo) last = make(map[uint8]ChunkDescriptor)
for i := 0; i < count; i++ { for i := 0; i < count; i++ {
chunk := generateRandomChunk() chunk := generateRandomChunk()
@ -246,7 +246,7 @@ func TestDB_SubscribePull_until(t *testing.T) {
wantedChunksCount++ wantedChunksCount++
} }
last[bin] = ChunkInfo{ last[bin] = ChunkDescriptor{
Address: chunk.Address(), Address: chunk.Address(),
StoreTimestamp: atomic.LoadInt64(&lastTimestamp), StoreTimestamp: atomic.LoadInt64(&lastTimestamp),
} }
@ -306,8 +306,8 @@ func TestDB_SubscribePull_sinceAndUntil(t *testing.T) {
return atomic.AddInt64(&lastTimestamp, 1) return atomic.AddInt64(&lastTimestamp, 1)
})() })()
uploadRandomChunks := func(count int, wanted bool) (last map[uint8]ChunkInfo) { uploadRandomChunks := func(count int, wanted bool) (last map[uint8]ChunkDescriptor) {
last = make(map[uint8]ChunkInfo) last = make(map[uint8]ChunkDescriptor)
for i := 0; i < count; i++ { for i := 0; i < count; i++ {
chunk := generateRandomChunk() chunk := generateRandomChunk()
@ -326,7 +326,7 @@ func TestDB_SubscribePull_sinceAndUntil(t *testing.T) {
wantedChunksCount++ wantedChunksCount++
} }
last[bin] = ChunkInfo{ last[bin] = ChunkDescriptor{
Address: chunk.Address(), Address: chunk.Address(),
StoreTimestamp: atomic.LoadInt64(&lastTimestamp), StoreTimestamp: atomic.LoadInt64(&lastTimestamp),
} }
@ -355,7 +355,7 @@ func TestDB_SubscribePull_sinceAndUntil(t *testing.T) {
errChan := make(chan error) errChan := make(chan error)
for bin := uint8(0); bin <= uint8(storage.MaxPO); bin++ { for bin := uint8(0); bin <= uint8(storage.MaxPO); bin++ {
var since *ChunkInfo var since *ChunkDescriptor
if c, ok := upload1[bin]; ok { if c, ok := upload1[bin]; ok {
since = &c since = &c
} }
@ -399,10 +399,10 @@ func uploadRandomChunksBin(t *testing.T, db *DB, uploader *Putter, addrs map[uin
} }
} }
// readPullSubscriptionBin is a helper function that reads all ChunkInfos from a channel and // readPullSubscriptionBin is a helper function that reads all ChunkDescriptors from a channel and
// sends error to errChan, even if it is nil, to count the number of ChunkInfos // sends error to errChan, even if it is nil, to count the number of ChunkDescriptors
// returned by the channel. // returned by the channel.
func readPullSubscriptionBin(ctx context.Context, bin uint8, ch <-chan ChunkInfo, addrs map[uint8][]storage.Address, errChan chan error) { func readPullSubscriptionBin(ctx context.Context, bin uint8, ch <-chan ChunkDescriptor, addrs map[uint8][]storage.Address, errChan chan error) {
var i int // address index var i int // address index
for { for {
select { select {