swarm/network/stream: implement TestIntervals test

swarm/network/stream/common_test.go

@@ -18,6 +18,7 @@ package stream
 
 import (
 	"context"
+	"encoding/binary"
 	"errors"
 	"flag"
 	"fmt"
@@ -200,7 +201,6 @@ func (r *TestRegistry) ReadAll(hash common.Hash) (int64, error) {
 
 type TestExternalRegistry struct {
 	*Registry
-	hashesChan chan []byte
 }
 
 func (r *TestExternalRegistry) APIs() []rpc.API {
@@ -215,10 +215,9 @@ func (r *TestExternalRegistry) APIs() []rpc.API {
 }
 
 func (r *TestExternalRegistry) GetHashes(ctx context.Context, peerId discover.NodeID, s Stream) (*rpc.Subscription, error) {
-
 	peer := r.getPeer(peerId)
 
-	client, err := peer.getClient(s)
+	client, err := peer.getClient(ctx, s)
 	if err != nil {
 		return nil, err
 	}
@@ -236,13 +235,17 @@ func (r *TestExternalRegistry) GetHashes(ctx context.Context, peerId discover.No
 		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.Warn(fmt.Sprintf("rpc sub notifier closed"))
+				log.Trace(fmt.Sprintf("rpc sub notifier closed"))
+				return
 			}
 		}
 	}()
@@ -250,6 +253,22 @@ func (r *TestExternalRegistry) GetHashes(ctx context.Context, peerId discover.No
 	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)
+	defer cancel()
+
+	client, err := peer.getClient(ctx, s)
+	if err != nil {
+		return err
+	}
+
+	close(client.Client.(*testExternalClient).enableNotificationsC)
+
+	return nil
+}
+
 // TODO: merge functionalities of testExternalClient and testExternalServer
 // with testClient and testServer.
 
@@ -258,46 +277,83 @@ type testExternalClient struct {
 	// wait0     chan bool
 	// batchDone chan bool
 	hashes               chan []byte
+	db                   *storage.DBAPI
+	enableNotificationsC chan struct{}
 }
 
-func newTestExternalClient(t []byte, hashesChan chan []byte) *testExternalClient {
+func newTestExternalClient(t []byte, db *storage.DBAPI) *testExternalClient {
 	return &testExternalClient{
 		t: t,
 		// wait0:     make(chan bool),
 		// batchDone: make(chan bool),
-		hashes: hashesChan,
+		hashes:               make(chan []byte),
+		db:                   db,
+		enableNotificationsC: make(chan struct{}),
 	}
 }
 
-func (self *testExternalClient) NeedData(hash []byte) func() {
+func (c *testExternalClient) NeedData(hash []byte) func() {
-	self.hashes <- hash
+	chunk, _ := c.db.GetOrCreateRequest(hash)
-	return func() {}
+	if chunk.ReqC == nil {
+		return nil
+	}
+	c.hashes <- hash
+	return func() {
+		chunk.WaitToStore()
+	}
 }
 
-func (self *testExternalClient) BatchDone(Stream, uint64, []byte, []byte) func() (*TakeoverProof, error) {
+func (c *testExternalClient) BatchDone(Stream, uint64, []byte, []byte) func() (*TakeoverProof, error) {
-	// close(self.batchDone)
 	return nil
 }
 
-func (self *testExternalClient) Close() {}
+func (c *testExternalClient) Close() {}
+
+const testExternalServerBatchSize = 10
 
 type testExternalServer struct {
 	t         []byte
+	keyFunc   func(key []byte, index uint64)
+	sessionAt uint64
+	maxKeys   uint64
+	streamer  *TestExternalRegistry
 }
 
-func newTestExternalServer(t []byte) *testExternalServer {
+func newTestExternalServer(t []byte, sessionAt, maxKeys uint64, keyFunc func(key []byte, index uint64)) *testExternalServer {
+	if keyFunc == nil {
+		keyFunc = binary.BigEndian.PutUint64
+	}
 	return &testExternalServer{
 		t:         t,
+		keyFunc:   keyFunc,
+		sessionAt: sessionAt,
+		maxKeys:   maxKeys,
 	}
 }
 
-func (self *testExternalServer) SetNextBatch(from uint64, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) {
+func (s *testExternalServer) SetNextBatch(from uint64, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) {
-	return make([]byte, HashSize), from + 1, to + 1, nil, nil
+	if from == 0 && to == 0 {
+		from = s.sessionAt
+		to = s.sessionAt + testExternalServerBatchSize
+	}
+	if to-from > testExternalServerBatchSize {
+		to = from + testExternalServerBatchSize - 1
+	}
+	if from >= s.maxKeys && to > s.maxKeys {
+		return nil, 0, 0, nil, io.EOF
+	}
+	if to > s.maxKeys {
+		to = s.maxKeys
+	}
+	b := make([]byte, HashSize*(to-from+1))
+	for i := from; i <= to; i++ {
+		s.keyFunc(b[(i-from)*HashSize:(i-from+1)*HashSize], i)
+	}
+	return b, from, to, nil, nil
 }
 
-func (self *testExternalServer) GetData([]byte) ([]byte, error) {
+func (s *testExternalServer) GetData([]byte) ([]byte, error) {
-	return nil, nil
+	return make([]byte, 4096), nil
 }
 
-func (self *testExternalServer) Close() {
+func (s *testExternalServer) Close() {}
-}

swarm/network/stream/intervals_test.go

@@ -36,7 +36,11 @@ import (
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
-var externalStreamName = "externalStream"
+var (
+	externalStreamName             = "externalStream"
+	externalStreamSessionAt uint64 = 50
+	externalStreamMaxKeys   uint64 = 100
+)
 
 func newIntervalsStreamerService(ctx *adapters.ServiceContext) (node.Service, error) {
 	id := ctx.Config.ID
@@ -47,23 +51,28 @@ func newIntervalsStreamerService(ctx *adapters.ServiceContext) (node.Service, er
 	delivery := NewDelivery(kad, db)
 	deliveries[id] = delivery
 	netStore := storage.NewNetStore(store, nil)
-	hashesChan := make(chan []byte) // this chanel is only for one client, in need for more clients, create a map
 	r := NewRegistry(addr, delivery, netStore, intervals.NewMemStore(), defaultSkipCheck)
 
 	r.RegisterClientFunc(externalStreamName, func(p *Peer, t []byte, live bool) (Client, error) {
-		return newTestExternalClient(t, hashesChan), nil
+		return newTestExternalClient(t, db), nil
 	})
 	r.RegisterServerFunc(externalStreamName, func(p *Peer, t []byte, live bool) (Server, error) {
-		return newTestExternalServer(t), nil
+		return newTestExternalServer(t, externalStreamSessionAt, externalStreamMaxKeys, nil), nil
 	})
 
 	go func() {
 		waitPeerErrC <- waitForPeers(r, 1*time.Second, peerCount(id))
 	}()
-	return &TestExternalRegistry{r, hashesChan}, nil
+	return &TestExternalRegistry{r}, nil
 }
 
-func XTestIntervals(t *testing.T) {
+func TestIntervals(t *testing.T) {
+	testIntervals(t, true, nil)
+	testIntervals(t, false, &Range{From: 9, To: 26})
+	testIntervals(t, true, &Range{From: 9, To: 26})
+}
+
+func testIntervals(t *testing.T, live bool, history *Range) {
 	nodes := 2
 	chunkCount := dataChunkCount
 	skipCheck := false
@@ -76,6 +85,7 @@ func XTestIntervals(t *testing.T) {
 		ConnLevel:      1,
 		ToAddr:         toAddr,
 		Services:       services,
+		DefaultService: "intervalsStreamer",
 	}
 
 	sim, teardown, err := streamTesting.NewSimulation(conf)
@@ -84,6 +94,12 @@ func XTestIntervals(t *testing.T) {
 		t.Fatal(err)
 	}
 
+	stores = make(map[discover.NodeID]storage.ChunkStore)
+	deliveries = make(map[discover.NodeID]*Delivery)
+	for i, id := range sim.IDs {
+		stores[id] = sim.Stores[i]
+	}
+
 	peerCount = func(id discover.NodeID) int {
 		return 1
 	}
@@ -101,6 +117,7 @@ func XTestIntervals(t *testing.T) {
 	errc := make(chan error, 1)
 	waitPeerErrC = make(chan error)
 	quitC := make(chan struct{})
+	defer close(quitC)
 
 	action := func(ctx context.Context) error {
 		i := 0
@@ -116,42 +133,148 @@ func XTestIntervals(t *testing.T) {
 
 		liveHashesChan := make(chan []byte)
 		historyHashesChan := make(chan []byte)
+
+		var historySubscription *rpc.ClientSubscription
+		var liveSubscription *rpc.ClientSubscription
+
 		id := sim.IDs[1]
 		err := sim.CallClient(id, func(client *rpc.Client) error {
 			err := streamTesting.WatchDisconnections(id, client, errc, quitC)
 			if err != nil {
 				return err
 			}
-			ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
+			ctx, cancel := context.WithTimeout(ctx, 100*time.Second)
 			defer cancel()
 			sid := sim.IDs[0]
-			err = client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream(externalStreamName, nil, true), &Range{From: 0, To: 5}, Top)
 
+			err = client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream(externalStreamName, nil, live), history, Top)
 			if err != nil {
 				return err
 			}
+
+			liveSubErrC := make(chan error)
+			historySubErrC := make(chan error)
+
+			go func() {
+				if live {
+					var err error
+					defer func() { liveSubErrC <- err }()
 					// live stream
-			_, err = client.Subscribe(ctx, "stream_getHashes", liveHashesChan, sid, NewStream(externalStreamName, nil, true))
+					liveSubscription, err = client.Subscribe(ctx, "stream", liveHashesChan, "getHashes", sid, NewStream(externalStreamName, nil, true))
 					if err != nil {
+						return
+					}
+					// we have got the channel, enable notifications
+					err = client.CallContext(ctx, nil, "stream_enableNotifications", sid, NewStream(externalStreamName, nil, true))
+				} else {
+					close(liveSubErrC)
+				}
+			}()
+
+			go func() {
+				if !live || history != nil {
+					var err error
+					defer func() { historySubErrC <- err }()
+
+					// history stream
+					historySubscription, err = client.Subscribe(ctx, "stream", historyHashesChan, "getHashes", sid, NewStream(externalStreamName, nil, false))
+					if err != nil {
+						return
+					}
+					// we have got the channel, enable notifications
+					err = client.CallContext(ctx, nil, "stream_enableNotifications", sid, NewStream(externalStreamName, nil, false))
+				} else {
+					close(historySubErrC)
+				}
+			}()
+
+			if err := <-liveSubErrC; err != nil {
 				return err
 			}
-			// history stream
+			if err := <-historySubErrC; err != nil {
-			_, err = client.Subscribe(ctx, "stream_getHashes", historyHashesChan, sid, NewStream(externalStreamName, nil, false))
 				return err
+			}
+
+			return nil
 		})
 		if err != nil {
 			return err
 		}
 
+		historyErrC := make(chan error)
+
 		go func() {
-			for i := uint64(0); i < 5; i++ {
+			defer close(historyErrC)
-				h := binary.BigEndian.Uint64(<-historyHashesChan)
+
+			if historySubscription == nil {
+				return
+			}
+
+			defer historySubscription.Unsubscribe()
+
+			i := history.From
+			historyTo := externalStreamMaxKeys
+			if history != nil && history.To != 0 {
+				historyTo = history.To
+			}
+
+			for {
+				select {
+				case hash := <-historyHashesChan:
+					h := binary.BigEndian.Uint64(hash)
 					if h != i {
-					errc <- fmt.Errorf("")
+						historyErrC <- fmt.Errorf("expected history hash %d, got %d", i, h)
+						return
+					}
+					i++
+					if i > historyTo {
+						return
+					}
+				case err := <-historySubscription.Err():
+					historyErrC <- err
+				case <-ctx.Done():
+					return
 				}
 			}
 		}()
-		return nil
+
+		liveErrC := make(chan error)
+
+		go func() {
+			defer close(liveErrC)
+
+			if liveSubscription == nil {
+				return
+			}
+
+			defer liveSubscription.Unsubscribe()
+
+			i := externalStreamSessionAt
+
+			for {
+				select {
+				case hash := <-liveHashesChan:
+					h := binary.BigEndian.Uint64(hash)
+					if h != i {
+						liveErrC <- fmt.Errorf("expected live hash %d, got %d", i, h)
+						return
+					}
+					i++
+					if i > externalStreamMaxKeys {
+						return
+					}
+				case err := <-liveSubscription.Err():
+					errc <- err
+				case <-ctx.Done():
+					return
+				}
+			}
+		}()
+
+		if err = <-historyErrC; err != nil {
+			return err
+		}
+		return <-liveErrC
 	}
 	check := func(ctx context.Context, id discover.NodeID) (bool, error) {
 		select {
@@ -168,7 +291,7 @@ func XTestIntervals(t *testing.T) {
 		Action:  action,
 		Trigger: streamTesting.Trigger(10*time.Millisecond, quitC, sim.IDs[0]),
 		Expect: &simulations.Expectation{
-			Nodes: sim.IDs[0:1],
+			Nodes: sim.IDs[1:1],
 			Check: check,
 		},
 	}

swarm/network/stream/messages.go

@@ -102,7 +102,7 @@ func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) (err error) {
 	}()
 
 	if req.Stream.Live && req.History != nil {
-		// subscribe to the history stream as well
+		// subscribe to the history stream
 		s, err := f(p, req.Stream.Key, false)
 		if err != nil {
 			return err
@@ -201,7 +201,7 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
 	if c.stream.Live {
 		c.sessionAt = req.From
 	}
-	from, to := c.nextBatch(req.To)
+	from, to := c.nextBatch(req.To + 1)
 	log.Trace("received offered batch", "peer", p.ID(), "stream", req.Stream, "from", req.From, "to", req.To)
 	if from == to {
 		return nil

swarm/network/stream/peer.go

@@ -18,7 +18,6 @@ package stream
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"sync"
 	"time"
@@ -32,11 +31,18 @@ import (
 
 var sendTimeout = 5 * time.Second
 
-var (
+type notFoundError struct {
-	errServerNotFound       = errors.New("server not found")
+	t string
-	errClientNotFound       = errors.New("client not found")
+	s Stream
-	errClientParamsNotFound = errors.New("client params not found")
+}
-)
+
+func newNotFoundError(t string, s Stream) *notFoundError {
+	return &notFoundError{t: t, s: s}
+}
+
+func (e *notFoundError) Error() string {
+	return fmt.Sprintf("%s not found for stream %q", e.t, e.s)
+}
 
 // Peer is the Peer extension for the streaming protocol
 type Peer struct {
@@ -44,8 +50,7 @@ type Peer struct {
 	streamer *Registry
 	pq       *pq.PriorityQueue
 	serverMu sync.RWMutex
-	clientMu       sync.RWMutex
+	clientMu sync.RWMutex // protects both clients and clientParams
-	clientParamsMu sync.RWMutex
 	servers  map[string]*server
 	clients  map[string]*client
 	// clientParams map keeps required client arguments
@@ -153,51 +158,71 @@ func (p *Peer) removeServer(s Stream) error {
 	sk := s.String()
 	server, ok := p.servers[sk]
 	if !ok {
-		return errServerNotFound
+		return newNotFoundError("server", s)
 	}
 	server.Close()
 	delete(p.servers, sk)
 	return nil
 }
 
-func (p *Peer) getClient(s Stream) (*client, error) {
+func (p *Peer) getClient(ctx context.Context, s Stream) (c *client, err error) {
+	var params *clientParams
+	sk := s.String()
+	func() {
 		p.clientMu.RLock()
 		defer p.clientMu.RUnlock()
 
-	client := p.clients[s.String()]
+		c = p.clients[sk]
-	if client == nil {
+		if c != nil {
-		return nil, fmt.Errorf("client '%v' not provided to peer %v", s, p.ID())
+			return
 		}
-	return client, nil
+		params = p.clientParams[sk]
+	}()
+	if c != nil {
+		return c, nil
+	}
+
+	if params != nil {
+		//debug.PrintStack()
+		if err := params.waitClient(ctx); err != nil {
+			return nil, err
+		}
+	}
+
+	p.clientMu.RLock()
+	defer p.clientMu.RUnlock()
+
+	c = p.clients[sk]
+	if c != nil {
+		return c, nil
+	}
+	return nil, newNotFoundError("client", s)
 }
 
-func (p *Peer) setClient(s Stream, from, to uint64) error {
+func (p *Peer) getOrSetClient(s Stream, from, to uint64) (c *client, created bool, err error) {
+	sk := s.String()
+
 	p.clientMu.Lock()
 	defer p.clientMu.Unlock()
 
-	sk := s.String()
+	c = p.clients[sk]
-	if p.clients[sk] != nil {
+	if c != nil {
-		return fmt.Errorf("client %v already registered", sk)
+		return c, false, nil
 	}
 
-	_, err := p.setClientNolock(s, from, to)
-	return err
-}
-
-func (p *Peer) setClientNolock(s Stream, from, to uint64) (c *client, err error) {
 	f, err := p.streamer.GetClientFunc(s.Name)
 	if err != nil {
-		return nil, err
+		return nil, false, err
 	}
 
 	is, err := f(p, s.Key, s.Live)
 	if err != nil {
-		return nil, err
+		return nil, false, err
 	}
 
 	cp, err := p.getClientParams(s)
 	if err != nil {
-		return nil, err
+		return nil, false, err
 	}
 	defer func() {
 		if err == nil {
@@ -232,7 +257,7 @@ func (p *Peer) setClientNolock(s Stream, from, to uint64) (c *client, err error)
 	}
 
 	if err := p.streamer.intervalsStore.Put(intervalsKey, intervals.NewIntervals(from)); err != nil {
-		return nil, err
+		return nil, false, err
 	}
 
 	next := make(chan error, 1)
@@ -240,29 +265,14 @@ func (p *Peer) setClientNolock(s Stream, from, to uint64) (c *client, err error)
 		Client:         is,
 		stream:         s,
 		priority:       cp.priority,
-		to:             to,
+		to:             cp.to,
 		next:           next,
 		intervalsStore: p.streamer.intervalsStore,
 		intervalsKey:   intervalsKey,
 	}
-	p.clients[s.String()] = c
+	p.clients[sk] = c
+	cp.clientCreated() // unblock all possible getClient calls that are waiting
 	next <- nil        // this is to allow wantedKeysMsg before first batch arrives
-	return c, nil
-}
-
-func (p *Peer) getOrSetClient(s Stream, from, to uint64) (c *client, created bool, err error) {
-	p.clientMu.RLock()
-	defer p.clientMu.RUnlock()
-
-	c = p.clients[s.String()]
-	if c != nil {
-		return c, false, nil
-	}
-
-	c, err = p.setClientNolock(s, from, to)
-	if err != nil {
-		return nil, false, err
-	}
 	return c, true, nil
 }
 
@@ -272,28 +282,20 @@ func (p *Peer) removeClient(s Stream) error {
 
 	client, ok := p.clients[s.String()]
 	if !ok {
-		return errClientNotFound
+		return newNotFoundError("client", s)
 	}
 	client.close()
 	return nil
 }
 
-func (p *Peer) getClientParams(s Stream) (*clientParams, error) {
-	p.clientParamsMu.RLock()
-	defer p.clientParamsMu.RUnlock()
-
-	params := p.clientParams[s.String()]
-	if params == nil {
-		return nil, fmt.Errorf("client params '%v' not provided to peer %v", s, p.ID())
-	}
-	return params, nil
-}
-
 func (p *Peer) setClientParams(s Stream, params *clientParams) error {
-	p.clientParamsMu.Lock()
+	p.clientMu.Lock()
-	defer p.clientParamsMu.Unlock()
+	defer p.clientMu.Unlock()
 
 	sk := s.String()
+	if p.clients[sk] != nil {
+		return fmt.Errorf("client %v already exists", sk)
+	}
 	if p.clientParams[sk] != nil {
 		return fmt.Errorf("client params %v already set", sk)
 	}
@@ -301,14 +303,19 @@ func (p *Peer) setClientParams(s Stream, params *clientParams) error {
 	return nil
 }
 
-func (p *Peer) removeClientParams(s Stream) error {
+func (p *Peer) getClientParams(s Stream) (*clientParams, error) {
-	p.clientParamsMu.Lock()
+	params := p.clientParams[s.String()]
-	defer p.clientParamsMu.Unlock()
+	if params == nil {
+		return nil, fmt.Errorf("client params '%v' not provided to peer %v", s, p.ID())
+	}
+	return params, nil
+}
 
+func (p *Peer) removeClientParams(s Stream) error {
 	sk := s.String()
 	_, ok := p.clientParams[sk]
 	if !ok {
-		return errClientParamsNotFound
+		return newNotFoundError("client params", s)
 	}
 	delete(p.clientParams, sk)
 	return nil

swarm/network/stream/stream.go

@@ -17,6 +17,7 @@
 package stream
 
 import (
+	"context"
 	"fmt"
 	"math"
 	"sync"
@@ -133,7 +134,12 @@ func (r *Registry) Subscribe(peerId discover.NodeID, s Stream, h *Range, priorit
 		return fmt.Errorf("peer not found %v", peerId)
 	}
 
-	err := peer.setClientParams(s, &clientParams{priority: priority})
+	var to uint64
+	if !s.Live && h != nil {
+		to = h.To
+	}
+
+	err := peer.setClientParams(s, newClientParams(priority, to))
 	if err != nil {
 		return err
 	}
@@ -141,9 +147,7 @@ func (r *Registry) Subscribe(peerId discover.NodeID, s Stream, h *Range, priorit
 	if s.Live && h != nil {
 		if err := peer.setClientParams(
 			getHistoryStream(s),
-			&clientParams{
+			newClientParams(getHistoryPriority(priority), h.To),
-				priority: getHistoryPriority(priority),
-			},
 		); err != nil {
 			return err
 		}
@@ -326,9 +330,15 @@ type Client interface {
 }
 
 func (c *client) nextBatch(from uint64) (nextFrom uint64, nextTo uint64) {
+	if c.to > 0 && from >= c.to {
+		return 0, 0
+	}
 	if c.stream.Live {
 		return from, 0
 	} else if from >= c.sessionAt {
+		if c.to > 0 {
+			return from, c.to
+		}
 		return from, math.MaxUint64
 	}
 	nextFrom, nextTo, err := c.NextInterval()
@@ -336,6 +346,9 @@ func (c *client) nextBatch(from uint64) (nextFrom uint64, nextTo uint64) {
 		log.Error("next intervals", "stream", c.stream)
 		return
 	}
+	if nextTo > c.to {
+		nextTo = c.to
+	}
 	if nextTo == 0 {
 		nextTo = c.sessionAt
 	}
@@ -372,6 +385,30 @@ func (c *client) close() {
 // between a subscription and initial offered hashes request handling.
 type clientParams struct {
 	priority uint8
+	to       uint64
+	// signal when the client is created
+	clientCreatedC chan struct{}
+}
+
+func newClientParams(priority uint8, to uint64) *clientParams {
+	return &clientParams{
+		priority:       priority,
+		to:             to,
+		clientCreatedC: make(chan struct{}),
+	}
+}
+
+func (c *clientParams) waitClient(ctx context.Context) error {
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-c.clientCreatedC:
+		return nil
+	}
+}
+
+func (c *clientParams) clientCreated() {
+	close(c.clientCreatedC)
 }
 
 // Spec is the spec of the streamer protocol

swarm/network/stream/streamer_test.go

@@ -168,7 +168,7 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 					Msg: &WantedHashesMsg{
 						Stream: stream,
 						Want:   []byte{5},
-						From:   8,
+						From:   9,
 						To:     0,
 					},
 					Peer: peerID,
@@ -386,75 +386,73 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) {
 	}
 }
 
-// TODO: fix: tests with TestExchanges are inconsistent because Expects check
+func TestStreamerUpstreamSubscribeLiveAndHistory(t *testing.T) {
-// 			  ordering is not guarrantied but fails if the order is wrong.
+	tester, streamer, _, teardown, err := newStreamerTester(t)
-// func TestStreamerUpstreamSubscribeLiveAndHistory(t *testing.T) {
+	defer teardown()
-// 	tester, streamer, _, teardown, err := newStreamerTester(t)
+	if err != nil {
-// 	defer teardown()
+		t.Fatal(err)
-// 	if err != nil {
+	}
-// 		t.Fatal(err)
-// 	}
 
-// 	stream := NewStream("foo", nil, true)
+	stream := NewStream("foo", nil, true)
 
-// 	streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) {
+	streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) {
-// 		return &testServer{
+		return &testServer{
-// 			t: t,
+			t: t,
-// 		}, nil
+		}, nil
-// 	})
+	})
 
-// 	peerID := tester.IDs[0]
+	peerID := tester.IDs[0]
 
-// 	err = tester.TestExchanges(p2ptest.Exchange{
+	err = tester.TestExchanges(p2ptest.Exchange{
-// 		Label: "Subscribe message",
+		Label: "Subscribe message",
-// 		Triggers: []p2ptest.Trigger{
+		Triggers: []p2ptest.Trigger{
-// 			{
+			{
-// 				Code: 4,
+				Code: 4,
-// 				Msg: &SubscribeMsg{
+				Msg: &SubscribeMsg{
-// 					Stream: stream,
+					Stream: stream,
-// 					History: &Range{
+					History: &Range{
-// 						From: 5,
+						From: 5,
-// 						To:   8,
+						To:   8,
-// 					},
+					},
-// 					Priority: Top,
+					Priority: Top,
-// 				},
+				},
-// 				Peer: peerID,
+				Peer: peerID,
-// 			},
+			},
-// 		},
+		},
-// 		Expects: []p2ptest.Expect{
+		Expects: []p2ptest.Expect{
-// 			{
+			{
-// 				Code: 1,
+				Code: 1,
-// 				Msg: &OfferedHashesMsg{
+				Msg: &OfferedHashesMsg{
-// 					Stream: NewStream("foo", nil, false),
+					Stream: NewStream("foo", nil, false),
-// 					HandoverProof: &HandoverProof{
+					HandoverProof: &HandoverProof{
-// 						Handover: &Handover{},
+						Handover: &Handover{},
-// 					},
+					},
-// 					Hashes:  make([]byte, HashSize),
+					Hashes: make([]byte, HashSize),
-// 					From:    6,
+					From:   6,
-// 					To:      9,
+					To:     9,
-// 				},
+				},
-// 				Peer: peerID,
+				Peer: peerID,
-// 			},
+			},
-// 			{
+			{
-// 				Code: 1,
+				Code: 1,
-// 				Msg: &OfferedHashesMsg{
+				Msg: &OfferedHashesMsg{
-// 					Stream: stream,
+					Stream: stream,
-// 					HandoverProof: &HandoverProof{
+					HandoverProof: &HandoverProof{
-// 						Handover: &Handover{},
+						Handover: &Handover{},
-// 					},
+					},
-// 					From:    1,
+					From:   1,
-// 					To:      1,
+					To:     1,
-// 					Hashes:  make([]byte, HashSize),
+					Hashes: make([]byte, HashSize),
-// 				},
+				},
-// 				Peer: peerID,
+				Peer: peerID,
-// 			},
+			},
-// 		},
+		},
-// 	})
+	})
 
-// 	if err != nil {
+	if err != nil {
-// 		t.Fatal(err)
+		t.Fatal(err)
-// 	}
+	}
-// }
+}
 
 func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
 	tester, streamer, _, teardown, err := newStreamerTester(t)
@@ -519,7 +517,7 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
 					Msg: &WantedHashesMsg{
 						Stream: stream,
 						Want:   []byte{5},
-						From:   8,
+						From:   9,
 						To:     0,
 					},
 					Peer: peerID,

swarm/network/stream/testing/testing.go

@@ -123,6 +123,7 @@ type RunConfig struct {
 	ConnLevel       int
 	ToAddr          func(discover.NodeID) *network.BzzAddr
 	Services        adapters.Services
+	DefaultService  string
 	EnableMsgEvents bool
 }
 
@@ -133,9 +134,13 @@ func NewSimulation(conf *RunConfig) (*Simulation, func(), error) {
 	if err != nil {
 		return nil, adapterTeardown, err
 	}
+	defaultService := "streamer"
+	if conf.DefaultService != "" {
+		defaultService = conf.DefaultService
+	}
 	net := simulations.NewNetwork(adapter, &simulations.NetworkConfig{
 		ID:             "0",
-		DefaultService: "streamer",
+		DefaultService: defaultService,
 	})
 	teardown := func() {
 		adapterTeardown()