diff --git a/swarm/storage/localstore/localstore.go b/swarm/storage/localstore/localstore.go index 5854fa9140..f04da404d0 100644 --- a/swarm/storage/localstore/localstore.go +++ b/swarm/storage/localstore/localstore.go @@ -255,7 +255,7 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) { return nil, err } // create a pull syncing feed used by SubscribePull function - db.pullFeed = newFeed(db.pullIndex, db.retrievalDataIndex) + db.pullFeed = newFeed(db.pullIndex) // push index contains as yet unsynced chunks db.pushIndex, err = db.shed.NewIndex("StoredTimestamp|Hash->nil", shed.IndexFuncs{ EncodeKey: func(fields shed.Item) (key []byte, err error) { @@ -280,7 +280,7 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) { return nil, err } // create a push syncing feed used by SubscribePush function - db.pushFeed = newFeed(db.pushIndex, db.retrievalDataIndex) + db.pushFeed = newFeed(db.pushIndex) // gc index for removable chunk ordered by ascending last access time db.gcIndex, err = db.shed.NewIndex("AccessTimestamp|StoredTimestamp|Hash->nil", shed.IndexFuncs{ EncodeKey: func(fields shed.Item) (key []byte, err error) { diff --git a/swarm/storage/localstore/subscriptions.go b/swarm/storage/localstore/subscriptions.go index a60260d989..3e21cced58 100644 --- a/swarm/storage/localstore/subscriptions.go +++ b/swarm/storage/localstore/subscriptions.go @@ -40,16 +40,16 @@ func (db *DB) SubscribePush(ctx context.Context) (s *Subscription, err error) { return db.pushFeed.subscribe(ctx, nil) } -// Subscription provides stream of Chunks in a particular order -// through the Chunks channel. That channel will not be closed -// when the last Chunk is read, but will block until the new Chunk +// Subscription provides stream of Addresses in a particular order +// through the Addrs channel. That channel will not be closed +// when the last Address is read, but will block until the new Address // is added to database index. Subscription should be used for -// getting Chunks and waiting for new ones. It provides methods +// getting Addresses and waiting for new ones. It provides methods // to control and get information about subscription state. type Subscription struct { - // Chunks is the read-only channel that provides stream of chunks. - // This is the subscription main purpose. - Chunks <-chan storage.Chunk + // Addrs is the read-only channel that provides stream of + // chunks addresses. This is the subscription main purpose. + Addrs <-chan storage.Address // subscribe to set of keys only with this prefix prefix []byte @@ -92,17 +92,15 @@ func (s *Subscription) Stop() { }) } -// feed is a collection of Chunks subscriptions of order given -// by sort index and Chunk data provided by data index. +// feed is a collection of Address subscriptions of order given +// by the sort index. // It provides methods to create, trigger and remove subscriptions. // It is the internal core component for push and pull // index subscriptions. type feed struct { - // index on which keys the order of Chunks will be + // index on which keys the order of Addresses will be // provided by subscriptions sortIndex shed.Index - // index that contains chunk data - dataIndex shed.Index // collection fo subscriptions on this feed subscriptions []*Subscription // protects subscriptions slice @@ -114,12 +112,10 @@ type feed struct { } // newFeed creates a new feed with from sort and data indexes. -// Sort index provides ordering of Chunks and data index -// provides Chunk data. -func newFeed(sortIndex, dataIndex shed.Index) (f *feed) { +// Sort index provides ordering of Addresses. +func newFeed(sortIndex shed.Index) (f *feed) { return &feed{ sortIndex: sortIndex, - dataIndex: dataIndex, subscriptions: make([]*Subscription, 0), closeChan: make(chan struct{}), } @@ -135,9 +131,9 @@ func (f *feed) subscribe(ctx context.Context, prefix []byte) (s *Subscription, e return nil, ErrSubscriptionFeedClosed default: } - chunks := make(chan storage.Chunk) + addrs := make(chan storage.Address) s = &Subscription{ - Chunks: chunks, + Addrs: addrs, prefix: prefix, stopChan: make(chan struct{}), doneChan: make(chan struct{}), @@ -169,14 +165,8 @@ func (f *feed) subscribe(ctx context.Context, prefix []byte) (s *Subscription, e // - subscription stop is called // - context is done err = f.sortIndex.Iterate(func(item shed.Item) (stop bool, err error) { - // get chunk data - dataItem, err := f.dataIndex.Get(item) - if err != nil { - return true, err - } - select { - case chunks <- storage.NewChunk(dataItem.Address, dataItem.Data): + case addrs <- storage.Address(item.Address): // set next iteration start item // when its chunk is successfully sent to channel startFrom = &item @@ -189,7 +179,7 @@ func (f *feed) subscribe(ctx context.Context, prefix []byte) (s *Subscription, e } }, &shed.IterateOptions{ StartFrom: startFrom, - // startFrom was sent as the last Chunk in the previous + // startFrom was sent as the last Address in the previous // iterator call, skip it in this one SkipStartFromItem: true, Prefix: prefix, @@ -245,7 +235,7 @@ func (f *feed) close() { }) } -// trigger signals all subscriptions with tprovided prefix +// trigger signals all subscriptions with provided prefix // that they should continue iterating over index keys // where they stopped in the last iteration. This method // should be called when new data is put to the index. diff --git a/swarm/storage/localstore/subscriptions_test.go b/swarm/storage/localstore/subscriptions_test.go index e421a69217..8ff817bdb5 100644 --- a/swarm/storage/localstore/subscriptions_test.go +++ b/swarm/storage/localstore/subscriptions_test.go @@ -28,7 +28,7 @@ import ( // TestSubscribePush uploads some chunks before and after // push syncing subscription is created and validates if -// all chunks are received in the right order. +// all addresses are received in the right order. func TestSubscribePush(t *testing.T) { t.Parallel() @@ -37,7 +37,7 @@ func TestSubscribePush(t *testing.T) { uploader := db.NewPutter(ModePutUpload) - chunks := make([]storage.Chunk, 0) + addrs := make([]storage.Address, 0) uploadRandomChunks := func(count int) { for i := 0; i < count; i++ { @@ -48,7 +48,7 @@ func TestSubscribePush(t *testing.T) { t.Fatal(err) } - chunks = append(chunks, chunk) + addrs = append(addrs, chunk.Address()) } } @@ -60,8 +60,8 @@ func TestSubscribePush(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() - // collect all errors from validating chunks, even nil ones - // to validate the number of chunks received by the subscription + // collect all errors from validating addresses, even nil ones + // to validate the number of addresses received by the subscription errChan := make(chan error) sub, err := db.SubscribePush(ctx) @@ -70,22 +70,19 @@ func TestSubscribePush(t *testing.T) { } defer sub.Stop() - // receive and validate chunks from the subscription + // receive and validate addresses from the subscription go func() { - var i int // chunk index + var i int // address index for { select { - case got := <-sub.Chunks: - want := chunks[i] + case got := <-sub.Addrs: + want := addrs[i] var err error - if !bytes.Equal(got.Data(), want.Data()) { - err = fmt.Errorf("got chunk %v data %x, want %x", i, got.Data(), want.Data()) - } - if !bytes.Equal(got.Address(), want.Address()) { - err = fmt.Errorf("got chunk %v address %s, want %s", i, got.Address().Hex(), want.Address().Hex()) + if !bytes.Equal(got, want) { + err = fmt.Errorf("got chunk %v address %s, want %s", i, got, want) } i++ - // send one and only one error per received chunk + // send one and only one error per received address errChan <- err case <-ctx.Done(): return @@ -101,7 +98,7 @@ func TestSubscribePush(t *testing.T) { // upload some chunks after some short time uploadRandomChunks(3) - totalChunks := len(chunks) + totalChunks := len(addrs) for i := 0; i < totalChunks; i++ { select { case err := <-errChan: @@ -116,7 +113,7 @@ func TestSubscribePush(t *testing.T) { // TestSubscribePush_multiple uploads chunks before and after // multiple push syncing subscriptions are created and -// validates if all chunks are received in the right order. +// validates if all addresses are received in the right order. func TestSubscribePush_multiple(t *testing.T) { t.Parallel() @@ -125,7 +122,7 @@ func TestSubscribePush_multiple(t *testing.T) { uploader := db.NewPutter(ModePutUpload) - chunks := make([]storage.Chunk, 0) + addrs := make([]storage.Address, 0) uploadRandomChunks := func(count int) { for i := 0; i < count; i++ { @@ -136,7 +133,7 @@ func TestSubscribePush_multiple(t *testing.T) { t.Fatal(err) } - chunks = append(chunks, chunk) + addrs = append(addrs, chunk.Address()) } } @@ -148,14 +145,14 @@ func TestSubscribePush_multiple(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() - // collect all errors from validating chunks, even nil ones - // to validate the number of chunks received by the subscription + // collect all errors from validating addresses, even nil ones + // to validate the number of addresses received by the subscription errChan := make(chan error) subsCount := 10 // start a number of subscriptions - // that all of them will write every chunk error to errChan + // that all of them will write every addresses error to errChan for j := 0; j < subsCount; j++ { sub, err := db.SubscribePush(ctx) if err != nil { @@ -163,22 +160,19 @@ func TestSubscribePush_multiple(t *testing.T) { } defer sub.Stop() - // receive and validate chunks from the subscription + // receive and validate addresses from the subscription go func(j int) { - var i int // chunk index + var i int // address index for { select { - case got := <-sub.Chunks: - want := chunks[i] + case got := <-sub.Addrs: + want := addrs[i] var err error - if !bytes.Equal(got.Data(), want.Data()) { - err = fmt.Errorf("got chunk %v on subscription %v data %x, want %x", i, j, got.Data(), want.Data()) - } - if !bytes.Equal(got.Address(), want.Address()) { - err = fmt.Errorf("got chunk %v on subscription %v address %s, want %s", i, j, got.Address().Hex(), want.Address().Hex()) + if !bytes.Equal(got, want) { + err = fmt.Errorf("got chunk %v address on subscription %v %s, want %s", i, j, got, want) } i++ - // send one and only one error per received chunk + // send one and only one error per received address errChan <- err case <-ctx.Done(): return @@ -195,8 +189,8 @@ func TestSubscribePush_multiple(t *testing.T) { // upload some chunks after some short time uploadRandomChunks(3) - // number of chunks received by all subscriptions - totalChunks := len(chunks) * subsCount + // number of addresses received by all subscriptions + totalChunks := len(addrs) * subsCount for i := 0; i < totalChunks; i++ { select { case err := <-errChan: @@ -211,7 +205,7 @@ func TestSubscribePush_multiple(t *testing.T) { // TestSubscribePull uploads some chunks before and after // pull syncing subscription is created and validates if -// all chunks are received in the right order +// all addresses are received in the right order // for expected proximity order bins. func TestSubscribePull(t *testing.T) { t.Parallel() @@ -221,7 +215,7 @@ func TestSubscribePull(t *testing.T) { uploader := db.NewPutter(ModePutUpload) - chunks := make(map[uint8][]storage.Chunk) + addrs := make(map[uint8][]storage.Address) var uploadedChunksCount int uploadRandomChunks := func(count int) { @@ -234,11 +228,11 @@ func TestSubscribePull(t *testing.T) { } bin := db.po(chunk.Address()) - if _, ok := chunks[bin]; !ok { - chunks[bin] = make([]storage.Chunk, 0) + if _, ok := addrs[bin]; !ok { + addrs[bin] = make([]storage.Address, 0) } - chunks[bin] = append(chunks[bin], chunk) + addrs[bin] = append(addrs[bin], chunk.Address()) uploadedChunksCount++ } } @@ -251,8 +245,8 @@ func TestSubscribePull(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() - // collect all errors from validating chunks, even nil ones - // to validate the number of chunks received by the subscription + // collect all errors from validating addresses, even nil ones + // to validate the number of addresses received by the subscription errChan := make(chan error) for bin := uint8(0); bin < uint8(storage.MaxPO); bin++ { @@ -262,22 +256,19 @@ func TestSubscribePull(t *testing.T) { } defer sub.Stop() - // receive and validate chunks from the subscription + // receive and validate addresses from the subscription go func(bin uint8) { - var i int // chunk index + var i int // address index for { select { - case got := <-sub.Chunks: - want := chunks[bin][i] + case got := <-sub.Addrs: + want := addrs[bin][i] var err error - if !bytes.Equal(got.Data(), want.Data()) { - err = fmt.Errorf("got chunk %v in bin %v data %x, want %x", i, bin, got.Data(), want.Data()) - } - if !bytes.Equal(got.Address(), want.Address()) { - err = fmt.Errorf("got chunk %v in bin %v address %s, want %s", i, bin, got.Address().Hex(), want.Address().Hex()) + if !bytes.Equal(got, want) { + err = fmt.Errorf("got chunk address %v in bin %v %s, want %s", i, bin, got, want) } i++ - // send one and only one error per received chunk + // send one and only one error per received address errChan <- err case <-ctx.Done(): return @@ -308,7 +299,7 @@ func TestSubscribePull(t *testing.T) { // TestSubscribePull_multiple uploads chunks before and after // multiple pull syncing subscriptions are created and -// validates if all chunks are received in the right order +// validates if all addresses are received in the right order // for expected proximity order bins. func TestSubscribePull_multiple(t *testing.T) { t.Parallel() @@ -318,7 +309,7 @@ func TestSubscribePull_multiple(t *testing.T) { uploader := db.NewPutter(ModePutUpload) - chunks := make(map[uint8][]storage.Chunk) + addrs := make(map[uint8][]storage.Address) var uploadedChunksCount int uploadRandomChunks := func(count int) { @@ -331,11 +322,11 @@ func TestSubscribePull_multiple(t *testing.T) { } bin := db.po(chunk.Address()) - if _, ok := chunks[bin]; !ok { - chunks[bin] = make([]storage.Chunk, 0) + if _, ok := addrs[bin]; !ok { + addrs[bin] = make([]storage.Address, 0) } - chunks[bin] = append(chunks[bin], chunk) + addrs[bin] = append(addrs[bin], chunk.Address()) uploadedChunksCount++ } } @@ -348,14 +339,14 @@ func TestSubscribePull_multiple(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() - // collect all errors from validating chunks, even nil ones - // to validate the number of chunks received by the subscription + // collect all errors from validating addresses, even nil ones + // to validate the number of addresses received by the subscription errChan := make(chan error) subsCount := 10 // start a number of subscriptions - // that all of them will write every chunk error to errChan + // that all of them will write every address error to errChan for j := 0; j < subsCount; j++ { for bin := uint8(0); bin < uint8(storage.MaxPO); bin++ { sub, err := db.SubscribePull(ctx, bin) @@ -364,22 +355,19 @@ func TestSubscribePull_multiple(t *testing.T) { } defer sub.Stop() - // receive and validate chunks from the subscription + // receive and validate addresses from the subscription go func(bin uint8, j int) { - var i int // chunk index + var i int // address index for { select { - case got := <-sub.Chunks: - want := chunks[bin][i] + case got := <-sub.Addrs: + want := addrs[bin][i] var err error - if !bytes.Equal(got.Data(), want.Data()) { - err = fmt.Errorf("got chunk %v in bin %v on subscription %v data %x, want %x", i, bin, j, got.Data(), want.Data()) - } - if !bytes.Equal(got.Address(), want.Address()) { - err = fmt.Errorf("got chunk %v in bin %v on subscription %v address %s, want %s", i, bin, j, got.Address().Hex(), want.Address().Hex()) + if !bytes.Equal(got, want) { + err = fmt.Errorf("got chunk address %v in bin %v on subscription %v %s, want %s", i, bin, j, got, want) } i++ - // send one and only one error per received chunk + // send one and only one error per received address errChan <- err case <-ctx.Done(): return