diff --git a/swarm/network/stream/common_test.go b/swarm/network/stream/common_test.go index b00c3f452f..f3dde60b74 100644 --- a/swarm/network/stream/common_test.go +++ b/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(): - log.Warn(fmt.Sprintf("caught subscription error in stream %s: %v", s, 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. @@ -257,47 +276,84 @@ type testExternalClient struct { t []byte // wait0 chan bool // batchDone chan bool - hashes chan []byte + 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() { - self.hashes <- hash - return func() {} +func (c *testExternalClient) NeedData(hash []byte) func() { + chunk, _ := c.db.GetOrCreateRequest(hash) + if chunk.ReqC == nil { + return nil + } + c.hashes <- hash + return func() { + chunk.WaitToStore() + } } -func (self *testExternalClient) BatchDone(Stream, uint64, []byte, []byte) func() (*TakeoverProof, error) { - // close(self.batchDone) +func (c *testExternalClient) BatchDone(Stream, uint64, []byte, []byte) func() (*TakeoverProof, error) { return nil } -func (self *testExternalClient) Close() {} +func (c *testExternalClient) Close() {} + +const testExternalServerBatchSize = 10 type testExternalServer struct { - t []byte + 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, + t: t, + keyFunc: keyFunc, + sessionAt: sessionAt, + maxKeys: maxKeys, } } -func (self *testExternalServer) SetNextBatch(from uint64, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) { - return make([]byte, HashSize), from + 1, to + 1, nil, nil +func (s *testExternalServer) SetNextBatch(from uint64, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) { + 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) { - return nil, nil +func (s *testExternalServer) GetData([]byte) ([]byte, error) { + return make([]byte, 4096), nil } -func (self *testExternalServer) Close() { -} +func (s *testExternalServer) Close() {} diff --git a/swarm/network/stream/intervals_test.go b/swarm/network/stream/intervals_test.go index b26d04efe3..169583102a 100644 --- a/swarm/network/stream/intervals_test.go +++ b/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 @@ -71,11 +80,12 @@ func XTestIntervals(t *testing.T) { defaultSkipCheck = skipCheck toAddr = network.NewAddrFromNodeID conf := &streamTesting.RunConfig{ - Adapter: *adapter, - NodeCount: nodes, - ConnLevel: 1, - ToAddr: toAddr, - Services: services, + Adapter: *adapter, + NodeCount: nodes, + 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 } - // live stream - _, err = client.Subscribe(ctx, "stream_getHashes", liveHashesChan, sid, NewStream(externalStreamName, nil, true)) - if err != nil { + + liveSubErrC := make(chan error) + historySubErrC := make(chan error) + + go func() { + if live { + var err error + defer func() { liveSubErrC <- err }() + // live stream + 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 - _, err = client.Subscribe(ctx, "stream_getHashes", historyHashesChan, sid, NewStream(externalStreamName, nil, false)) - return err + if err := <-historySubErrC; err != nil { + return err + } + + return nil }) if err != nil { return err } + historyErrC := make(chan error) + go func() { - for i := uint64(0); i < 5; i++ { - h := binary.BigEndian.Uint64(<-historyHashesChan) - if h != i { - errc <- fmt.Errorf("") + defer close(historyErrC) + + 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 { + 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, }, } diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go index 0b9e68ef04..0c2ffae6ef 100644 --- a/swarm/network/stream/messages.go +++ b/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 diff --git a/swarm/network/stream/peer.go b/swarm/network/stream/peer.go index 468ce9febc..c9b623e792 100644 --- a/swarm/network/stream/peer.go +++ b/swarm/network/stream/peer.go @@ -18,7 +18,6 @@ package stream import ( "context" - "errors" "fmt" "sync" "time" @@ -32,22 +31,28 @@ import ( var sendTimeout = 5 * time.Second -var ( - errServerNotFound = errors.New("server not found") - errClientNotFound = errors.New("client not found") - errClientParamsNotFound = errors.New("client params not found") -) +type notFoundError struct { + t string + s Stream +} + +func newNotFoundError(t string, s Stream) *notFoundError { + return ¬FoundError{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 { *protocols.Peer - streamer *Registry - pq *pq.PriorityQueue - serverMu sync.RWMutex - clientMu sync.RWMutex - clientParamsMu sync.RWMutex - servers map[string]*server - clients map[string]*client + streamer *Registry + pq *pq.PriorityQueue + serverMu sync.RWMutex + clientMu sync.RWMutex // protects both clients and clientParams + servers map[string]*server + clients map[string]*client // clientParams map keeps required client arguments // that are set on Registry.Subscribe and used // on creating a new client in offered hashes handler. @@ -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() + + c = p.clients[sk] + if c != nil { + return + } + 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() - client := p.clients[s.String()] - if client == nil { - return nil, fmt.Errorf("client '%v' not provided to peer %v", s, p.ID()) + c = p.clients[sk] + if c != nil { + return c, nil } - return client, 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() - if p.clients[sk] != nil { - return fmt.Errorf("client %v already registered", sk) + c = p.clients[sk] + if c != nil { + 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 - 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 - } + 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, 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() - defer p.clientParamsMu.Unlock() + p.clientMu.Lock() + 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 { - p.clientParamsMu.Lock() - defer p.clientParamsMu.Unlock() +func (p *Peer) getClientParams(s Stream) (*clientParams, error) { + 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) 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 diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index b981a0ef8b..d0e230eb37 100644 --- a/swarm/network/stream/stream.go +++ b/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{ - priority: getHistoryPriority(priority), - }, + newClientParams(getHistoryPriority(priority), h.To), ); 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 diff --git a/swarm/network/stream/streamer_test.go b/swarm/network/stream/streamer_test.go index 1edd1fc03b..44175b254f 100644 --- a/swarm/network/stream/streamer_test.go +++ b/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 -// ordering is not guarrantied but fails if the order is wrong. -// func TestStreamerUpstreamSubscribeLiveAndHistory(t *testing.T) { -// tester, streamer, _, teardown, err := newStreamerTester(t) -// defer teardown() -// if err != nil { -// t.Fatal(err) -// } +func TestStreamerUpstreamSubscribeLiveAndHistory(t *testing.T) { + tester, streamer, _, teardown, err := newStreamerTester(t) + defer teardown() + 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) { -// return &testServer{ -// t: t, -// }, nil -// }) + streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) { + return &testServer{ + t: t, + }, nil + }) -// peerID := tester.IDs[0] + peerID := tester.IDs[0] -// err = tester.TestExchanges(p2ptest.Exchange{ -// Label: "Subscribe message", -// Triggers: []p2ptest.Trigger{ -// { -// Code: 4, -// Msg: &SubscribeMsg{ -// Stream: stream, -// History: &Range{ -// From: 5, -// To: 8, -// }, -// Priority: Top, -// }, -// Peer: peerID, -// }, -// }, -// Expects: []p2ptest.Expect{ -// { -// Code: 1, -// Msg: &OfferedHashesMsg{ -// Stream: NewStream("foo", nil, false), -// HandoverProof: &HandoverProof{ -// Handover: &Handover{}, -// }, -// Hashes: make([]byte, HashSize), -// From: 6, -// To: 9, -// }, -// Peer: peerID, -// }, -// { -// Code: 1, -// Msg: &OfferedHashesMsg{ -// Stream: stream, -// HandoverProof: &HandoverProof{ -// Handover: &Handover{}, -// }, -// From: 1, -// To: 1, -// Hashes: make([]byte, HashSize), -// }, -// Peer: peerID, -// }, -// }, -// }) + err = tester.TestExchanges(p2ptest.Exchange{ + Label: "Subscribe message", + Triggers: []p2ptest.Trigger{ + { + Code: 4, + Msg: &SubscribeMsg{ + Stream: stream, + History: &Range{ + From: 5, + To: 8, + }, + Priority: Top, + }, + Peer: peerID, + }, + }, + Expects: []p2ptest.Expect{ + { + Code: 1, + Msg: &OfferedHashesMsg{ + Stream: NewStream("foo", nil, false), + HandoverProof: &HandoverProof{ + Handover: &Handover{}, + }, + Hashes: make([]byte, HashSize), + From: 6, + To: 9, + }, + Peer: peerID, + }, + { + Code: 1, + Msg: &OfferedHashesMsg{ + Stream: stream, + HandoverProof: &HandoverProof{ + Handover: &Handover{}, + }, + From: 1, + To: 1, + Hashes: make([]byte, HashSize), + }, + Peer: peerID, + }, + }, + }) -// if err != nil { -// t.Fatal(err) -// } -// } + if err != nil { + 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, diff --git a/swarm/network/stream/testing/testing.go b/swarm/network/stream/testing/testing.go index 39b2c1df1d..25987f7330 100644 --- a/swarm/network/stream/testing/testing.go +++ b/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()