swarm/network/stream: fix delivery_test bugs and refactor

swarm/network/stream/common_test.go

@@ -28,15 +28,14 @@ import (
 	"math/rand"
 	"os"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
-	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/log"
-	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/p2p/discover"
 	p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
-	"github.com/ethereum/go-ethereum/rpc"
+	//"github.com/ethereum/go-ethereum/rpc"
 	"github.com/ethereum/go-ethereum/swarm/network"
 	"github.com/ethereum/go-ethereum/swarm/network/simulation"
 	"github.com/ethereum/go-ethereum/swarm/pot"
@@ -67,6 +66,7 @@ var (
 
 func init() {
 	flag.Parse()
+	rand.Seed(time.Now().UnixNano())
 
 	log.PrintOrigins(true)
 	log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(colorable.NewColorableStderr(), log.TerminalFormat(true))))
@@ -142,20 +142,31 @@ func waitForPeers(streamer *Registry, timeout time.Duration, expectedPeers int)
 	}
 }
 
-type TestRegistry struct {
+type roundRobinStore struct {
-	*Registry
+	index  uint32
-	fileStore *storage.FileStore
+	stores []storage.ChunkStore
 }
 
-func (r *TestRegistry) APIs() []rpc.API {
+func newRoundRobinStore(stores ...storage.ChunkStore) *roundRobinStore {
-	a := r.Registry.APIs()
+	return &roundRobinStore{
-	a = append(a, rpc.API{
+		stores: stores,
-		Namespace: "stream",
+	}
-		Version:   "3.0",
+}
-		Service:   r,
+
-		Public:    true,
+func (rrs *roundRobinStore) Get(ctx context.Context, addr storage.Address) (*storage.Chunk, error) {
-	})
+	return nil, errors.New("get not well defined on round robin store")
-	return a
+}
+
+func (rrs *roundRobinStore) Put(ctx context.Context, chunk *storage.Chunk) {
+	i := atomic.AddUint32(&rrs.index, 1)
+	idx := int(i) % len(rrs.stores)
+	rrs.stores[idx].Put(ctx, chunk)
+}
+
+func (rrs *roundRobinStore) Close() {
+	for _, store := range rrs.stores {
+		store.Close()
+	}
 }
 
 func readAll(fileStore *storage.FileStore, hash []byte) (int64, error) {
@@ -174,34 +185,8 @@ func readAll(fileStore *storage.FileStore, hash []byte) (int64, error) {
 	return total, nil
 }
 
-func (r *TestRegistry) ReadAll(hash common.Hash) (int64, error) {
+//func getHashes(r *Registry, ctx context.Context, peerId discover.NodeID, s Stream) (*rpc.Subscription, error) {
-	return readAll(r.fileStore, hash[:])
+func getHashes(r *Registry, ctx context.Context, peerId discover.NodeID, s Stream) (chan []byte, error) {
-}
-
-func (r *TestRegistry) Start(server *p2p.Server) error {
-	return r.Registry.Start(server)
-}
-
-func (r *TestRegistry) Stop() error {
-	return r.Registry.Stop()
-}
-
-type TestExternalRegistry struct {
-	*Registry
-}
-
-func (r *TestExternalRegistry) APIs() []rpc.API {
-	a := r.Registry.APIs()
-	a = append(a, rpc.API{
-		Namespace: "stream",
-		Version:   "3.0",
-		Service:   r,
-		Public:    true,
-	})
-	return a
-}
-
-func (r *TestExternalRegistry) GetHashes(ctx context.Context, peerId discover.NodeID, s Stream) (*rpc.Subscription, error) {
 	peer := r.getPeer(peerId)
 
 	client, err := peer.getClient(ctx, s)
@@ -211,41 +196,10 @@ func (r *TestExternalRegistry) GetHashes(ctx context.Context, peerId discover.No
 
 	c := client.Client.(*testExternalClient)
 
-	notifier, supported := rpc.NotifierFromContext(ctx)
+	return c.hashes, nil
-	if !supported {
-		return nil, fmt.Errorf("Subscribe not supported")
 }
 
-	sub := notifier.CreateSubscription()
+func enableNotifications(r *Registry, peerId discover.NodeID, s Stream) error {
-
-	go func() {
-		// if we begin sending event immediately some events
-		// will probably be dropped since the subscription ID might not be send to
-		// the client.
-		// ref: rpc/subscription_test.go#L65
-		time.Sleep(1 * time.Second)
-		for {
-			select {
-			case h := <-c.hashes:
-				<-c.enableNotificationsC // wait for notification subscription to complete
-				if err := notifier.Notify(sub.ID, h); err != nil {
-					log.Warn(fmt.Sprintf("rpc sub notifier notify stream %s: %v", s, err))
-				}
-			case err := <-sub.Err():
-				if err != nil {
-					log.Warn(fmt.Sprintf("caught subscription error in stream %s: %v", s, err))
-				}
-			case <-notifier.Closed():
-				log.Trace(fmt.Sprintf("rpc sub notifier closed"))
-				return
-			}
-		}
-	}()
-
-	return sub, nil
-}
-
-func (r *TestExternalRegistry) EnableNotifications(peerId discover.NodeID, s Stream) error {
 	peer := r.getPeer(peerId)
 
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
@@ -302,7 +256,6 @@ type testExternalServer struct {
 	keyFunc   func(key []byte, index uint64)
 	sessionAt uint64
 	maxKeys   uint64
-	streamer  *TestExternalRegistry
 }
 
 func newTestExternalServer(t string, sessionAt, maxKeys uint64, keyFunc func(key []byte, index uint64)) *testExternalServer {
@@ -382,10 +335,6 @@ func uploadFilesToNodes(sim *simulation.Simulation) ([]storage.Address, []string
 	return rootAddrs, rfiles, nil
 }
 
-func init() {
-	rand.Seed(time.Now().UnixNano())
-}
-
 //generate a random file (string)
 func generateRandomFile() (string, error) {
 	//generate a random file size between minFileSize and maxFileSize

swarm/network/stream/delivery_test.go

@@ -27,7 +27,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/node"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
@@ -332,6 +331,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 			r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), &RegistryOptions{
 				SkipCheck: skipCheck,
 			})
+			bucket.Store(bucketKeyRegistry, r)
 
 			retrieveFunc := func(ctx context.Context, chunk *storage.Chunk) error {
 				return delivery.RequestFromPeers(ctx, chunk.Addr[:], skipCheck)
@@ -339,16 +339,15 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 			netStore := storage.NewNetStore(localStore, retrieveFunc)
 			fileStore := storage.NewFileStore(netStore, storage.NewFileStoreParams())
 			bucket.Store(bucketKeyFileStore, fileStore)
-			testRegistry := &TestRegistry{Registry: r, fileStore: fileStore}
 
-			return testRegistry, cleanup, nil
+			return r, cleanup, nil
 
 		},
 	})
 	defer sim.Close()
 
 	log.Info("Adding nodes to simulation")
-	_, err := sim.AddNodesAndConnectFull(nodes)
+	_, err := sim.AddNodesAndConnectChain(nodes)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -357,19 +356,34 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 	ctx := context.Background()
 	result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error {
 		nodeIDs := sim.UpNodeIDs()
-		// create a retriever FileStore for the pivot node
+		//determine the pivot node to be the first node of the simulation
-		log.Debug("Selecting pivot node")
+		sim.SetPivotNode(nodeIDs[0])
-		//select first node
+		//distribute chunks of a random file into Stores of nodes 1 to nodes
-		node := nodeIDs[0]
+		//we will do this by creating a file store with an underlying round-robin store:
-
+		//the file store will create a hash for the uploaded file, but every chunk will be
-		item, ok := sim.NodeItem(node, bucketKeyFileStore)
+		//distributed to different nodes via round-robin scheduling
-		if !ok {
+		log.Debug("Writing file to round-robin file store")
-			return fmt.Errorf("No filestore")
+		//to do this, we create an array for chunkstores (length minus one, the pivot node)
+		stores := make([]storage.ChunkStore, len(nodeIDs)-1)
+		//we then need to get all stores from the sim....
+		lStores := sim.NodesItems(bucketKeyStore)
+		i := 0
+		//...iterate the buckets...
+		for id, bucketVal := range lStores {
+			//...and remove the one which is the pivot node
+			if id == *sim.PivotNodeID() {
+				continue
 			}
-		fileStore := item.(*storage.FileStore)
+			//the other ones are added to the array...
+			stores[i] = bucketVal.(storage.ChunkStore)
+			i++
+		}
+		//...which then gets passed to the round-robin file store
+		roundRobinFileStore := storage.NewFileStore(newRoundRobinStore(stores...), storage.NewFileStoreParams())
+		//now we can actually upload a (random) file to the round-robin store
 		size := chunkCount * chunkSize
 		log.Debug("Storing data to file store")
-		fileHash, wait, err := fileStore.Store(ctx, io.LimitReader(crand.Reader, int64(size)), int64(size), false)
+		fileHash, wait, err := roundRobinFileStore.Store(ctx, io.LimitReader(crand.Reader, int64(size)), int64(size), false)
 		// wait until all chunks stored
 		if err != nil {
 			return err
@@ -379,22 +393,31 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 			return err
 		}
 
-		for j := 0; j < len(nodeIDs)-1; j++ {
+		//each of the nodes (except pivot node) subscribes to the stream of the next node
-			client, err := sim.Net.GetNode(nodeIDs[j]).Client()
+		for j, node := range nodeIDs[0 : nodes-1] {
+			sid := nodeIDs[j+1]
+			item, ok := sim.NodeItem(node, bucketKeyRegistry)
+			if !ok {
+				return fmt.Errorf("No registry")
+			}
+			registry := item.(*Registry)
+			err = registry.Subscribe(sid, NewStream(swarmChunkServerStreamName, "", false), NewRange(0, 0), Top)
 			if err != nil {
 				return err
 			}
-			ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
-			defer cancel()
-			sid := nodeIDs[j+1]
-			client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream(swarmChunkServerStreamName, "", false), NewRange(0, 0), Top)
 		}
 
+		//get the pivot node's filestore
+		item, ok := sim.NodeItem(*sim.PivotNodeID(), bucketKeyFileStore)
+		if !ok {
+			return fmt.Errorf("No filestore")
+		}
+		pivotFileStore := item.(*storage.FileStore)
 		log.Debug("Starting retrieval routine")
 		go func() {
 			// start the retrieval on the pivot node - this will spawn retrieve requests for missing chunks
 			// we must wait for the peer connections to have started before requesting
-			n, err := readAll(fileStore, fileHash)
+			n, err := readAll(pivotFileStore, fileHash)
 			log.Info(fmt.Sprintf("retrieved %v", fileHash), "read", n, "err", err)
 			if err != nil {
 				t.Fatalf("requesting chunks action error: %v", err)
@@ -424,26 +447,20 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 			}
 		}()
 
+		//finally check that the pivot node gets all chunks via the root hash
 		log.Debug("Check retrieval")
-		allSuccess := true
+		success := true
-		id := nodeIDs[0]
-		client, err := sim.Net.GetNode(id).Client()
-		if err != nil {
-			return err
-		}
 		var total int64
-		ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+		total, err = readAll(pivotFileStore, fileHash)
-		defer cancel()
-		err = client.CallContext(ctx, &total, "stream_readAll", common.BytesToHash(fileHash))
 		if err != nil {
 			return err
 		}
 		log.Info(fmt.Sprintf("check if %08x is available locally: number of bytes read %v/%v (error: %v)", fileHash, total, size, err))
 		if err != nil || total != int64(size) {
-			allSuccess = false
+			success = false
 		}
 
-		if !allSuccess {
+		if !success {
 			return fmt.Errorf("Test failed, chunks not available on all nodes")
 		}
 		log.Debug("Test terminated successfully")
@@ -510,18 +527,17 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip
 				return delivery.RequestFromPeers(ctx, chunk.Addr[:], skipCheck)
 			}
 			netStore := storage.NewNetStore(localStore, retrieveFunc)
-			bucket.Store(bucketKeyNetStore, netStore)
 			fileStore := storage.NewFileStore(netStore, storage.NewFileStoreParams())
-			testRegistry := &TestRegistry{Registry: r, fileStore: fileStore}
+			bucket.Store(bucketKeyFileStore, fileStore)
 
-			return testRegistry, cleanup, nil
+			return r, cleanup, nil
 
 		},
 	})
 	defer sim.Close()
 
 	log.Info("Initializing test config")
-	_, err := sim.AddNodesAndConnectFull(nodes)
+	_, err := sim.AddNodesAndConnectChain(nodes)
 	if err != nil {
 		b.Fatal(err)
 	}

swarm/network/stream/intervals_test.go

@@ -25,7 +25,6 @@ import (
 	"os"
 	"sync"
 	"testing"
-	"time"
 
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/node"
@@ -79,6 +78,7 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
 			r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), &RegistryOptions{
 				SkipCheck: skipCheck,
 			})
+			bucket.Store(bucketKeyRegistry, r)
 
 			r.RegisterClientFunc(externalStreamName, func(p *Peer, t string, live bool) (Client, error) {
 				return newTestExternalClient(db), nil
@@ -90,9 +90,7 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
 			fileStore := storage.NewFileStore(localStore, storage.NewFileStoreParams())
 			bucket.Store(bucketKeyFileStore, fileStore)
 
-			testRegistry := &TestExternalRegistry{r}
+			return r, cleanup, nil
-
-			return testRegistry, cleanup, nil
 
 		},
 	})
@@ -129,16 +127,13 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
 			t.Fatal(err)
 		}
 
-		client, err := sim.Net.GetNode(checker).Client()
+		item, ok = sim.NodeItem(checker, bucketKeyRegistry)
-		if err != nil {
+		if !ok {
-			log.Error("Get Client error: %v", "err", err)
+			return fmt.Errorf("No registry")
-			return err
 		}
-		ctx, cancel := context.WithTimeout(ctx, 100*time.Second)
+		registry := item.(*Registry)
-		defer cancel()
+		err = registry.Subscribe(storer, NewStream(externalStreamName, "", live), history, Top)
-		err = client.CallContext(ctx, nil, "stream_subscribeStream", storer, NewStream(externalStreamName, "", live), history, Top)
 		if err != nil {
-			log.Error("Stream subscribe error: %v", "err", err)
 			return err
 		}
 
@@ -180,17 +175,16 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
 			}()
 
 			// live stream
-			liveHashesChan := make(chan []byte)
+			var liveHashesChan chan []byte
-			liveSubscription, err := client.Subscribe(ctx, "stream", liveHashesChan, "getHashes", storer, NewStream(externalStreamName, "", true))
+			liveHashesChan, err = getHashes(registry, ctx, storer, NewStream(externalStreamName, "", true))
 			if err != nil {
 				log.Error("Subscription error: %v", "err", err)
 				return
 			}
-			defer liveSubscription.Unsubscribe()
 			i := externalStreamSessionAt
 
 			// we have subscribed, enable notifications
-			err = client.CallContext(ctx, nil, "stream_enableNotifications", storer, NewStream(externalStreamName, "", true))
+			err = enableNotifications(registry, storer, NewStream(externalStreamName, "", true))
 			if err != nil {
 				return
 			}
@@ -207,8 +201,8 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
 					if i > externalStreamMaxKeys {
 						return
 					}
-				case err = <-liveSubscription.Err():
+				//case err = <-liveSubscription.Err():
-					return
+				//	return
 				case <-ctx.Done():
 					return
 				}
@@ -227,12 +221,11 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
 			}()
 
 			// history stream
-			historyHashesChan := make(chan []byte)
+			var historyHashesChan chan []byte
-			historySubscription, err := client.Subscribe(ctx, "stream", historyHashesChan, "getHashes", storer, NewStream(externalStreamName, "", false))
+			historyHashesChan, err = getHashes(registry, ctx, storer, NewStream(externalStreamName, "", false))
 			if err != nil {
 				return
 			}
-			defer historySubscription.Unsubscribe()
 
 			var i uint64
 			historyTo := externalStreamMaxKeys
@@ -244,7 +237,7 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
 			}
 
 			// we have subscribed, enable notifications
-			err = client.CallContext(ctx, nil, "stream_enableNotifications", storer, NewStream(externalStreamName, "", false))
+			err = enableNotifications(registry, storer, NewStream(externalStreamName, "", false))
 			if err != nil {
 				return
 			}
@@ -261,8 +254,8 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
 					if i > historyTo {
 						return
 					}
-				case err = <-historySubscription.Err():
+				//case err = <-historySubscription.Err():
-					return
+				//	return
 				case <-ctx.Done():
 					return
 				}

swarm/network/stream/snapshot_retrieval_test.go

@@ -21,6 +21,7 @@ import (
 	"os"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/ethereum/go-ethereum/node"
 	"github.com/ethereum/go-ethereum/p2p/discover"
@@ -45,7 +46,10 @@ const (
 //Number of nodes can be provided via commandline too.
 func TestFileRetrieval(t *testing.T) {
 	if *nodes != 0 {
-		fileRetrievalTest(t, *nodes)
+		err := runFileRetrievalTest(*nodes)
+		if err != nil {
+			t.Fatal(err)
+		}
 	} else {
 		nodeCnt := []int{16}
 		//if the `longrunning` flag has been provided
@@ -54,7 +58,10 @@ func TestFileRetrieval(t *testing.T) {
 			nodeCnt = append(nodeCnt, 32, 64, 128)
 		}
 		for _, n := range nodeCnt {
-			fileRetrievalTest(t, n)
+			err := runFileRetrievalTest(n)
+			if err != nil {
+				t.Fatal(err)
+			}
 		}
 	}
 }
@@ -69,7 +76,10 @@ func TestRetrieval(t *testing.T) {
 	//if nodes/chunks have been provided via commandline,
 	//run the tests with these values
 	if *nodes != 0 && *chunks != 0 {
-		retrievalTest(t, *chunks, *nodes)
+		err := runRetrievalTest(*chunks, *nodes)
+		if err != nil {
+			t.Fatal(err)
+		}
 	} else {
 		var nodeCnt []int
 		var chnkCnt []int
@@ -85,25 +95,12 @@ func TestRetrieval(t *testing.T) {
 		}
 		for _, n := range nodeCnt {
 			for _, c := range chnkCnt {
-				retrievalTest(t, c, n)
+				err := runRetrievalTest(c, n)
-			}
-		}
-	}
-}
-
-//Run the file retrieval test
-func fileRetrievalTest(t *testing.T, nodeCount int) {
-	err := runFileRetrievalTest(nodeCount)
 				if err != nil {
 					t.Fatal(err)
 				}
 			}
-
+		}
-//Run the chunks retrieval test
-func retrievalTest(t *testing.T, chunkCount int, nodeCount int) {
-	err := runRetrievalTest(chunkCount, nodeCount)
-	if err != nil {
-		t.Fatal(err)
 	}
 }
 
@@ -142,22 +139,15 @@ func runFileRetrievalTest(nodeCount int) error {
 			})
 
 			fileStore := storage.NewFileStore(storage.NewNetStore(localStore, nil), storage.NewFileStoreParams())
-			bucketKeyFileStore = simulation.BucketKey("filestore")
 			bucket.Store(bucketKeyFileStore, fileStore)
-			testRegistry := &TestRegistry{Registry: r, fileStore: fileStore}
 
-			return testRegistry, cleanup, nil
+			return r, cleanup, nil
 
 		},
 	})
 	defer sim.Close()
 
 	log.Info("Initializing test config")
-	_, err := sim.AddNodesAndConnectFull(3)
-	if err != nil {
-		return err
-	}
-	ctx := context.Background()
 
 	conf := &synctestConfig{}
 	//map of discover ID to indexes of chunks expected at that ID
@@ -167,11 +157,14 @@ func runFileRetrievalTest(nodeCount int) error {
 	//array where the generated chunk hashes will be stored
 	conf.hashes = make([]storage.Address, 0)
 
-	err = sim.UploadSnapshot(fmt.Sprintf("testing/snapshot_%d.json", nodeCount))
+	err := sim.UploadSnapshot(fmt.Sprintf("testing/snapshot_%d.json", nodeCount))
 	if err != nil {
 		return err
 	}
 
+	ctx, cancelSimRun := context.WithTimeout(context.Background(), 1*time.Minute)
+	defer cancelSimRun()
+
 	result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error {
 		nodeIDs := sim.UpNodeIDs()
 		for _, n := range nodeIDs {
@@ -291,9 +284,8 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
 			fileStore := storage.NewFileStore(storage.NewNetStore(localStore, nil), storage.NewFileStoreParams())
 			bucketKeyFileStore = simulation.BucketKey("filestore")
 			bucket.Store(bucketKeyFileStore, fileStore)
-			testRegistry := &TestRegistry{Registry: r, fileStore: fileStore}
 
-			return testRegistry, cleanup, nil
+			return r, cleanup, nil
 
 		},
 	})

swarm/network/stream/snapshot_sync_test.go

@@ -62,12 +62,12 @@ type synctestConfig struct {
 //to the pivot node, and we check that nodes get the chunks
 //they are expected to store based on the syncing protocol.
 //Number of chunks and nodes can be provided via commandline too.
-func TestSyncingViaRegistry(t *testing.T) {
+func TestSyncingViaGlobalSync(t *testing.T) {
 	//if nodes/chunks have been provided via commandline,
 	//run the tests with these values
 	if *nodes != 0 && *chunks != 0 {
 		log.Info(fmt.Sprintf("Running test with %d chunks and %d nodes...", *chunks, *nodes))
-		testSyncing(t, *chunks, *nodes)
+		testSyncingViaGlobalSync(t, *chunks, *nodes)
 	} else {
 		var nodeCnt []int
 		var chnkCnt []int
@@ -84,18 +84,21 @@ func TestSyncingViaRegistry(t *testing.T) {
 		for _, chnk := range chnkCnt {
 			for _, n := range nodeCnt {
 				log.Info(fmt.Sprintf("Long running test with %d chunks and %d nodes...", chnk, n))
-				testSyncing(t, chnk, n)
+				testSyncingViaGlobalSync(t, chnk, n)
 			}
 		}
 	}
 }
 
-func TestSyncingViaRPC(t *testing.T) {
+func TestSyncingViaDirectSubscribe(t *testing.T) {
 	//if nodes/chunks have been provided via commandline,
 	//run the tests with these values
 	if *nodes != 0 && *chunks != 0 {
 		log.Info(fmt.Sprintf("Running test with %d chunks and %d nodes...", *chunks, *nodes))
-		testSyncingViaRPC(t, *chunks, *nodes)
+		err := testSyncingViaDirectSubscribe(*chunks, *nodes)
+		if err != nil {
+			t.Fatal(err)
+		}
 	} else {
 		var nodeCnt []int
 		var chnkCnt []int
@@ -112,13 +115,16 @@ func TestSyncingViaRPC(t *testing.T) {
 		for _, chnk := range chnkCnt {
 			for _, n := range nodeCnt {
 				log.Info(fmt.Sprintf("Long running test with %d chunks and %d nodes...", chnk, n))
-				testSyncingViaRPC(t, chnk, n)
+				err := testSyncingViaDirectSubscribe(chnk, n)
+				if err != nil {
+					t.Fatal(err)
+				}
 			}
 		}
 	}
 }
 
-func testSyncing(t *testing.T, chunkCount int, nodeCount int) {
+func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) {
 	sim := simulation.New(map[string]simulation.ServiceFunc{
 		"streamer": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
 
@@ -142,11 +148,9 @@ func testSyncing(t *testing.T, chunkCount int, nodeCount int) {
 				DoSync:          true,
 				SyncUpdateDelay: 3 * time.Second,
 			})
+			bucket.Store(bucketKeyRegistry, r)
 
-			fileStore := storage.NewFileStore(storage.NewNetStore(localStore, nil), storage.NewFileStoreParams())
+			return r, cleanup, nil
-			testRegistry := &TestRegistry{Registry: r, fileStore: fileStore}
-
-			return testRegistry, cleanup, nil
 
 		},
 	})
@@ -267,14 +271,6 @@ func testSyncing(t *testing.T, chunkCount int, nodeCount int) {
 	}
 }
 
-//Do run the tests
-func testSyncingViaRPC(t *testing.T, chunkCount int, nodeCount int) {
-	err := runSyncTest(chunkCount, nodeCount)
-	if err != nil {
-		t.Fatal(err)
-	}
-}
-
 /*
 The test generates the given number of chunks
 
@@ -287,7 +283,7 @@ assuming that the snapshot file identifies a healthy
 kademlia network. The snapshot should have 'streamer' in its service list.
 
 */
-func runSyncTest(chunkCount int, nodeCount int) error {
+func testSyncingViaDirectSubscribe(chunkCount int, nodeCount int) error {
 	sim := simulation.New(map[string]simulation.ServiceFunc{
 		"streamer": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
 
@@ -308,11 +304,12 @@ func runSyncTest(chunkCount int, nodeCount int) error {
 			delivery := NewDelivery(kad, db)
 
 			r := NewRegistry(addr, delivery, db, state.NewInmemoryStore(), nil)
+			bucket.Store(bucketKeyRegistry, r)
 
 			fileStore := storage.NewFileStore(storage.NewNetStore(localStore, nil), storage.NewFileStoreParams())
-			testRegistry := &TestRegistry{Registry: r, fileStore: fileStore}
+			bucket.Store(bucketKeyFileStore, fileStore)
 
-			return testRegistry, cleanup, nil
+			return r, cleanup, nil
 
 		},
 	})
@@ -353,15 +350,17 @@ func runSyncTest(chunkCount int, nodeCount int) error {
 		filter := simulation.NewPeerEventsFilter().Type(p2p.PeerEventTypeMsgRecv).Protocol("stream").MsgCode(4)
 		eventC := sim.PeerEvents(ctx, nodeIDs, filter)
 
-		for j, id := range nodeIDs {
+		for j, node := range nodeIDs {
 			log.Trace(fmt.Sprintf("Start syncing subscriptions: %d", j))
-			client, err := sim.Net.GetNode(id).Client()
-			if err != nil {
-				return err
-			}
 			//start syncing!
+			item, ok := sim.NodeItem(node, bucketKeyRegistry)
+			if !ok {
+				return fmt.Errorf("No registry")
+			}
+			registry := item.(*Registry)
+
 			var cnt int
-			err = client.CallContext(ctx, &cnt, "stream_startSyncing")
+			cnt, err = startSyncing(registry)
 			if err != nil {
 				return err
 			}
@@ -379,8 +378,7 @@ func runSyncTest(chunkCount int, nodeCount int) error {
 				break
 			}
 		}
-		//get the the node at that index
+		//select a random node for upload
-		//this is the node selected for upload
 		node := sim.RandomUpNode()
 		item, ok := sim.NodeItem(node.ID, bucketKeyStore)
 		if !ok {
@@ -463,7 +461,7 @@ func runSyncTest(chunkCount int, nodeCount int) error {
 //issues `RequestSubscriptionMsg` to peers, based on po, by iterating over
 //the kademlia's `EachBin` function.
 //returns the number of subscriptions requested
-func (r *TestRegistry) StartSyncing(ctx context.Context) (int, error) {
+func startSyncing(r *Registry) (int, error) {
 	var err error
 
 	kad, ok := r.delivery.overlay.(*network.Kademlia)

swarm/network/stream/syncer_test.go

@@ -114,10 +114,8 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 
 			fileStore := storage.NewFileStore(storage.NewNetStore(localStore, nil), storage.NewFileStoreParams())
 			bucket.Store(bucketKeyFileStore, fileStore)
-			testRegistry := &TestRegistry{Registry: r, fileStore: fileStore}
-			bucket.Store(bucketKeyRegistry, testRegistry)
 
-			return testRegistry, cleanup, nil
+			return r, cleanup, nil
 
 		},
 	})