swarm/network/stream: remove watchSim and its panic

swarm/network/stream/snapshot_sync_test.go

@@ -17,6 +17,7 @@ package stream
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io/ioutil"
 	"os"
@@ -181,22 +182,26 @@ func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) {
 		t.Fatal(err)
 	}
 
-	disconnected := watchDisconnections(ctx, sim)
-
 	result := runSim(conf, ctx, sim, chunkCount)
 
 	if result.Error != nil {
 		t.Fatal(result.Error)
 	}
-	if yes, ok := disconnected.Load().(bool); ok && yes {
-		t.Fatal("disconnect events received")
-	}
 	log.Info("Simulation ended")
 }
 
 func runSim(conf *synctestConfig, ctx context.Context, sim *simulation.Simulation, chunkCount int) simulation.Result {
 
-	return sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error {
+	return sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) (err error) {
+		disconnected := watchDisconnections(ctx, sim)
+		defer func() {
+			if err != nil {
+				if yes, ok := disconnected.Load().(bool); ok && yes {
+					err = errors.New("disconnect events received")
+				}
+			}
+		}()
+
 		nodeIDs := sim.UpNodeIDs()
 		for _, n := range nodeIDs {
 			//get the kademlia overlay address from this ID

swarm/network/stream/visualized_snapshot_sync_sim_test.go

@@ -66,40 +66,6 @@ func setupSim(serviceMap map[string]simulation.ServiceFunc) (int, int, *simulati
 	return nodeCount, chunkCount, sim
 }
 
-//watch for disconnections and wait for healthy
-func watchSim(sim *simulation.Simulation) (context.Context, context.CancelFunc) {
-	ctx, cancelSimRun := context.WithTimeout(context.Background(), 1*time.Minute)
-
-	if _, err := sim.WaitTillHealthy(ctx); err != nil {
-		panic(err)
-	}
-
-	disconnections := sim.PeerEvents(
-		context.Background(),
-		sim.NodeIDs(),
-		simulation.NewPeerEventsFilter().Drop(),
-	)
-
-	go func() {
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			case d := <-disconnections:
-				if d.Error != nil {
-					log.Error("peer drop event error", "node", d.NodeID, "peer", d.PeerID, "err", err)
-				} else {
-					log.Error("peer drop", "node", d.NodeID, "peer", d.PeerID)
-				}
-				panic("unexpected disconnect")
-				cancelSimRun()
-			}
-		}
-	}()
-
-	return ctx, cancelSimRun
-}
-
 //This test requests bogus hashes into the network
 func TestNonExistingHashesWithServer(t *testing.T) {
 
@@ -111,14 +77,20 @@ func TestNonExistingHashesWithServer(t *testing.T) {
 		panic(err)
 	}
 
-	ctx, cancelSimRun := watchSim(sim)
-	defer cancelSimRun()
-
 	//in order to get some meaningful visualization, it is beneficial
 	//to define a minimum duration of this test
 	testDuration := 20 * time.Second
 
-	result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error {
+	result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) (err error) {
+		disconnected := watchDisconnections(ctx, sim)
+		defer func() {
+			if err != nil {
+				if yes, ok := disconnected.Load().(bool); ok && yes {
+					err = errors.New("disconnect events received")
+				}
+			}
+		}()
+
 		//check on the node's FileStore (netstore)
 		id := sim.Net.GetRandomUpNode().ID()
 		item, ok := sim.NodeItem(id, bucketKeyFileStore)
@@ -222,9 +194,6 @@ func TestSnapshotSyncWithServer(t *testing.T) {
 		panic(err)
 	}
 
-	ctx, cancelSimRun := watchSim(sim)
-	defer cancelSimRun()
-
 	//run the sim
 	result := runSim(conf, ctx, sim, chunkCount)