forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-19 10:22:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

swarm/network/stream: port to p2p/enode

Browse source
This commit is contained in:
Felix Lange 2018-09-11 18:59:44 +02:00
parent 46e37902fd
commit 421a83db7e
16 changed files with 151 additions and 158 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
swarm/network/stream/common_test.go
View file

@ -32,7 +32,7 @@ import (
"time"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/enode"
p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/network/simulation"
@ -114,12 +114,12 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *stora
delivery := NewDelivery(to, netStore)
netStore.NewNetFetcherFunc = network.NewFetcherFactory(delivery.RequestFromPeers, true).New
streamer := NewRegistry(addr, delivery, netStore, state.NewInmemoryStore(), nil)
streamer := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), nil)
teardown := func() {
streamer.Close()
removeDataDir()
}
protocolTester := p2ptest.NewProtocolTester(t, network.NewNodeIDFromAddr(addr), 1, streamer.runProtocol)
protocolTester := p2ptest.NewProtocolTester(t, addr.ID(), 1, streamer.runProtocol)
err = waitForPeers(streamer, 1*time.Second, 1)
if err != nil {
@ -240,7 +240,7 @@ func generateRandomFile() (string, error) {
}
//create a local store for the given node
func createTestLocalStorageForID(id discover.NodeID, addr *network.BzzAddr) (storage.ChunkStore, string, error) {
func createTestLocalStorageForID(id enode.ID, addr *network.BzzAddr) (storage.ChunkStore, string, error) {
var datadir string
var err error
datadir, err = ioutil.TempDir("", fmt.Sprintf("syncer-test-%s", id.TerminalString()))

6
swarm/network/stream/delivery.go
View file

@ -23,7 +23,7 @@ import (
"fmt"
"github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/swarm/log"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/spancontext"
@ -47,7 +47,7 @@ var (
type Delivery struct {
chunkStore storage.SyncChunkStore
kad *network.Kademlia
getPeer func(discover.NodeID) *Peer
getPeer func(enode.ID) *Peer
}
func NewDelivery(kad *network.Kademlia, chunkStore storage.SyncChunkStore) *Delivery {
@ -213,7 +213,7 @@ func (d *Delivery) handleChunkDeliveryMsg(ctx context.Context, sp *Peer, req *Ch
}
// RequestFromPeers sends a chunk retrieve request to
func (d *Delivery) RequestFromPeers(ctx context.Context, req *network.Request) (*discover.NodeID, chan struct{}, error) {
func (d *Delivery) RequestFromPeers(ctx context.Context, req *network.Request) (*enode.ID, chan struct{}, error) {
requestFromPeersCount.Inc(1)
var sp *Peer
spID := req.Source

49
swarm/network/stream/delivery_test.go
View file

@ -45,7 +45,7 @@ func TestStreamerRetrieveRequest(t *testing.T) {
t.Fatal(err)
}
peerID := tester.IDs[0]
node := tester.Nodes[0]
ctx := context.Background()
req := network.NewRequest(
@ -64,7 +64,7 @@ func TestStreamerRetrieveRequest(t *testing.T) {
Addr: hash0[:],
SkipCheck: true,
},
Peer: peerID,
Peer: node.ID(),
},
},
})
@ -81,11 +81,11 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) {
t.Fatal(err)
}
peerID := tester.IDs[0]
node := tester.Nodes[0]
chunk := storage.NewChunk(storage.Address(hash0[:]), nil)
peer := streamer.getPeer(peerID)
peer := streamer.getPeer(node.ID())
peer.handleSubscribeMsg(context.TODO(), &SubscribeMsg{
Stream: NewStream(swarmChunkServerStreamName, "", false),
@ -101,7 +101,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) {
Msg: &RetrieveRequestMsg{
Addr: chunk.Address()[:],
},
Peer: peerID,
Peer: node.ID(),
},
},
Expects: []p2ptest.Expect{
@ -113,7 +113,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) {
From: 0,
To: 0,
},
Peer: peerID,
Peer: node.ID(),
},
},
})
@ -133,8 +133,8 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
t.Fatal(err)
}
peerID := tester.IDs[0]
peer := streamer.getPeer(peerID)
node := tester.Nodes[0]
peer := streamer.getPeer(node.ID())
stream := NewStream(swarmChunkServerStreamName, "", false)
@ -159,7 +159,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
Msg: &RetrieveRequestMsg{
Addr: hash,
},
Peer: peerID,
Peer: node.ID(),
},
},
Expects: []p2ptest.Expect{
@ -175,7 +175,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
To: 32,
Stream: stream,
},
Peer: peerID,
Peer: node.ID(),
},
},
})
@ -200,7 +200,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
Addr: hash,
SkipCheck: true,
},
Peer: peerID,
Peer: node.ID(),
},
},
Expects: []p2ptest.Expect{
@ -210,7 +210,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
Addr: hash,
SData: hash,
},
Peer: peerID,
Peer: node.ID(),
},
},
})
@ -233,10 +233,10 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) {
}, nil
})
peerID := tester.IDs[0]
node := tester.Nodes[0]
stream := NewStream("foo", "", true)
err = streamer.Subscribe(peerID, stream, NewRange(5, 8), Top)
err = streamer.Subscribe(node.ID(), stream, NewRange(5, 8), Top)
if err != nil {
t.Fatalf("Expected no error, got %v", err)
}
@ -254,7 +254,7 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) {
History: NewRange(5, 8),
Priority: Top,
},
Peer: peerID,
Peer: node.ID(),
},
},
},
@ -267,7 +267,7 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) {
Addr: chunkKey,
SData: chunkData,
},
Peer: peerID,
Peer: node.ID(),
},
},
})
@ -314,10 +314,9 @@ func TestDeliveryFromNodes(t *testing.T) {
func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck bool) {
sim := simulation.New(map[string]simulation.ServiceFunc{
"streamer": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
id := ctx.Config.ID
addr := network.NewAddrFromNodeID(id)
store, datadir, err := createTestLocalStorageForID(id, addr)
node := ctx.Config.Node()
addr := network.NewAddr(node)
store, datadir, err := createTestLocalStorageForID(node.ID(), addr)
if err != nil {
return nil, nil, err
}
@ -336,7 +335,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
delivery := NewDelivery(kad, netStore)
netStore.NewNetFetcherFunc = network.NewFetcherFactory(delivery.RequestFromPeers, true).New
r := NewRegistry(addr, delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
SkipCheck: skipCheck,
})
bucket.Store(bucketKeyRegistry, r)
@ -502,9 +501,9 @@ func BenchmarkDeliveryFromNodesWithCheck(b *testing.B) {
func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skipCheck bool) {
sim := simulation.New(map[string]simulation.ServiceFunc{
"streamer": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
id := ctx.Config.ID
addr := network.NewAddrFromNodeID(id)
store, datadir, err := createTestLocalStorageForID(id, addr)
node := ctx.Config.Node()
addr := network.NewAddr(node)
store, datadir, err := createTestLocalStorageForID(node.ID(), addr)
if err != nil {
return nil, nil, err
}
@ -522,7 +521,7 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip
delivery := NewDelivery(kad, netStore)
netStore.NewNetFetcherFunc = network.NewFetcherFactory(delivery.RequestFromPeers, true).New
r := NewRegistry(addr, delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
SkipCheck: skipCheck,
DoSync: true,
SyncUpdateDelay: 0,

15
swarm/network/stream/intervals_test.go
View file

@ -30,7 +30,7 @@ import (
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/network/simulation"
@ -62,10 +62,9 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
sim := simulation.New(map[string]simulation.ServiceFunc{
"intervalsStreamer": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
id := ctx.Config.ID
addr := network.NewAddrFromNodeID(id)
store, datadir, err := createTestLocalStorageForID(id, addr)
n := ctx.Config.Node()
addr := network.NewAddr(n)
store, datadir, err := createTestLocalStorageForID(n.ID(), addr)
if err != nil {
return nil, nil, err
}
@ -83,7 +82,7 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
delivery := NewDelivery(kad, netStore)
netStore.NewNetFetcherFunc = network.NewFetcherFactory(delivery.RequestFromPeers, true).New
r := NewRegistry(addr, delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
SkipCheck: skipCheck,
})
bucket.Store(bucketKeyRegistry, r)
@ -281,7 +280,7 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
}
}
func getHashes(ctx context.Context, r *Registry, peerID discover.NodeID, s Stream) (chan []byte, error) {
func getHashes(ctx context.Context, r *Registry, peerID enode.ID, s Stream) (chan []byte, error) {
peer := r.getPeer(peerID)
client, err := peer.getClient(ctx, s)
@ -294,7 +293,7 @@ func getHashes(ctx context.Context, r *Registry, peerID discover.NodeID, s Strea
return c.hashes, nil
}
func enableNotifications(r *Registry, peerID discover.NodeID, s Stream) error {
func enableNotifications(r *Registry, peerID enode.ID, s Stream) error {
peer := r.getPeer(peerID)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)

6
swarm/network/stream/messages.go
View file

@ -75,7 +75,7 @@ type RequestSubscriptionMsg struct {
}
func (p *Peer) handleRequestSubscription(ctx context.Context, req *RequestSubscriptionMsg) (err error) {
log.Debug(fmt.Sprintf("handleRequestSubscription: streamer %s to subscribe to %s with stream %s", p.streamer.addr.ID(), p.ID(), req.Stream))
log.Debug(fmt.Sprintf("handleRequestSubscription: streamer %s to subscribe to %s with stream %s", p.streamer.addr, p.ID(), req.Stream))
return p.streamer.Subscribe(p.ID(), req.Stream, req.History, req.Priority)
}
@ -92,7 +92,7 @@ func (p *Peer) handleSubscribeMsg(ctx context.Context, req *SubscribeMsg) (err e
}
}()
log.Debug("received subscription", "from", p.streamer.addr.ID(), "peer", p.ID(), "stream", req.Stream, "history", req.History)
log.Debug("received subscription", "from", p.streamer.addr, "peer", p.ID(), "stream", req.Stream, "history", req.History)
f, err := p.streamer.GetServerFunc(req.Stream.Name)
if err != nil {
@ -254,7 +254,7 @@ func (p *Peer) handleOfferedHashesMsg(ctx context.Context, req *OfferedHashesMsg
c.sessionAt = req.From
}
from, to := c.nextBatch(req.To + 1)
log.Trace("set next batch", "peer", p.ID(), "stream", req.Stream, "from", req.From, "to", req.To, "addr", p.streamer.addr.ID())
log.Trace("set next batch", "peer", p.ID(), "stream", req.Stream, "from", req.From, "to", req.To, "addr", p.streamer.addr)
if from == to {
return nil
}

36
swarm/network/stream/snapshot_retrieval_test.go
View file

@ -24,7 +24,7 @@ import (
"time"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/swarm/log"
"github.com/ethereum/go-ethereum/swarm/network"
@ -116,10 +116,9 @@ The snapshot should have 'streamer' in its service list.
func runFileRetrievalTest(nodeCount int) error {
sim := simulation.New(map[string]simulation.ServiceFunc{
"streamer": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
id := ctx.Config.ID
addr := network.NewAddrFromNodeID(id)
store, datadir, err := createTestLocalStorageForID(id, addr)
node := ctx.Config.Node()
addr := network.NewAddr(node)
store, datadir, err := createTestLocalStorageForID(node.ID(), addr)
if err != nil {
return nil, nil, err
}
@ -134,7 +133,7 @@ func runFileRetrievalTest(nodeCount int) error {
delivery := NewDelivery(kad, netStore)
netStore.NewNetFetcherFunc = network.NewFetcherFactory(delivery.RequestFromPeers, true).New
r := NewRegistry(addr, delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
DoSync: true,
SyncUpdateDelay: 3 * time.Second,
})
@ -158,9 +157,9 @@ func runFileRetrievalTest(nodeCount int) error {
conf := &synctestConfig{}
//map of discover ID to indexes of chunks expected at that ID
conf.idToChunksMap = make(map[discover.NodeID][]int)
conf.idToChunksMap = make(map[enode.ID][]int)
//map of overlay address to discover ID
conf.addrToIDMap = make(map[string]discover.NodeID)
conf.addrToIDMap = make(map[string]enode.ID)
//array where the generated chunk hashes will be stored
conf.hashes = make([]storage.Address, 0)
@ -176,11 +175,11 @@ func runFileRetrievalTest(nodeCount int) error {
nodeIDs := sim.UpNodeIDs()
for _, n := range nodeIDs {
//get the kademlia overlay address from this ID
a := network.ToOverlayAddr(n.Bytes())
a := n.Bytes()
//append it to the array of all overlay addresses
conf.addrs = append(conf.addrs, a)
//the proximity calculation is on overlay addr,
//the p2p/simulations check func triggers on discover.NodeID,
//the p2p/simulations check func triggers on enode.ID,
//so we need to know which overlay addr maps to which nodeID
conf.addrToIDMap[string(a)] = n
}
@ -266,10 +265,9 @@ The snapshot should have 'streamer' in its service list.
func runRetrievalTest(chunkCount int, nodeCount int) error {
sim := simulation.New(map[string]simulation.ServiceFunc{
"streamer": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
id := ctx.Config.ID
addr := network.NewAddrFromNodeID(id)
store, datadir, err := createTestLocalStorageForID(id, addr)
node := ctx.Config.Node()
addr := network.NewAddr(node)
store, datadir, err := createTestLocalStorageForID(node.ID(), addr)
if err != nil {
return nil, nil, err
}
@ -284,7 +282,7 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
delivery := NewDelivery(kad, netStore)
netStore.NewNetFetcherFunc = network.NewFetcherFactory(delivery.RequestFromPeers, true).New
r := NewRegistry(addr, delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
DoSync: true,
SyncUpdateDelay: 0,
})
@ -307,9 +305,9 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
conf := &synctestConfig{}
//map of discover ID to indexes of chunks expected at that ID
conf.idToChunksMap = make(map[discover.NodeID][]int)
conf.idToChunksMap = make(map[enode.ID][]int)
//map of overlay address to discover ID
conf.addrToIDMap = make(map[string]discover.NodeID)
conf.addrToIDMap = make(map[string]enode.ID)
//array where the generated chunk hashes will be stored
conf.hashes = make([]storage.Address, 0)
@ -323,11 +321,11 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
nodeIDs := sim.UpNodeIDs()
for _, n := range nodeIDs {
//get the kademlia overlay address from this ID
a := network.ToOverlayAddr(n.Bytes())
a := n.Bytes()
//append it to the array of all overlay addresses
conf.addrs = append(conf.addrs, a)
//the proximity calculation is on overlay addr,
//the p2p/simulations check func triggers on discover.NodeID,
//the p2p/simulations check func triggers on enode.ID,
//so we need to know which overlay addr maps to which nodeID
conf.addrToIDMap[string(a)] = n
}

47
swarm/network/stream/snapshot_sync_test.go
View file

@ -30,7 +30,7 @@ import (
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/network/simulation"
@ -45,14 +45,14 @@ const MaxTimeout = 600
type synctestConfig struct {
addrs [][]byte
hashes []storage.Address
idToChunksMap map[discover.NodeID][]int
idToChunksMap map[enode.ID][]int
//chunksToNodesMap map[string][]int
addrToIDMap map[string]discover.NodeID
addrToIDMap map[string]enode.ID
}
// Tests in this file should not request chunks from peers.
// This function will panic indicating that there is a problem if request has been made.
func dummyRequestFromPeers(_ context.Context, req *network.Request) (*discover.NodeID, chan struct{}, error) {
func dummyRequestFromPeers(_ context.Context, req *network.Request) (*enode.ID, chan struct{}, error) {
panic(fmt.Sprintf("unexpected request: address %s, source %s", req.Addr.String(), req.Source.String()))
}
@ -134,10 +134,9 @@ func TestSyncingViaDirectSubscribe(t *testing.T) {
func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) {
sim := simulation.New(map[string]simulation.ServiceFunc{
"streamer": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
id := ctx.Config.ID
addr := network.NewAddrFromNodeID(id)
store, datadir, err := createTestLocalStorageForID(id, addr)
n := ctx.Config.Node()
addr := network.NewAddr(n)
store, datadir, err := createTestLocalStorageForID(n.ID(), addr)
if err != nil {
return nil, nil, err
}
@ -151,7 +150,7 @@ func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) {
delivery := NewDelivery(kad, netStore)
netStore.NewNetFetcherFunc = network.NewFetcherFactory(dummyRequestFromPeers, true).New
r := NewRegistry(addr, delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
DoSync: true,
SyncUpdateDelay: 3 * time.Second,
})
@ -173,9 +172,9 @@ func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) {
conf := &synctestConfig{}
//map of discover ID to indexes of chunks expected at that ID
conf.idToChunksMap = make(map[discover.NodeID][]int)
conf.idToChunksMap = make(map[enode.ID][]int)
//map of overlay address to discover ID
conf.addrToIDMap = make(map[string]discover.NodeID)
conf.addrToIDMap = make(map[string]enode.ID)
//array where the generated chunk hashes will be stored
conf.hashes = make([]storage.Address, 0)
@ -209,11 +208,11 @@ func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) {
nodeIDs := sim.UpNodeIDs()
for _, n := range nodeIDs {
//get the kademlia overlay address from this ID
a := network.ToOverlayAddr(n.Bytes())
a := n.Bytes()
//append it to the array of all overlay addresses
conf.addrs = append(conf.addrs, a)
//the proximity calculation is on overlay addr,
//the p2p/simulations check func triggers on discover.NodeID,
//the p2p/simulations check func triggers on enode.ID,
//so we need to know which overlay addr maps to which nodeID
conf.addrToIDMap[string(a)] = n
}
@ -317,10 +316,9 @@ kademlia network. The snapshot should have 'streamer' in its service list.
func testSyncingViaDirectSubscribe(t *testing.T, chunkCount int, nodeCount int) error {
sim := simulation.New(map[string]simulation.ServiceFunc{
"streamer": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
id := ctx.Config.ID
addr := network.NewAddrFromNodeID(id)
store, datadir, err := createTestLocalStorageForID(id, addr)
n := ctx.Config.Node()
addr := network.NewAddr(n)
store, datadir, err := createTestLocalStorageForID(n.ID(), addr)
if err != nil {
return nil, nil, err
}
@ -334,7 +332,7 @@ func testSyncingViaDirectSubscribe(t *testing.T, chunkCount int, nodeCount int)
delivery := NewDelivery(kad, netStore)
netStore.NewNetFetcherFunc = network.NewFetcherFactory(dummyRequestFromPeers, true).New
r := NewRegistry(addr, delivery, netStore, state.NewInmemoryStore(), nil)
r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), nil)
bucket.Store(bucketKeyRegistry, r)
fileStore := storage.NewFileStore(netStore, storage.NewFileStoreParams())
@ -357,9 +355,9 @@ func testSyncingViaDirectSubscribe(t *testing.T, chunkCount int, nodeCount int)
conf := &synctestConfig{}
//map of discover ID to indexes of chunks expected at that ID
conf.idToChunksMap = make(map[discover.NodeID][]int)
conf.idToChunksMap = make(map[enode.ID][]int)
//map of overlay address to discover ID
conf.addrToIDMap = make(map[string]discover.NodeID)
conf.addrToIDMap = make(map[string]enode.ID)
//array where the generated chunk hashes will be stored
conf.hashes = make([]storage.Address, 0)
@ -390,11 +388,11 @@ func testSyncingViaDirectSubscribe(t *testing.T, chunkCount int, nodeCount int)
nodeIDs := sim.UpNodeIDs()
for _, n := range nodeIDs {
//get the kademlia overlay address from this ID
a := network.ToOverlayAddr(n.Bytes())
a := n.Bytes()
//append it to the array of all overlay addresses
conf.addrs = append(conf.addrs, a)
//the proximity calculation is on overlay addr,
//the p2p/simulations check func triggers on discover.NodeID,
//the p2p/simulations check func triggers on enode.ID,
//so we need to know which overlay addr maps to which nodeID
conf.addrToIDMap[string(a)] = n
}
@ -525,9 +523,8 @@ func startSyncing(r *Registry, conf *synctestConfig) (int, error) {
kad := r.delivery.kad
subCnt := 0
//iterate over each bin and solicit needed subscription to bins
kad.EachBin(r.addr.Over(), pof, 0, func(conn *network.Peer, po int) bool {
kad.EachBin(r.addr[:], pof, 0, func(conn *network.Peer, po int) bool {
//identify begin and start index of the bin(s) we want to subscribe to
subCnt++
err = r.RequestSubscription(conf.addrToIDMap[string(conn.Address())], NewStream("SYNC", FormatSyncBinKey(uint8(po)), true), NewRange(0, 0), High)
if err != nil {
@ -580,7 +577,7 @@ func mapKeysToNodes(conf *synctestConfig) {
}
//upload a file(chunks) to a single local node store
func uploadFileToSingleNodeStore(id discover.NodeID, chunkCount int, lstore *storage.LocalStore) ([]storage.Address, error) {
func uploadFileToSingleNodeStore(id enode.ID, chunkCount int, lstore *storage.LocalStore) ([]storage.Address, error) {
log.Debug(fmt.Sprintf("Uploading to node id: %s", id))
fileStore := storage.NewFileStore(lstore, storage.NewFileStoreParams())
size := chunkSize

36
swarm/network/stream/stream.go
View file

@ -25,7 +25,7 @@ import (
"github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/protocols"
"github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/swarm/log"
@ -48,15 +48,15 @@ const (
// Registry registry for outgoing and incoming streamer constructors
type Registry struct {
addr enode.ID
api *API
addr *network.BzzAddr
skipCheck bool
clientMu sync.RWMutex
serverMu sync.RWMutex
peersMu sync.RWMutex
serverFuncs map[string]func(*Peer, string, bool) (Server, error)
clientFuncs map[string]func(*Peer, string, bool) (Client, error)
peers map[discover.NodeID]*Peer
peers map[enode.ID]*Peer
delivery *Delivery
intervalsStore state.Store
doRetrieve bool
@ -71,7 +71,7 @@ type RegistryOptions struct {
}
// NewRegistry is Streamer constructor
func NewRegistry(addr *network.BzzAddr, delivery *Delivery, syncChunkStore storage.SyncChunkStore, intervalsStore state.Store, options *RegistryOptions) *Registry {
func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.SyncChunkStore, intervalsStore state.Store, options *RegistryOptions) *Registry {
if options == nil {
options = &RegistryOptions{}
}
@ -79,11 +79,11 @@ func NewRegistry(addr *network.BzzAddr, delivery *Delivery, syncChunkStore stora
options.SyncUpdateDelay = 15 * time.Second
}
streamer := &Registry{
addr: addr,
addr: localID,
skipCheck: options.SkipCheck,
serverFuncs: make(map[string]func(*Peer, string, bool) (Server, error)),
clientFuncs: make(map[string]func(*Peer, string, bool) (Client, error)),
peers: make(map[discover.NodeID]*Peer),
peers: make(map[enode.ID]*Peer),
delivery: delivery,
intervalsStore: intervalsStore,
doRetrieve: options.DoRetrieve,
@ -220,7 +220,7 @@ func (r *Registry) GetServerFunc(stream string) (func(*Peer, string, bool) (Serv
return f, nil
}
func (r *Registry) RequestSubscription(peerId discover.NodeID, s Stream, h *Range, prio uint8) error {
func (r *Registry) RequestSubscription(peerId enode.ID, s Stream, h *Range, prio uint8) error {
// check if the stream is registered
if _, err := r.GetServerFunc(s.Name); err != nil {
return err
@ -248,7 +248,7 @@ func (r *Registry) RequestSubscription(peerId discover.NodeID, s Stream, h *Rang
}
// Subscribe initiates the streamer
func (r *Registry) Subscribe(peerId discover.NodeID, s Stream, h *Range, priority uint8) error {
func (r *Registry) Subscribe(peerId enode.ID, s Stream, h *Range, priority uint8) error {
// check if the stream is registered
if _, err := r.GetClientFunc(s.Name); err != nil {
return err
@ -288,7 +288,7 @@ func (r *Registry) Subscribe(peerId discover.NodeID, s Stream, h *Range, priorit
return peer.SendPriority(context.TODO(), msg, priority)
}
func (r *Registry) Unsubscribe(peerId discover.NodeID, s Stream) error {
func (r *Registry) Unsubscribe(peerId enode.ID, s Stream) error {
peer := r.getPeer(peerId)
if peer == nil {
return fmt.Errorf("peer not found %v", peerId)
@ -307,7 +307,7 @@ func (r *Registry) Unsubscribe(peerId discover.NodeID, s Stream) error {
// Quit sends the QuitMsg to the peer to remove the
// stream peer client and terminate the streaming.
func (r *Registry) Quit(peerId discover.NodeID, s Stream) error {
func (r *Registry) Quit(peerId enode.ID, s Stream) error {
peer := r.getPeer(peerId)
if peer == nil {
log.Debug("stream quit: peer not found", "peer", peerId, "stream", s)
@ -327,7 +327,7 @@ func (r *Registry) NodeInfo() interface{} {
return nil
}
func (r *Registry) PeerInfo(id discover.NodeID) interface{} {
func (r *Registry) PeerInfo(id enode.ID) interface{} {
return nil
}
@ -335,7 +335,7 @@ func (r *Registry) Close() error {
return r.intervalsStore.Close()
}
func (r *Registry) getPeer(peerId discover.NodeID) *Peer {
func (r *Registry) getPeer(peerId enode.ID) *Peer {
r.peersMu.RLock()
defer r.peersMu.RUnlock()
@ -390,7 +390,7 @@ func (r *Registry) updateSyncing() {
// map of all SYNC streams for all peers
// used at the and of the function to remove servers
// that are not needed anymore
subs := make(map[discover.NodeID]map[Stream]struct{})
subs := make(map[enode.ID]map[Stream]struct{})
r.peersMu.RLock()
for id, peer := range r.peers {
peer.serverMu.RLock()
@ -407,8 +407,8 @@ func (r *Registry) updateSyncing() {
r.peersMu.RUnlock()
// request subscriptions for all nodes and bins
kad.EachBin(r.addr.Over(), pot.DefaultPof(256), 0, func(p *network.Peer, bin int) bool {
log.Debug(fmt.Sprintf("Requesting subscription by: registry %s from peer %s for bin: %d", r.addr.ID(), p.ID(), bin))
kad.EachBin(r.addr[:], pot.DefaultPof(256), 0, func(p *network.Peer, bin int) bool {
log.Debug(fmt.Sprintf("Requesting subscription by: registry %s from peer %s for bin: %d", r.addr, p.ID(), bin))
// bin is always less then 256 and it is safe to convert it to type uint8
stream := NewStream("SYNC", FormatSyncBinKey(uint8(bin)), true)
@ -446,7 +446,7 @@ func (r *Registry) updateSyncing() {
func (r *Registry) runProtocol(p *p2p.Peer, rw p2p.MsgReadWriter) error {
peer := protocols.NewPeer(p, rw, Spec)
bp := network.NewBzzPeer(peer, r.addr)
bp := network.NewBzzPeer(peer)
np := network.NewPeer(bp, r.delivery.kad)
r.delivery.kad.On(np)
defer r.delivery.kad.Off(np)
@ -724,10 +724,10 @@ func NewAPI(r *Registry) *API {
}
}
func (api *API) SubscribeStream(peerId discover.NodeID, s Stream, history *Range, priority uint8) error {
func (api *API) SubscribeStream(peerId enode.ID, s Stream, history *Range, priority uint8) error {
return api.streamer.Subscribe(peerId, s, history, priority)
}
func (api *API) UnsubscribeStream(peerId discover.NodeID, s Stream) error {
func (api *API) UnsubscribeStream(peerId enode.ID, s Stream) error {
return api.streamer.Unsubscribe(peerId, s)
}

78
swarm/network/stream/streamer_test.go
View file

@ -34,7 +34,7 @@ func TestStreamerSubscribe(t *testing.T) {
}
stream := NewStream("foo", "", true)
err = streamer.Subscribe(tester.IDs[0], stream, NewRange(0, 0), Top)
err = streamer.Subscribe(tester.Nodes[0].ID(), stream, NewRange(0, 0), Top)
if err == nil || err.Error() != "stream foo not registered" {
t.Fatalf("Expected error %v, got %v", "stream foo not registered", err)
}
@ -48,7 +48,7 @@ func TestStreamerRequestSubscription(t *testing.T) {
}
stream := NewStream("foo", "", false)
err = streamer.RequestSubscription(tester.IDs[0], stream, &Range{}, Top)
err = streamer.RequestSubscription(tester.Nodes[0].ID(), stream, &Range{}, Top)
if err == nil || err.Error() != "stream foo not registered" {
t.Fatalf("Expected error %v, got %v", "stream foo not registered", err)
}
@ -135,10 +135,10 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
return newTestClient(t), nil
})
peerID := tester.IDs[0]
node := tester.Nodes[0]
stream := NewStream("foo", "", true)
err = streamer.Subscribe(peerID, stream, NewRange(5, 8), Top)
err = streamer.Subscribe(node.ID(), stream, NewRange(5, 8), Top)
if err != nil {
t.Fatalf("Expected no error, got %v", err)
}
@ -154,7 +154,7 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
History: NewRange(5, 8),
Priority: Top,
},
Peer: peerID,
Peer: node.ID(),
},
},
},
@ -173,7 +173,7 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
To: 8,
Stream: stream,
},
Peer: peerID,
Peer: node.ID(),
},
},
Expects: []p2ptest.Expect{
@ -185,7 +185,7 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
From: 9,
To: 0,
},
Peer: peerID,
Peer: node.ID(),
},
},
},
@ -194,7 +194,7 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
t.Fatal(err)
}
err = streamer.Unsubscribe(peerID, stream)
err = streamer.Unsubscribe(node.ID(), stream)
if err != nil {
t.Fatalf("Expected no error, got %v", err)
}
@ -207,7 +207,7 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
Msg: &UnsubscribeMsg{
Stream: stream,
},
Peer: peerID,
Peer: node.ID(),
},
},
})
@ -230,7 +230,7 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
return newTestServer(t), nil
})
peerID := tester.IDs[0]
node := tester.Nodes[0]
err = tester.TestExchanges(p2ptest.Exchange{
Label: "Subscribe message",
@ -242,7 +242,7 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
History: NewRange(5, 8),
Priority: Top,
},
Peer: peerID,
Peer: node.ID(),
},
},
Expects: []p2ptest.Expect{
@ -257,7 +257,7 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
From: 6,
To: 9,
},
Peer: peerID,
Peer: node.ID(),
},
},
})
@ -274,7 +274,7 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
Msg: &UnsubscribeMsg{
Stream: stream,
},
Peer: peerID,
Peer: node.ID(),
},
},
})
@ -297,7 +297,7 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchangeLive(t *testing.T) {
return newTestServer(t), nil
})
peerID := tester.IDs[0]
node := tester.Nodes[0]
err = tester.TestExchanges(p2ptest.Exchange{
Label: "Subscribe message",
@ -308,7 +308,7 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchangeLive(t *testing.T) {
Stream: stream,
Priority: Top,
},
Peer: peerID,
Peer: node.ID(),
},
},
Expects: []p2ptest.Expect{
@ -323,7 +323,7 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchangeLive(t *testing.T) {
From: 1,
To: 1,
},
Peer: peerID,
Peer: node.ID(),
},
},
})
@ -340,7 +340,7 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchangeLive(t *testing.T) {
Msg: &UnsubscribeMsg{
Stream: stream,
},
Peer: peerID,
Peer: node.ID(),
},
},
})
@ -363,7 +363,7 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) {
stream := NewStream("bar", "", true)
peerID := tester.IDs[0]
node := tester.Nodes[0]
err = tester.TestExchanges(p2ptest.Exchange{
Label: "Subscribe message",
@ -375,7 +375,7 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) {
History: NewRange(5, 8),
Priority: Top,
},
Peer: peerID,
Peer: node.ID(),
},
},
Expects: []p2ptest.Expect{
@ -384,7 +384,7 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) {
Msg: &SubscribeErrorMsg{
Error: "stream bar not registered",
},
Peer: peerID,
Peer: node.ID(),
},
},
})
@ -409,7 +409,7 @@ func TestStreamerUpstreamSubscribeLiveAndHistory(t *testing.T) {
}, nil
})
peerID := tester.IDs[0]
node := tester.Nodes[0]
err = tester.TestExchanges(p2ptest.Exchange{
Label: "Subscribe message",
@ -421,7 +421,7 @@ func TestStreamerUpstreamSubscribeLiveAndHistory(t *testing.T) {
History: NewRange(5, 8),
Priority: Top,
},
Peer: peerID,
Peer: node.ID(),
},
},
Expects: []p2ptest.Expect{
@ -436,7 +436,7 @@ func TestStreamerUpstreamSubscribeLiveAndHistory(t *testing.T) {
From: 6,
To: 9,
},
Peer: peerID,
Peer: node.ID(),
},
{
Code: 1,
@ -449,7 +449,7 @@ func TestStreamerUpstreamSubscribeLiveAndHistory(t *testing.T) {
To: 1,
Hashes: make([]byte, HashSize),
},
Peer: peerID,
Peer: node.ID(),
},
},
})
@ -475,9 +475,9 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
return tc, nil
})
peerID := tester.IDs[0]
node := tester.Nodes[0]
err = streamer.Subscribe(peerID, stream, NewRange(5, 8), Top)
err = streamer.Subscribe(node.ID(), stream, NewRange(5, 8), Top)
if err != nil {
t.Fatalf("Expected no error, got %v", err)
}
@ -492,7 +492,7 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
History: NewRange(5, 8),
Priority: Top,
},
Peer: peerID,
Peer: node.ID(),
},
},
},
@ -510,7 +510,7 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
To: 8,
Stream: stream,
},
Peer: peerID,
Peer: node.ID(),
},
},
Expects: []p2ptest.Expect{
@ -522,7 +522,7 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
From: 9,
To: 0,
},
Peer: peerID,
Peer: node.ID(),
},
},
})
@ -569,10 +569,10 @@ func TestStreamerRequestSubscriptionQuitMsgExchange(t *testing.T) {
return newTestServer(t), nil
})
peerID := tester.IDs[0]
node := tester.Nodes[0]
stream := NewStream("foo", "", true)
err = streamer.RequestSubscription(peerID, stream, NewRange(5, 8), Top)
err = streamer.RequestSubscription(node.ID(), stream, NewRange(5, 8), Top)
if err != nil {
t.Fatalf("Expected no error, got %v", err)
}
@ -588,7 +588,7 @@ func TestStreamerRequestSubscriptionQuitMsgExchange(t *testing.T) {
History: NewRange(5, 8),
Priority: Top,
},
Peer: peerID,
Peer: node.ID(),
},
},
},
@ -602,7 +602,7 @@ func TestStreamerRequestSubscriptionQuitMsgExchange(t *testing.T) {
History: NewRange(5, 8),
Priority: Top,
},
Peer: peerID,
Peer: node.ID(),
},
},
Expects: []p2ptest.Expect{
@ -617,7 +617,7 @@ func TestStreamerRequestSubscriptionQuitMsgExchange(t *testing.T) {
From: 6,
To: 9,
},
Peer: peerID,
Peer: node.ID(),
},
{
Code: 1,
@ -630,7 +630,7 @@ func TestStreamerRequestSubscriptionQuitMsgExchange(t *testing.T) {
To: 1,
Hashes: make([]byte, HashSize),
},
Peer: peerID,
Peer: node.ID(),
},
},
},
@ -639,7 +639,7 @@ func TestStreamerRequestSubscriptionQuitMsgExchange(t *testing.T) {
t.Fatal(err)
}
err = streamer.Quit(peerID, stream)
err = streamer.Quit(node.ID(), stream)
if err != nil {
t.Fatalf("Expected no error, got %v", err)
}
@ -652,7 +652,7 @@ func TestStreamerRequestSubscriptionQuitMsgExchange(t *testing.T) {
Msg: &QuitMsg{
Stream: stream,
},
Peer: peerID,
Peer: node.ID(),
},
},
})
@ -663,7 +663,7 @@ func TestStreamerRequestSubscriptionQuitMsgExchange(t *testing.T) {
historyStream := getHistoryStream(stream)
err = streamer.Quit(peerID, historyStream)
err = streamer.Quit(node.ID(), historyStream)
if err != nil {
t.Fatalf("Expected no error, got %v", err)
}
@ -676,7 +676,7 @@ func TestStreamerRequestSubscriptionQuitMsgExchange(t *testing.T) {
Msg: &QuitMsg{
Stream: historyStream,
},
Peer: peerID,
Peer: node.ID(),
},
},
})

2
swarm/network/stream/syncer.go
View file

@ -197,7 +197,7 @@ func NewSwarmSyncerClient(p *Peer, store storage.SyncChunkStore, stream Stream)
// // StartSyncing is called on the Peer to start the syncing process
// // the idea is that it is called only after kademlia is close to healthy
// func StartSyncing(s *Streamer, peerId discover.NodeID, po uint8, nn bool) {
// func StartSyncing(s *Streamer, peerId enode.ID, po uint8, nn bool) {
// lastPO := po
// if nn {
// lastPO = maxPO

16
swarm/network/stream/syncer_test.go
View file

@ -31,7 +31,7 @@ import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/swarm/log"
"github.com/ethereum/go-ethereum/swarm/network"
@ -50,7 +50,7 @@ func TestSyncerSimulation(t *testing.T) {
testSyncBetweenNodes(t, 16, 1, dataChunkCount, true, 1)
}
func createMockStore(globalStore *mockdb.GlobalStore, id discover.NodeID, addr *network.BzzAddr) (lstore storage.ChunkStore, datadir string, err error) {
func createMockStore(globalStore *mockdb.GlobalStore, id enode.ID, addr *network.BzzAddr) (lstore storage.ChunkStore, datadir string, err error) {
address := common.BytesToAddress(id.Bytes())
mockStore := globalStore.NewNodeStore(address)
params := storage.NewDefaultLocalStoreParams()
@ -72,8 +72,8 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
var globalStore *mockdb.GlobalStore
var gDir, datadir string
id := ctx.Config.ID
addr := network.NewAddrFromNodeID(id)
node := ctx.Config.Node()
addr := network.NewAddr(node)
//hack to put addresses in same space
addr.OAddr[0] = byte(0)
@ -82,9 +82,9 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
if err != nil {
return nil, nil, fmt.Errorf("Something went wrong; using mockStore enabled but globalStore is nil")
}
store, datadir, err = createMockStore(globalStore, id, addr)
store, datadir, err = createMockStore(globalStore, node.ID(), addr)
} else {
store, datadir, err = createTestLocalStorageForID(id, addr)
store, datadir, err = createTestLocalStorageForID(node.ID(), addr)
}
if err != nil {
return nil, nil, err
@ -113,7 +113,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
bucket.Store(bucketKeyDelivery, delivery)
r := NewRegistry(addr, delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
SkipCheck: skipCheck,
})
@ -139,7 +139,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error {
nodeIDs := sim.UpNodeIDs()
nodeIndex := make(map[discover.NodeID]int)
nodeIndex := make(map[enode.ID]int)
for i, id := range nodeIDs {
nodeIndex[id] = i
}

2
swarm/network/stream/testing/snapshot_128.json
View file

File diff suppressed because one or more lines are too long

2
swarm/network/stream/testing/snapshot_16.json
View file

File diff suppressed because one or more lines are too long

2
swarm/network/stream/testing/snapshot_256.json
View file

File diff suppressed because one or more lines are too long

2
swarm/network/stream/testing/snapshot_32.json
View file

File diff suppressed because one or more lines are too long

2
swarm/network/stream/testing/snapshot_64.json
View file

File diff suppressed because one or more lines are too long