diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index ec6148a0bf819600ec86eb7ed2753b9828dce140..949645558b17305591e87b7506fd9b986cfb3053 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -416,7 +416,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck return fmt.Errorf("No registry") } registry := item.(*Registry) - err = registry.Subscribe(sid, NewStream(swarmChunkServerStreamName, "", true), NewRange(0, 0), Top) + err = registry.Subscribe(sid, NewStream(swarmChunkServerStreamName, "", true), nil, Top) if err != nil { return err } diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go index 68503fe1f0e3d123537b5cd6a96abffb950e70c9..0fe3e5eb4f7b73463edfff0e8ca912230145899f 100644 --- a/swarm/network/stream/messages.go +++ b/swarm/network/stream/messages.go @@ -158,6 +158,7 @@ type SubscribeErrorMsg struct { } func (p *Peer) handleSubscribeErrorMsg(req *SubscribeErrorMsg) (err error) { + //TODO the error should be channeled to whoever calls the subscribe return fmt.Errorf("subscribe to peer %s: %v", p.ID(), req.Error) } diff --git a/swarm/network/stream/snapshot_retrieval_test.go b/swarm/network/stream/snapshot_retrieval_test.go index 7625e02d7a2ee0b5b5f5e1b1a6dd8c508eaac9fa..b81cfc0ca082b4ae576abd926c7c585effff872b 100644 --- a/swarm/network/stream/snapshot_retrieval_test.go +++ b/swarm/network/stream/snapshot_retrieval_test.go @@ -218,7 +218,7 @@ func runFileRetrievalTest(nodeCount int) error { reader, _ := fileStore.Retrieve(context.TODO(), hash) //check that we can read the file size and that it corresponds to the generated file size if s, err := reader.Size(ctx, nil); err != nil || s != int64(len(randomFiles[i])) { - log.Warn("Retrieve error", "err", err, "hash", hash, "nodeId", id) + log.Debug("Retrieve error", "err", err, "hash", hash, "nodeId", id) time.Sleep(500 * time.Millisecond) continue REPEAT } @@ -309,7 +309,7 @@ func runRetrievalTest(chunkCount int, nodeCount int) error { reader, _ := fileStore.Retrieve(context.TODO(), hash) //check that we can read the chunk size and that it corresponds to the generated chunk size if s, err := reader.Size(ctx, nil); err != nil || s != int64(chunkSize) { - log.Warn("Retrieve error", "err", err, "hash", hash, "nodeId", id, "size", s) + log.Debug("Retrieve error", "err", err, "hash", hash, "nodeId", id, "size", s) time.Sleep(500 * time.Millisecond) continue REPEAT } diff --git a/swarm/network/stream/snapshot_sync_test.go b/swarm/network/stream/snapshot_sync_test.go index 96e92c5cf10f20abb49b735d1b27a994093baef2..8d89f28cba860a10555da5dea1ebd03c9c45e75f 100644 --- a/swarm/network/stream/snapshot_sync_test.go +++ b/swarm/network/stream/snapshot_sync_test.go @@ -310,7 +310,7 @@ func runSim(conf *synctestConfig, ctx context.Context, sim *simulation.Simulatio _, err = lstore.Get(ctx, chunk) } if err != nil { - log.Warn(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id)) + log.Debug(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id)) // Do not get crazy with logging the warn message time.Sleep(500 * time.Millisecond) continue REPEAT @@ -514,7 +514,7 @@ func testSyncingViaDirectSubscribe(t *testing.T, chunkCount int, nodeCount int) _, err = lstore.Get(ctx, chunk) } if err != nil { - log.Warn(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id)) + log.Debug(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id)) // Do not get crazy with logging the warn message time.Sleep(500 * time.Millisecond) continue REPEAT diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index 1dc2a8cba854acaad8c3eef6168427fad264e335..9d0e6c68b8190e521cad27a74d8f9dde3a4712a0 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -18,6 +18,7 @@ package stream import ( "context" + "errors" "fmt" "math" "sync" @@ -96,7 +97,10 @@ func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.Sy delivery.getPeer = streamer.getPeer if options.DoServeRetrieve { - streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, _ string, _ bool) (Server, error) { + streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, _ string, live bool) (Server, error) { + if !live { + return nil, errors.New("only live retrieval requests supported") + } return NewSwarmChunkServer(delivery.chunkStore), nil }) } @@ -279,7 +283,6 @@ func (r *Registry) Subscribe(peerId enode.ID, s Stream, h *Range, priority uint8 if err != nil { return err } - if s.Live && h != nil { if err := peer.setClientParams( getHistoryStream(s),