swarm/network/stream: add concurrent counter for tests

swarm/network/stream/streamer_test.go

@@ -1177,6 +1177,7 @@ starts the simulation, waits for SyncUpdateDelay in order to kick off
 stream registration, then tests that there are subscriptions.
 */
 func TestGetSubscriptionsRPC(t *testing.T) {
+
 	// arbitrarily set to 4
 	nodeCount := 4
 	// run with more nodes if `longrunning` flag is set
@@ -1188,19 +1189,16 @@ func TestGetSubscriptionsRPC(t *testing.T) {
 	// holds the msg code for SubscribeMsg
 	var subscribeMsgCode uint64
 	var ok bool
-	var expectedMsgCount = 0
+	var expectedMsgCount counter
 
 	// this channel signalizes that the expected amount of subscriptiosn is done
 	allSubscriptionsDone := make(chan struct{})
-	lock := sync.RWMutex{}
 	// after the test, we need to reset the subscriptionFunc to the default
 	defer func() { subscriptionFunc = doRequestSubscription }()
 
 	// we use this subscriptionFunc for this test: just increases count and calls the actual subscription
 	subscriptionFunc = func(r *Registry, p *network.Peer, bin uint8, subs map[enode.ID]map[Stream]struct{}) bool {
-		lock.Lock()
+		expectedMsgCount.inc()
-		expectedMsgCount++
-		lock.Unlock()
 		doRequestSubscription(r, p, bin, subs)
 		return true
 	}
@@ -1293,9 +1291,7 @@ func TestGetSubscriptionsRPC(t *testing.T) {
 			return errors.New("Context timed out")
 		}
 
-		lock.RLock()
+		log.Debug("Expected message count: ", "expectedMsgCount", expectedMsgCount.count())
-		log.Debug("Expected message count: ", "expectedMsgCount", expectedMsgCount)
-		lock.RUnlock()
 		//now iterate again, this time we call each node via RPC to get its subscriptions
 		realCount := 0
 		for _, node := range nodes {
@@ -1326,9 +1322,7 @@ func TestGetSubscriptionsRPC(t *testing.T) {
 			}
 		}
 		// every node is mutually subscribed to each other, so the actual count is half of it
-		lock.RLock()
+		emc := expectedMsgCount.count()
-		emc := expectedMsgCount
-		lock.RUnlock()
 		if realCount/2 != emc {
 			return fmt.Errorf("Real subscriptions and expected amount don't match; real: %d, expected: %d", realCount/2, emc)
 		}
@@ -1338,3 +1332,26 @@ func TestGetSubscriptionsRPC(t *testing.T) {
 		t.Fatal(result.Error)
 	}
 }
+
+// counter is used to concurrently increment
+// and read an integer value.
+type counter struct {
+	v  int
+	mu sync.RWMutex
+}
+
+// Increment the counter.
+func (c *counter) inc() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	c.v++
+}
+
+// Read the counter value.
+func (c *counter) count() int {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	return c.v
+}