From f0f62218a35cfa350f5552ccab800337750c9f7c Mon Sep 17 00:00:00 2001 From: Balint Gabor Date: Thu, 18 Jan 2018 18:47:43 +0100 Subject: [PATCH] swarm/network, swarm/storage: Further refactor fixes --- swarm/network/stream/common_test.go | 6 +-- swarm/network/stream/delivery_test.go | 56 +++++++++++------------ swarm/network/stream/syncer_test.go | 60 ++++++++++++------------- swarm/network/stream/testing/testing.go | 25 ++++++----- swarm/storage/{dbaccess.go => dbapi.go} | 0 5 files changed, 72 insertions(+), 75 deletions(-) rename swarm/storage/{dbaccess.go => dbapi.go} (100%) diff --git a/swarm/network/stream/common_test.go b/swarm/network/stream/common_test.go index c46cc47144..4b9bc86d8a 100644 --- a/swarm/network/stream/common_test.go +++ b/swarm/network/stream/common_test.go @@ -96,13 +96,13 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *stora delivery := NewDelivery(to, db) streamer := NewRegistry(delivery) run := func(p *p2p.Peer, rw p2p.MsgReadWriter) error { - BzzPeer := &BzzPeer{ + bzzPeer := &network.BzzPeer{ Peer: protocols.NewPeer(p, rw, Spec), localAddr: addr, BzzAddr: network.NewAddrFromNodeID(p.ID()), } - to.On(BzzPeer) - return streamer.Run(BzzPeer) + to.On(bzzPeer) + return streamer.Run(bzzPeer) } protocolTester := p2ptest.NewProtocolTester(t, network.NewNodeIDFromAddr(addr), 1, run) diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index 317132be61..f19cbb9a4d 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -33,6 +33,7 @@ import ( "github.com/ethereum/go-ethereum/p2p/simulations" "github.com/ethereum/go-ethereum/p2p/simulations/adapters" p2ptest "github.com/ethereum/go-ethereum/p2p/testing" + "github.com/ethereum/go-ethereum/swarm/network" "github.com/ethereum/go-ethereum/swarm/storage" ) @@ -323,10 +324,10 @@ func testDeliveryFromNodes(nodes, conns, size int, skipCheck bool) func(adapter action := func(net *simulations.Network) func(context.Context) error { // here we distribute chunks of a random file into localstores of nodes 1 to nodes - rrdpa := storage.NewDPA(newRoundRobinStore(localStores[1:]...), storage.NewChunkerParams()) + rrdpa := storage.NewDPA(newRoundRobinStore(testing.LocalStores[1:]...), storage.NewChunkerParams()) rrdpa.Start() // create a retriever dpa for the pivot node - dpacs := storage.NewNetStore(localStores[0].(*storage.LocalStore), func(chunk *storage.Chunk) error { return delivery.RequestFromPeers(chunk.Key[:], skipCheck) }) + dpacs := storage.NewNetStore(testing.LocalStores[0].(*storage.LocalStore), func(chunk *storage.Chunk) error { return delivery.RequestFromPeers(chunk.Key[:], skipCheck) }) dpa := storage.NewDPA(dpacs, storage.NewChunkerParams()) dpa.Start() return func(context.Context) error { @@ -404,41 +405,38 @@ func newDeliveryService(ctx *adapters.ServiceContext) (node.Service, error) { id := ctx.Config.ID addr := NewAddrFromNodeID(id) kad := NewKademlia(addr.Over(), NewKadParams()) - localStore := localStores[nodeCount] + localStore := testing.LocalStores[testing.NodeCount] db := NewDBAPI(localStore.(*storage.LocalStore)) streamer := NewStreamerRegistry(NewDelivery(kad, db)) - if nodeCount == 0 { + if testing.NodeCount == 0 { // the delivery service for the pivot node is assigned globally // so that the simulation action call can use it for the // swarm enabled dpa delivery = streamer.delivery } - self := &testStreamerService{ - addr: addr, - streamer: streamer, - } - self.run = self.runDelivery - nodeCount++ - return self, nil + testing.NodeCount++ + return testing.NewTestStreamerService(Spec, makeRunFunc(addr, streamer)), nil } -func (b *testStreamerService) runDelivery(p *p2p.Peer, rw p2p.MsgReadWriter) error { - BzzPeer := &BzzPeer{ - Peer: protocols.NewPeer(p, rw, StreamerSpec), - localAddr: b.addr, - BzzAddr: NewAddrFromNodeID(p.ID()), - } - b.streamer.delivery.overlay.On(BzzPeer) - defer b.streamer.delivery.overlay.Off(BzzPeer) - go func() { - // each node Subscribes to each other's retrieveRequestStream - // need to wait till an aynchronous process registers the peers in streamer.peers - // that is used by Subscribe - time.Sleep(1 * time.Second) - err := b.streamer.Subscribe(p.ID(), retrieveRequestStream, nil, 0, 0, Top, true) - if err != nil { - log.Warn("error in subscribe", "err", err) +func makeRunFunc(addr network.Addr, streamer *Registry) (func(p *p2p.Peer, rw p2p.MsgReadWriter), error) { + return func(p *p2p.Peer, rw p2p.MsgReadWriter) error { + bzzPeer := &network.BzzPeer{ + Peer: protocols.NewPeer(p, rw, Spec), + localAddr: addr, + BzzAddr: NewAddrFromNodeID(p.ID()), } - }() - return b.streamer.Run(BzzPeer) + streamer.delivery.overlay.On(bzzPeer) + defer streamer.delivery.overlay.Off(bzzPeer) + go func() { + // each node Subscribes to each other's retrieveRequestStream + // need to wait till an aynchronous process registers the peers in streamer.peers + // that is used by Subscribe + time.Sleep(1 * time.Second) + err := streamer.Subscribe(p.ID(), retrieveRequestStream, nil, 0, 0, Top, true) + if err != nil { + log.Warn("error in subscribe", "err", err) + } + }() + return streamer.Run(bzzPeer) + } } diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index c1a9bd6fac..a6e0ab6b1a 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -58,7 +58,7 @@ func testSyncBetweenNodes(nodes, conns, size int, skipCheck bool, po uint8) func action := func(net *simulations.Network) func(context.Context) error { // here we distribute chunks of a random file into localstores of nodes 1 to nodes - rrdpa := storage.NewDPA(newRoundRobinStore(localStores[1:]...), storage.NewChunkerParams()) + rrdpa := storage.NewDPA(newRoundRobinStore(testing.LocalStores[1:]...), storage.NewChunkerParams()) rrdpa.Start() // create a retriever dpa for the pivot node return func(context.Context) error { @@ -78,7 +78,7 @@ func testSyncBetweenNodes(nodes, conns, size int, skipCheck bool, po uint8) func dbs := make([]*storage.DBAPI, nodes) for i := 0; i < nodes; i++ { - dbs[i] = NewDbAccess(localStores[i].(*storage.LocalStore)) + dbs[i] = NewDbAccess(testing.LocalStores[i].(*storage.LocalStore)) } return func(ctx context.Context, id discover.NodeID) (bool, error) { if id != net.Nodes[0].ID() { @@ -128,42 +128,38 @@ func newSyncerService(ctx *adapters.ServiceContext) (node.Service, error) { // for the test we make all peers share 8 bits so that syncing full bins make sense addr.OAddr[0] = byte(0) kad := NewKademlia(addr.Over(), NewKadParams()) - localStore := localStores[nodeCount] + localStore := testing.LocalStores[testing.NodeCount] db := NewDbAccess(localStore.(*storage.LocalStore)) streamer := NewRegistry(NewDelivery(kad, db)) RegisterIncomingSyncer(streamer, db) RegisterOutgoingSyncer(streamer, db) - self := &testStreamerService{ - index: nodeCount, - addr: addr, - streamer: streamer, - } - self.run = self.runSyncer - nodeCount++ - return self, nil + testing.NodeCount++ + return testing.NewTestStreamerService(Spec, makeRunFunc(addr, streamer)), nil } -func (b *testStreamerService) runSyncer(p *p2p.Peer, rw p2p.MsgReadWriter) error { - addr := network.NewAddrFromNodeID(p.ID()) - addr.OAddr[0] = byte(0) - BzzPeer := &BzzPeer{ - Peer: protocols.NewPeer(p, rw, Spec), - localAddr: b.addr, - BzzAddr: addr, - } - b.streamer.delivery.overlay.On(BzzPeer) - defer b.streamer.delivery.overlay.Off(BzzPeer) - // if len(addr) > b.index+1 && bytes.Equal(addrs[b.index+1], addr) { - go func() { - // each node Subscribes to each other's retrieveRequestStream - // need to wait till an aynchronous process registers the peers in streamer.peers - // that is used by Subscribe - time.Sleep(1 * time.Second) - if err := b.streamer.Subscribe(p.ID(), "SYNC", []byte{uint8(1)}, 0, 0, Top, false); err != nil { - log.Warn("error in subscribe", "err", err) +func makeRunFunc(localAddr network.Addr, streamer *Registry) (func(p *p2p.Peer, rw p2p.MsgReadWriter), error) { + return func(p *p2p.Peer, rw p2p.MsgReadWriter) error { + remoteAddr := network.NewAddrFromNodeID(p.ID()) + remoteAddr.OAddr[0] = byte(0) + bzzPeer := &network.BzzPeer{ + Peer: protocols.NewPeer(p, rw, Spec), + localAddr: localAddr, + BzzAddr: remoteAddr, } - }() - // } - return b.streamer.Run(BzzPeer) + streamer.delivery.overlay.On(bzzPeer) + defer streamer.delivery.overlay.Off(bzzPeer) + // if len(addr) > b.index+1 && bytes.Equal(testing.Addrs[b.index+1], addr) { + go func() { + // each node Subscribes to each other's retrieveRequestStream + // need to wait till an aynchronous process registers the peers in streamer.peers + // that is used by Subscribe + time.Sleep(1 * time.Second) + if err := streamer.Subscribe(p.ID(), "SYNC", []byte{uint8(1)}, 0, 0, Top, false); err != nil { + log.Warn("error in subscribe", "err", err) + } + }() + // } + return streamer.Run(bzzPeer) + } } diff --git a/swarm/network/stream/testing/testing.go b/swarm/network/stream/testing/testing.go index e0da4d0336..d3a78ecde2 100644 --- a/swarm/network/stream/testing/testing.go +++ b/swarm/network/stream/testing/testing.go @@ -31,11 +31,11 @@ import ( "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/p2p/protocols" "github.com/ethereum/go-ethereum/p2p/simulations" "github.com/ethereum/go-ethereum/p2p/simulations/adapters" "github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/swarm/network" - "github.com/ethereum/go-ethereum/swarm/network/stream" "github.com/ethereum/go-ethereum/swarm/storage" ) @@ -223,23 +223,26 @@ func (rrs *roundRobinStore) Close() { } type TestStreamerService struct { - index int - addr *network.BzzAddr - streamer *stream.Registry - run func(s *TestStreamerService, p *p2p.Peer, rw p2p.MsgReadWriter) error + // index int + // addr *network.BzzAddr + // // streamer *stream.Registry + run func(p *p2p.Peer, rw p2p.MsgReadWriter) error + spec *protocols.Spec } -func NewTestStreamerService(run func(s *TestStreamerService, p *p2p.Peer, rw p2p.MsgReadWriter) error) TestStreamerService { - t := &TestStreamerService{} - t.run = run +func NewTestStreamerService(spec *protocols.Spec, run func(p *p2p.Peer, rw p2p.MsgReadWriter) error) *TestStreamerService { + return &TestStreamerService{ + run: run, + spec: spec, + } } func (tds *TestStreamerService) Protocols() []p2p.Protocol { return []p2p.Protocol{ { - Name: stream.Spec.Name, - Version: stream.Spec.Version, - Length: stream.Spec.Length(), + Name: tds.spec.Name, + Version: tds.spec.Version, + Length: tds.spec.Length(), Run: tds.run, // NodeInfo: , // PeerInfo: , diff --git a/swarm/storage/dbaccess.go b/swarm/storage/dbapi.go similarity index 100% rename from swarm/storage/dbaccess.go rename to swarm/storage/dbapi.go