mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-08-19 18:32:23 +00:00
swarm/network/stream: addressed PR comments
This commit is contained in:
parent
3facdcb0ec
commit
b866d31bb8
2 changed files with 19 additions and 10 deletions
|
|
@ -38,6 +38,7 @@ import (
|
||||||
"github.com/ethereum/go-ethereum/swarm/storage"
|
"github.com/ethereum/go-ethereum/swarm/storage"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
//Tests initializing a retrieve request
|
||||||
func TestStreamerRetrieveRequest(t *testing.T) {
|
func TestStreamerRetrieveRequest(t *testing.T) {
|
||||||
regOpts := &RegistryOptions{
|
regOpts := &RegistryOptions{
|
||||||
Retrieval: RetrievalClientOnly,
|
Retrieval: RetrievalClientOnly,
|
||||||
|
|
@ -64,7 +65,7 @@ func TestStreamerRetrieveRequest(t *testing.T) {
|
||||||
err = tester.TestExchanges(p2ptest.Exchange{
|
err = tester.TestExchanges(p2ptest.Exchange{
|
||||||
Label: "RetrieveRequestMsg",
|
Label: "RetrieveRequestMsg",
|
||||||
Expects: []p2ptest.Expect{
|
Expects: []p2ptest.Expect{
|
||||||
{
|
{ //start expecting a subscription for RETRIEVE_REQUEST due to `RetrievalClientOnly`
|
||||||
Code: 4,
|
Code: 4,
|
||||||
Msg: &SubscribeMsg{
|
Msg: &SubscribeMsg{
|
||||||
Stream: stream,
|
Stream: stream,
|
||||||
|
|
@ -73,7 +74,7 @@ func TestStreamerRetrieveRequest(t *testing.T) {
|
||||||
},
|
},
|
||||||
Peer: node.ID(),
|
Peer: node.ID(),
|
||||||
},
|
},
|
||||||
{
|
{ //expect a retrieve request message for the given hash
|
||||||
Code: 5,
|
Code: 5,
|
||||||
Msg: &RetrieveRequestMsg{
|
Msg: &RetrieveRequestMsg{
|
||||||
Addr: hash0[:],
|
Addr: hash0[:],
|
||||||
|
|
@ -89,10 +90,12 @@ func TestStreamerRetrieveRequest(t *testing.T) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
//Test requesting a chunk from a peer then issuing a "empty" OfferedHashesMsg (no hashes available yet)
|
||||||
|
//Should time out as the peer does not have the chunk (no syncing happened previously)
|
||||||
func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) {
|
func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) {
|
||||||
tester, streamer, _, teardown, err := newStreamerTester(t, &RegistryOptions{
|
tester, streamer, _, teardown, err := newStreamerTester(t, &RegistryOptions{
|
||||||
Retrieval: RetrievalEnabled,
|
Retrieval: RetrievalEnabled,
|
||||||
Syncing: SyncingDisabled,
|
Syncing: SyncingDisabled, //do no syncing
|
||||||
})
|
})
|
||||||
defer teardown()
|
defer teardown()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|
@ -106,15 +109,17 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) {
|
||||||
peer := streamer.getPeer(node.ID())
|
peer := streamer.getPeer(node.ID())
|
||||||
|
|
||||||
stream := NewStream(swarmChunkServerStreamName, "", true)
|
stream := NewStream(swarmChunkServerStreamName, "", true)
|
||||||
|
//simulate pre-subscription to RETRIEVE_REQUEST stream on peer
|
||||||
peer.handleSubscribeMsg(context.TODO(), &SubscribeMsg{
|
peer.handleSubscribeMsg(context.TODO(), &SubscribeMsg{
|
||||||
Stream: stream,
|
Stream: stream,
|
||||||
History: nil,
|
History: nil,
|
||||||
Priority: Top,
|
Priority: Top,
|
||||||
})
|
})
|
||||||
|
|
||||||
|
//test the exchange
|
||||||
err = tester.TestExchanges(p2ptest.Exchange{
|
err = tester.TestExchanges(p2ptest.Exchange{
|
||||||
Expects: []p2ptest.Expect{
|
Expects: []p2ptest.Expect{
|
||||||
{
|
{ //first expect a subscription to the RETRIEVE_REQUEST stream
|
||||||
Code: 4,
|
Code: 4,
|
||||||
Msg: &SubscribeMsg{
|
Msg: &SubscribeMsg{
|
||||||
Stream: stream,
|
Stream: stream,
|
||||||
|
|
@ -127,7 +132,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) {
|
||||||
}, p2ptest.Exchange{
|
}, p2ptest.Exchange{
|
||||||
Label: "RetrieveRequestMsg",
|
Label: "RetrieveRequestMsg",
|
||||||
Triggers: []p2ptest.Trigger{
|
Triggers: []p2ptest.Trigger{
|
||||||
{
|
{ //then the actual RETRIEVE_REQUEST....
|
||||||
Code: 5,
|
Code: 5,
|
||||||
Msg: &RetrieveRequestMsg{
|
Msg: &RetrieveRequestMsg{
|
||||||
Addr: chunk.Address()[:],
|
Addr: chunk.Address()[:],
|
||||||
|
|
@ -136,7 +141,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) {
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
Expects: []p2ptest.Expect{
|
Expects: []p2ptest.Expect{
|
||||||
{
|
{ //to which the peer responds with offered hashes
|
||||||
Code: 1,
|
Code: 1,
|
||||||
Msg: &OfferedHashesMsg{
|
Msg: &OfferedHashesMsg{
|
||||||
HandoverProof: nil,
|
HandoverProof: nil,
|
||||||
|
|
@ -149,6 +154,8 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) {
|
||||||
},
|
},
|
||||||
})
|
})
|
||||||
|
|
||||||
|
//should fail with a timeout as the peer we are requesting
|
||||||
|
//the chunk from does not have the chunk
|
||||||
expectedError := `exchange #1 "RetrieveRequestMsg": timed out`
|
expectedError := `exchange #1 "RetrieveRequestMsg": timed out`
|
||||||
if err == nil || err.Error() != expectedError {
|
if err == nil || err.Error() != expectedError {
|
||||||
t.Fatalf("Expected error %v, got %v", expectedError, err)
|
t.Fatalf("Expected error %v, got %v", expectedError, err)
|
||||||
|
|
@ -285,6 +292,7 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) {
|
||||||
|
|
||||||
node := tester.Nodes[0]
|
node := tester.Nodes[0]
|
||||||
|
|
||||||
|
//subscribe to custom stream
|
||||||
stream := NewStream("foo", "", true)
|
stream := NewStream("foo", "", true)
|
||||||
err = streamer.Subscribe(node.ID(), stream, NewRange(5, 8), Top)
|
err = streamer.Subscribe(node.ID(), stream, NewRange(5, 8), Top)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|
@ -297,7 +305,7 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) {
|
||||||
err = tester.TestExchanges(p2ptest.Exchange{
|
err = tester.TestExchanges(p2ptest.Exchange{
|
||||||
Label: "Subscribe message",
|
Label: "Subscribe message",
|
||||||
Expects: []p2ptest.Expect{
|
Expects: []p2ptest.Expect{
|
||||||
{
|
{ //first expect subscription to the custom stream...
|
||||||
Code: 4,
|
Code: 4,
|
||||||
Msg: &SubscribeMsg{
|
Msg: &SubscribeMsg{
|
||||||
Stream: stream,
|
Stream: stream,
|
||||||
|
|
@ -311,7 +319,8 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) {
|
||||||
p2ptest.Exchange{
|
p2ptest.Exchange{
|
||||||
Label: "ChunkDelivery message",
|
Label: "ChunkDelivery message",
|
||||||
Triggers: []p2ptest.Trigger{
|
Triggers: []p2ptest.Trigger{
|
||||||
{
|
{ //...then trigger a chunk delivery for the given chunk from peer in order for
|
||||||
|
//local node to get the chunk delivered
|
||||||
Code: 6,
|
Code: 6,
|
||||||
Msg: &ChunkDeliveryMsg{
|
Msg: &ChunkDeliveryMsg{
|
||||||
Addr: chunkKey,
|
Addr: chunkKey,
|
||||||
|
|
|
||||||
|
|
@ -62,12 +62,12 @@ const (
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
//Retrieval disabled
|
//Retrieval disabled. Used mostly for tests to isolate syncing features (i.e. syncing only)
|
||||||
RetrievalDisabled RetrievalOption = iota
|
RetrievalDisabled RetrievalOption = iota
|
||||||
//Only the client side of the retrieve request is registered.
|
//Only the client side of the retrieve request is registered.
|
||||||
//(light nodes do not serve retrieve requests)
|
//(light nodes do not serve retrieve requests)
|
||||||
//once the client is registered, subscription to retrieve request stream is always sent
|
//once the client is registered, subscription to retrieve request stream is always sent
|
||||||
RetrievalClientOn
|
RetrievalClientOnly
|
||||||
//Both client and server funcs are registered, subscribe sent automatically
|
//Both client and server funcs are registered, subscribe sent automatically
|
||||||
RetrievalEnabled
|
RetrievalEnabled
|
||||||
)
|
)
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue