swarm/network/stream: fix syncer tests

swarm/network/stream/common_test.go

@@ -38,6 +38,10 @@ var (
 	loglevel = flag.Int("loglevel", 2, "verbosity of logs")
 )
 
+var (
+	waitPeerErrC chan error
+)
+
 var services = adapters.Services{
 	"streamer": NewStreamerService,
 }
@@ -49,6 +53,7 @@ func init() {
 	adapters.RegisterServices(services)
 
 	log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(os.Stderr, log.TerminalFormat(false))))
+
 }
 
 // newService
@@ -60,7 +65,14 @@ func NewStreamerService(ctx *adapters.ServiceContext) (node.Service, error) {
 	db := storage.NewDBAPI(store.(*storage.LocalStore))
 	delivery := NewDelivery(kad, db)
 	deliveries[id] = delivery
-	return NewRegistry(addr, delivery, store), nil
+	//netStore := storage.NewNetStore(store.(*storage.LocalStore), func(*storage.Chunk) error { return errors.New("not retrieved yet") })
+	r := NewRegistry(addr, delivery, store)
+	RegisterSwarmSyncerServer(r, db)
+	RegisterSwarmSyncerClient(r, db)
+	go func() {
+		waitPeerErrC <- waitForPeers(r, 1*time.Second, 1)
+	}()
+	return r, nil
 }
 
 func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *storage.LocalStore, func(), error) {
@@ -87,7 +99,7 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *stora
 	streamer := NewRegistry(addr, delivery, localStore)
 	protocolTester := p2ptest.NewProtocolTester(t, network.NewNodeIDFromAddr(addr), 1, streamer.runProtocol)
 
-	err = waitForPeers(streamer, 1*time.Second)
+	err = waitForPeers(streamer, 1*time.Second, 1)
 	if err != nil {
 		return nil, nil, nil, nil, errors.New("timeout: peer is not created")
 	}
@@ -95,13 +107,13 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *stora
 	return protocolTester, streamer, localStore, teardown, nil
 }
 
-func waitForPeers(streamer *Registry, timeout time.Duration) error {
+func waitForPeers(streamer *Registry, timeout time.Duration, expectedPeers int) error {
 	ticker := time.NewTicker(10 * time.Millisecond)
 	timeoutTimer := time.NewTimer(timeout)
 	for {
 		select {
 		case <-ticker.C:
-			if len(streamer.peers) > 0 {
+			if streamer.peersCount() >= expectedPeers {
 				return nil
 			}
 		case <-timeoutTimer.C:

swarm/network/stream/delivery_test.go

@@ -350,22 +350,49 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, size int, skipCheck bool)
 	if err != nil {
 		t.Fatal(err.Error())
 	}
-	// create a retriever dpa for the pivot node
+
-	delivery := deliveries[sim.IDs[0]]
+	waitPeerErrC = make(chan error)
-	dpacs := storage.NewNetStore(sim.Stores[0].(*storage.LocalStore), func(chunk *storage.Chunk) error { return delivery.RequestFromPeers(chunk.Key[:], skipCheck) })
+
-	dpa := storage.NewDPA(dpacs, storage.NewChunkerParams())
-	dpa.Start()
 	action := func(context.Context) error {
-		dpa := storage.NewDPA(sim.Stores[0], storage.NewChunkerParams())
+
+		i := 0
+		for err := range waitPeerErrC {
+			if err != nil {
+				return fmt.Errorf("error waiting for peers: %s", err)
+			}
+			i++
+			if i == nodes {
+				break
+			}
+		}
+
+		for i := 0; i < len(sim.IDs)-1; i++ {
+			id := sim.IDs[i]
+			node := sim.Net.GetNode(id)
+			if node == nil {
+				return fmt.Errorf("unknown node: %s", id)
+			}
+			client, err := node.Client()
+			if err != nil {
+				return fmt.Errorf("error getting node client: %s", err)
+			}
+			sid := sim.IDs[i+1]
+			if err := client.Call(nil, "stream_subscribeStream", sid, swarmChunkServerStreamName, nil, 0, 0, Top, false); err != nil {
+				return fmt.Errorf("error subscribing: %s", err)
+			}
+		}
+
+		// create a retriever dpa for the pivot node
+		delivery := deliveries[sim.IDs[0]]
+		dpacs := storage.NewNetStore(sim.Stores[0].(*storage.LocalStore), func(chunk *storage.Chunk) error { return delivery.RequestFromPeers(chunk.Key[:], skipCheck) })
+		dpa := storage.NewDPA(dpacs, storage.NewChunkerParams())
 		dpa.Start()
-		// defer dpa.Stop()
 
 		go func() {
 			defer dpa.Stop()
 			log.Debug(fmt.Sprintf("retrieve %v", fileHash))
 			// 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
-			time.Sleep(2 * time.Second)
 			n, err := mustReadAll(dpa, fileHash)
 			log.Debug(fmt.Sprintf("retrieved %v", fileHash), "read", n, "err", err)
 		}()
@@ -388,9 +415,9 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, size int, skipCheck bool)
 			return false, fmt.Errorf("error getting node client: %s", err)
 		}
 		var total int64
-		if err := client.Call(&total, "stream_readAll", fileHash); err != nil {
+		ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
-			return false, fmt.Errorf("error reading all: %s (read %v)", err, total)
+		defer cancel()
-		}
+		err = client.CallContext(ctx, &total, "stream_readAll", fileHash)
 		// total, err := mustReadAll(dpa, fileHash)
 		log.Debug(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) {

swarm/network/stream/stream.go

@@ -181,6 +181,13 @@ func (r *Registry) deletePeer(peer *Peer) {
 	r.peersMu.Unlock()
 }
 
+func (r *Registry) peersCount() (c int) {
+	r.peersMu.Lock()
+	c = len(r.peers)
+	r.peersMu.Unlock()
+	return
+}
+
 // Run protocol run function
 func (r *Registry) run(p *protocols.Peer) error {
 	sp := NewPeer(p, r)
@@ -373,9 +380,22 @@ func mustReadAll(dpa *storage.DPA, hash []byte) (int64, error) {
 }
 
 func (api *API) ReadAll(hash []byte) (int64, error) {
-	return mustReadAll(api.dpa, hash)
+	r := api.dpa.Retrieve(hash)
+	buf := make([]byte, 1024)
+	var n int
+	var total int64
+	var err error
+	for (total == 0 || n > 0) && err == nil {
+		n, err = r.ReadAt(buf, total)
+		total += int64(n)
+	}
+	if err != nil && err != io.EOF {
+		return total, err
+	}
+	return total, nil
+	//return mustReadAll(api.dpa, hash)
 }
 
-func (api *API) Subscribe(peerId discover.NodeID, s string, t []byte, from, to uint64, priority uint8, live bool) error {
+func (api *API) SubscribeStream(peerId discover.NodeID, s string, t []byte, from, to uint64, priority uint8, live bool) error {
 	return api.streamer.Subscribe(peerId, s, t, from, to, priority, live)
 }

swarm/network/stream/syncer_test.go

@@ -83,8 +83,21 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, size int, skipCheck bool,
 	if err != nil {
 		t.Fatal(err.Error())
 	}
+	waitPeerErrC = make(chan error)
 	// create a retriever dpa for the pivot node
 	action := func(context.Context) error {
+
+		i := 0
+		for err := range waitPeerErrC {
+			if err != nil {
+				return fmt.Errorf("error waiting for peers: %s", err)
+			}
+			i++
+			if i == nodes {
+				break
+			}
+		}
+
 		for i := 0; i < len(sim.IDs)-1; i++ {
 			id := sim.IDs[i]
 			// if err := streamer.Subscribe(p.ID(), "SYNC", []byte{uint8(1)}, 0, 0, Top, false); err != nil {
@@ -98,9 +111,8 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, size int, skipCheck bool,
 			if err != nil {
 				return fmt.Errorf("error getting node client: %s", err)
 			}
-			var n int64
 			sid := sim.IDs[i+1]
-			if err := client.Call(&n, "stream_subscribe", sid, "SYNC", []byte{uint8(1)}, 0, 0, Top, false); err != nil {
+			if err := client.Call(nil, "stream_subscribeStream", sid, "SYNC", []byte{uint8(1)}, 0, 0, Top, false); err != nil {
 				return fmt.Errorf("error subscribing: %s", err)
 			}
 		}