swarm/network: simplify stream test code, continue refactor

swarm/network/protocol.go

@@ -253,6 +253,14 @@ type BzzPeer struct {
 	lastActive      time.Time // time is updated whenever mutexes are releasing
 }
 
+func NewBzzTestPeer(p *protocols.Peer, addr *BzzAddr) *BzzPeer {
+	return &BzzPeer{
+		Peer:      p,
+		localAddr: addr,
+		BzzAddr:   NewAddrFromNodeID(p.ID()),
+	}
+}
+
 // Off returns the overlay peer record for offline persistance
 func (p *BzzPeer) Off() OverlayAddr {
 	return p.BzzAddr

swarm/network/stream/common_test.go

@@ -19,15 +19,14 @@ package stream
 import (
 	"errors"
 	"flag"
-	"io"
 	"io/ioutil"
 	"os"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/ethereum/go-ethereum/log"
-	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/node"
-	"github.com/ethereum/go-ethereum/p2p/protocols"
 	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
 	p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
 	"github.com/ethereum/go-ethereum/swarm/network"
@@ -40,8 +39,7 @@ var (
 )
 
 var services = adapters.Services{
-	"delivery": newDeliveryService,
+	"streamer": NewStreamerService,
-	"syncer":   newSyncerService,
 }
 
 func init() {
@@ -53,24 +51,16 @@ func init() {
 	log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(os.Stderr, log.TerminalFormat(false))))
 }
 
-var (
+// newService
-	delivery *Delivery
+func NewStreamerService(ctx *adapters.ServiceContext) (node.Service, error) {
-	fileHash storage.Key
+	id := ctx.Config.ID
-)
+	addr := toAddr(id)
-
+	kad := network.NewKademlia(addr.Over(), network.NewKadParams())
-func mustReadAll(dpa *storage.DPA, hash storage.Key) (int, error) {
+	store := stores[id]
-	r := dpa.Retrieve(fileHash)
+	db := storage.NewDBAPI(store.(*storage.LocalStore))
-	buf := make([]byte, 1024)
+	delivery := NewDelivery(kad, db)
-	var n, total int
+	deliveries[id] = delivery
-	var err error
+	return NewRegistry(addr, delivery, store), nil
-	for (total == 0 || n > 0) && err == nil {
-		n, err = r.ReadAt(buf, int64(total))
-		total += n
-	}
-	if err != nil && err != io.EOF {
-		return total, err
-	}
-	return total, nil
 }
 
 func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *storage.LocalStore, func(), error) {
@@ -94,17 +84,8 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *stora
 
 	db := storage.NewDBAPI(localStore)
 	delivery := NewDelivery(to, db)
-	streamer := NewRegistry(delivery)
+	streamer := NewRegistry(addr, delivery, localStore)
-	run := func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
+	protocolTester := p2ptest.NewProtocolTester(t, network.NewNodeIDFromAddr(addr), 1, streamer.runProtocol)
-		bzzPeer := &network.BzzPeer{
-			Peer:      protocols.NewPeer(p, rw, Spec),
-			localAddr: addr,
-			BzzAddr:   network.NewAddrFromNodeID(p.ID()),
-		}
-		to.On(bzzPeer)
-		return streamer.Run(bzzPeer)
-	}
-	protocolTester := p2ptest.NewProtocolTester(t, network.NewNodeIDFromAddr(addr), 1, run)
 
 	err = waitForPeers(streamer, 1*time.Second)
 	if err != nil {
@@ -128,3 +109,30 @@ func waitForPeers(streamer *Registry, timeout time.Duration) error {
 		}
 	}
 }
+
+type roundRobinStore struct {
+	index  uint32
+	stores []storage.ChunkStore
+}
+
+func newRoundRobinStore(stores ...storage.ChunkStore) *roundRobinStore {
+	return &roundRobinStore{
+		stores: stores,
+	}
+}
+
+func (rrs *roundRobinStore) Get(key storage.Key) (*storage.Chunk, error) {
+	return nil, errors.New("get not well defined on round robin store")
+}
+
+func (rrs *roundRobinStore) Put(chunk *storage.Chunk) {
+	i := atomic.AddUint32(&rrs.index, 1)
+	idx := int(i) % len(rrs.stores)
+	rrs.stores[idx].Put(chunk)
+}
+
+func (rrs *roundRobinStore) Close() {
+	for _, store := range rrs.stores {
+		store.Close()
+	}
+}

swarm/network/stream/delivery_test.go

@@ -26,17 +26,20 @@ import (
 	"time"
 
 	"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/protocols"
 	"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"
+	streamTesting "github.com/ethereum/go-ethereum/swarm/network/stream/testing"
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
+var (
+	deliveries map[discover.NodeID]*Delivery
+	stores     map[discover.NodeID]storage.ChunkStore
+	toAddr     func(discover.NodeID) *network.BzzAddr
+)
+
 func TestStreamerRetrieveRequest(t *testing.T) {
 	tester, streamer, _, teardown, err := newStreamerTester(t)
 	defer teardown()
@@ -81,7 +84,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) {
 	peer := streamer.getPeer(peerID)
 
 	peer.handleSubscribeMsg(&SubscribeMsg{
-		Stream:   retrieveRequestStream,
+		Stream:   swarmChunkServerStreamName,
 		Key:      nil,
 		From:     0,
 		To:       0,
@@ -132,7 +135,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
 	peer := streamer.getPeer(peerID)
 
 	peer.handleSubscribeMsg(&SubscribeMsg{
-		Stream:   retrieveRequestStream,
+		Stream:   swarmChunkServerStreamName,
 		Key:      nil,
 		From:     0,
 		To:       0,
@@ -168,7 +171,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
 					// TODO: why is this 32???
 					To:     32,
 					Key:    []byte{},
-					Stream: retrieveRequestStream,
+					Stream: swarmChunkServerStreamName,
 				},
 				Peer: peerID,
 			},
@@ -221,8 +224,8 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	streamer.RegisterIncomingStreamer("foo", func(p *StreamerPeer, t []byte) (IncomingStreamer, error) {
+	streamer.RegisterClientFunc("foo", func(p *Peer, t []byte) (Client, error) {
-		return &testIncomingStreamer{
+		return &testClient{
 			t: t,
 		}, nil
 	})
@@ -301,47 +304,62 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) {
 }
 
 func TestDeliveryFromNodes(t *testing.T) {
-	testSimulation(t, testDeliveryFromNodes(2, 1, 8100, true))
+	testDeliveryFromNodes(t, 2, 1, 8100, true)
-	testSimulation(t, testDeliveryFromNodes(2, 1, 8100, false))
+	testDeliveryFromNodes(t, 2, 1, 8100, false)
-	testSimulation(t, testDeliveryFromNodes(3, 1, 8100, true))
+	testDeliveryFromNodes(t, 3, 1, 8100, true)
-	testSimulation(t, testDeliveryFromNodes(3, 1, 8100, false))
+	testDeliveryFromNodes(t, 3, 1, 8100, false)
 }
 
-func testDeliveryFromNodes(nodes, conns, size int, skipCheck bool) func(adapter adapters.NodeAdapter) (*simulations.StepResult, error) {
+func testDeliveryFromNodes(t *testing.T, nodes, conns, size int, skipCheck bool) {
-	return func(adapter adapters.NodeAdapter) (*simulations.StepResult, error) {
+	toAddr = network.NewAddrFromNodeID
-		trigger := func(net *simulations.Network) chan discover.NodeID {
+	conf := &streamTesting.RunConfig{
-			triggerC := make(chan discover.NodeID)
+		Adapter:   *adapter,
-			ticker := time.NewTicker(500 * time.Millisecond)
+		NodeCount: nodes,
-			go func() {
+		ConnLevel: conns,
-				defer ticker.Stop()
+		ToAddr:    toAddr,
-				// we are only testing the pivot node (net.Nodes[0])
+		Services:  services,
-				for range ticker.C {
-					triggerC <- net.Nodes[0].ID()
-				}
-			}()
-			return triggerC
 	}
 
-		action := func(net *simulations.Network) func(context.Context) error {
+	sim, teardown, err := streamTesting.NewSimulation(conf)
-			// here we distribute chunks of a random file into localstores of nodes 1 to nodes
+	defer teardown()
-			rrdpa := storage.NewDPA(newRoundRobinStore(testing.LocalStores[1:]...), storage.NewChunkerParams())
-			rrdpa.Start()
-			// create a retriever dpa for the pivot node
-			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 {
-				defer rrdpa.Stop()
-				// upload an actual random file of size size
-				hash, wait, err := rrdpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size))
 	if err != nil {
-					return err
+		t.Fatal(err.Error())
+	}
+	stores = make(map[discover.NodeID]storage.ChunkStore)
+	deliveries = make(map[discover.NodeID]*Delivery)
+	for i, id := range sim.IDs {
+		stores[id] = sim.Stores[i]
+	}
+
+	// here we distribute chunks of a random file into Stores of nodes 1 to nodes
+	rrdpa := storage.NewDPA(newRoundRobinStore(sim.Stores[1:]...), storage.NewChunkerParams())
+	rrdpa.Start()
+	fileHash, wait, err := rrdpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size))
+	defer rrdpa.Stop()
+	if err != nil {
+		t.Fatal(err.Error())
 	}
 	// wait until all chunks stored
 	// TODO: is wait() necessary?
 	wait()
-				// assign the fileHash to a global so that it is available for the check function
+	// each node Subscribes to each other's swarmChunkServerStreamName
-				fileHash = hash
+	// 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(), swarmChunkServerStreamName, nil, 0, 0, Top, true)
+	if err != nil {
+		t.Fatal(err.Error())
+	}
+	// create a retriever dpa for the pivot node
+	delivery := deliveries[sim.IDs[0]]
+	dpacs := storage.NewNetStore(sim.Stores[0].(*storage.LocalStore), func(chunk *storage.Chunk) error { return delivery.RequestFromPeers(chunk.Key[:], skipCheck) })
+	dpa := storage.NewDPA(dpacs, storage.NewChunkerParams())
+	dpa.Start()
+	action := func(context.Context) error {
+		dpa := storage.NewDPA(sim.Stores[0], storage.NewChunkerParams())
+		dpa.Start()
+		// defer dpa.Stop()
+
 		go func() {
 			defer dpa.Stop()
 			log.Debug(fmt.Sprintf("retrieve %v", fileHash))
@@ -353,90 +371,60 @@ func testDeliveryFromNodes(nodes, conns, size int, skipCheck bool) func(adapter
 		}()
 		return nil
 	}
-		}
 
-		check := func(net *simulations.Network, dpa *storage.DPA) func(ctx context.Context, id discover.NodeID) (bool, error) {
+	check := func(ctx context.Context, id discover.NodeID) (bool, error) {
-			return func(ctx context.Context, id discover.NodeID) (bool, error) {
-				if id != net.Nodes[0].ID() {
-					return true, nil
-				}
 		select {
 		case <-ctx.Done():
 			return false, ctx.Err()
 		default:
 		}
 		// try to locally retrieve the file to check if retrieve requests have been successful
-				total, err := mustReadAll(dpa, fileHash)
+		node := sim.Net.GetNode(id)
+		if node == nil {
+			return false, fmt.Errorf("unknown node: %s", id)
+		}
+		client, err := node.Client()
+		if err != nil {
+			return false, fmt.Errorf("error getting node client: %s", err)
+		}
+		var total int64
+		if err := client.Call(&total, "stream_readAll", fileHash); err != nil {
+			return false, fmt.Errorf("error reading all: %s (read %v)", err, total)
+		}
+		// total, err := mustReadAll(dpa, fileHash)
 		log.Debug(fmt.Sprintf("check if %08x is available locally: number of bytes read %v/%v (error: %v)", fileHash, total, size, err))
-				if err != nil || total != size {
+		if err != nil || total != int64(size) {
 			return false, nil
 		}
 		return true, nil
-				// node := net.GetNode(id)
-				// if node == nil {
-				// 	return false, fmt.Errorf("unknown node: %s", id)
-				// }
-				// client, err := node.Client()
-				// if err != nil {
-				// 	return false, fmt.Errorf("error getting node client: %s", err)
-				// }
-				// var response int
-				// if err := client.Call(&response, "test_haslocal", hash); err != nil {
-				// 	return false, fmt.Errorf("error getting bzz_has response: %s", err)
-				// }
-				// log.Debug(fmt.Sprintf("node has: %v\n%v", id, response))
-				// return response == 0, nil
-			}
 	}
 
-		result, err := runSimulation(nodes, conns, "delivery", NewAddrFromNodeID, action, trigger, check, adapter)
+	trigger := make(chan discover.NodeID)
-		if err != nil {
+	ticker := time.NewTicker(500 * time.Millisecond)
-			return nil, fmt.Errorf("Setting up simulation failed: %v", err)
-		}
-		if result.Error != nil {
-			return nil, fmt.Errorf("Simulation failed: %s", result.Error)
-		}
-		return result, err
-	}
-}
-
-// newDeliveryService
-func newDeliveryService(ctx *adapters.ServiceContext) (node.Service, error) {
-	id := ctx.Config.ID
-	addr := NewAddrFromNodeID(id)
-	kad := NewKademlia(addr.Over(), NewKadParams())
-	localStore := testing.LocalStores[testing.NodeCount]
-	db := NewDBAPI(localStore.(*storage.LocalStore))
-	streamer := NewStreamerRegistry(NewDelivery(kad, db))
-	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
-	}
-	testing.NodeCount++
-	return testing.NewTestStreamerService(Spec, makeRunFunc(addr, streamer)), nil
-}
-
-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()),
-		}
-		streamer.delivery.overlay.On(bzzPeer)
-		defer streamer.delivery.overlay.Off(bzzPeer)
 	go func() {
-			// each node Subscribes to each other's retrieveRequestStream
+		defer ticker.Stop()
-			// need to wait till an aynchronous process registers the peers in streamer.peers
+		// we are only testing the pivot node (net.Nodes[0])
-			// that is used by Subscribe
+		for range ticker.C {
-			time.Sleep(1 * time.Second)
+			trigger <- sim.Net.Nodes[0].ID()
-			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)
+
+	conf.Step = &simulations.Step{
+		Action:  action,
+		Trigger: trigger,
+		Expect: &simulations.Expectation{
+			Nodes: sim.IDs[0:1],
+			Check: check,
+		},
 	}
+	startedAt := time.Now()
+	result, err := sim.Run(conf)
+	finishedAt := time.Now()
+	if err != nil {
+		t.Fatalf("Setting up simulation failed: %v", err)
+	}
+	if result.Error != nil {
+		t.Fatalf("Simulation failed: %s", result.Error)
+	}
+	streamTesting.CheckResult(t, result, startedAt, finishedAt)
 }

swarm/network/stream/stream.go

@@ -18,14 +18,17 @@ package stream
 
 import (
 	"fmt"
+	"io"
 	"math"
 	"sync"
 
 	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/rpc"
 
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p/discover"
 	"github.com/ethereum/go-ethereum/p2p/protocols"
+	"github.com/ethereum/go-ethereum/swarm/network"
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
@@ -41,6 +44,7 @@ const (
 
 // Registry registry for outgoing and incoming streamer constructors
 type Registry struct {
+	addr        *network.BzzAddr
 	clientMu    sync.RWMutex
 	serverMu    sync.RWMutex
 	peersMu     sync.RWMutex
@@ -48,11 +52,14 @@ type Registry struct {
 	clientFuncs map[string]func(*Peer, []byte) (Client, error)
 	peers       map[discover.NodeID]*Peer
 	delivery    *Delivery
+	store       storage.ChunkStore
 }
 
 // NewRegistry is Streamer constructor
-func NewRegistry(delivery *Delivery) *Registry {
+func NewRegistry(addr *network.BzzAddr, delivery *Delivery, store storage.ChunkStore) *Registry {
 	streamer := &Registry{
+		addr:        addr,
+		store:       store,
 		serverFuncs: make(map[string]func(*Peer, []byte) (Server, error)),
 		clientFuncs: make(map[string]func(*Peer, []byte) (Client, error)),
 		peers:       make(map[discover.NodeID]*Peer),
@@ -175,17 +182,22 @@ func (r *Registry) deletePeer(peer *Peer) {
 }
 
 // Run protocol run function
-func (r *Registry) Run(p *protocols.Peer) error {
+func (r *Registry) run(p *protocols.Peer) error {
 	sp := NewPeer(p, r)
-	// load saved intervals
-
 	r.setPeer(sp)
-
 	defer r.deletePeer(sp)
 	defer close(sp.quit)
 	return sp.Run(sp.HandleMsg)
 }
 
+func (r *Registry) runProtocol(p *p2p.Peer, rw p2p.MsgReadWriter) error {
+	peer := protocols.NewPeer(p, rw, Spec)
+	bzzPeer := network.NewBzzTestPeer(peer, r.addr)
+	r.delivery.overlay.On(bzzPeer)
+	defer r.delivery.overlay.Off(bzzPeer)
+	return r.run(peer)
+}
+
 // HandleMsg is the message handler that delegates incoming messages
 func (p *Peer) HandleMsg(msg interface{}) error {
 	switch msg := msg.(type) {
@@ -305,12 +317,65 @@ func (r *Registry) Protocols() []p2p.Protocol {
 			Name:    Spec.Name,
 			Version: Spec.Version,
 			Length:  Spec.Length(),
-			Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
+			Run:     r.runProtocol,
-				peer := protocols.NewPeer(p, rw, Spec)
+			// NodeInfo: ,
-				return r.Run(peer)
+			// PeerInfo: ,
-			},
-			NodeInfo: r.NodeInfo,
-			PeerInfo: r.PeerInfo,
 		},
 	}
 }
+
+func (r *Registry) APIs() []rpc.API {
+	return []rpc.API{
+		{
+			Namespace: "stream",
+			Version:   "0.1",
+			Service:   NewAPI(r, r.store),
+			Public:    true,
+		},
+	}
+}
+
+func (r *Registry) Start(server *p2p.Server) error {
+	return nil
+}
+
+func (r *Registry) Stop() error {
+	return nil
+}
+
+type API struct {
+	streamer *Registry
+	dpa      *storage.DPA
+}
+
+func NewAPI(r *Registry, store storage.ChunkStore) *API {
+	dpa := storage.NewDPA(store, storage.NewChunkerParams())
+	return &API{
+		streamer: r,
+		dpa:      dpa,
+	}
+}
+
+func mustReadAll(dpa *storage.DPA, hash []byte) (int64, error) {
+	r := dpa.Retrieve(hash)
+	buf := make([]byte, 1024)
+	var n int
+	var total int64
+	var err error
+	for (total == 0 || n > 0) && err == nil {
+		n, err = r.ReadAt(buf, total)
+		total += int64(n)
+	}
+	if err != nil && err != io.EOF {
+		return total, err
+	}
+	return total, nil
+}
+
+func (api *API) ReadAll(hash []byte) (int64, error) {
+	return mustReadAll(api.dpa, hash)
+}
+
+func (api *API) Subscribe(peerId discover.NodeID, s string, t []byte, from, to uint64, priority uint8, live bool) error {
+	return api.streamer.Subscribe(peerId, s, t, from, to, priority, live)
+}

swarm/network/stream/streamer_test.go

@@ -50,15 +50,15 @@ var (
 	batchDone                        = make(chan bool)
 )
 
-type testIncomingStreamer struct {
+type testClient struct {
 	t []byte
 }
 
-type testOutgoingStreamer struct {
+type testServer struct {
 	t []byte
 }
 
-func (self *testIncomingStreamer) NeedData(hash []byte) func() {
+func (self *testClient) NeedData(hash []byte) func() {
 	receivedHashes[string(hash)] = hash
 	if bytes.Equal(hash, hash0[:]) {
 		return func() {
@@ -72,16 +72,16 @@ func (self *testIncomingStreamer) NeedData(hash []byte) func() {
 	return nil
 }
 
-func (self *testIncomingStreamer) BatchDone(string, uint64, []byte, []byte) func() (*TakeoverProof, error) {
+func (self *testClient) BatchDone(string, uint64, []byte, []byte) func() (*TakeoverProof, error) {
 	close(batchDone)
 	return nil
 }
 
-func (self *testOutgoingStreamer) SetNextBatch(from uint64, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) {
+func (self *testServer) SetNextBatch(from uint64, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) {
 	return make([]byte, HashSize), from + 1, to + 1, nil, nil
 }
 
-func (self *testOutgoingStreamer) GetData([]byte) []byte {
+func (self *testServer) GetData([]byte) []byte {
 	return nil
 }
 
@@ -92,8 +92,8 @@ func TestStreamerDownstreamSubscribeMsgExchange(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	streamer.RegisterClientFunc("foo", func(p *StreamerPeer, t []byte) (IncomingStreamer, error) {
+	streamer.RegisterClientFunc("foo", func(p *Peer, t []byte) (Client, error) {
-		return &testIncomingStreamer{
+		return &testClient{
 			t: t,
 		}, nil
 	})
@@ -134,8 +134,8 @@ func TestStreamerUpstreamSubscribeMsgExchange(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	streamer.RegisterServerFunc("foo", func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) {
+	streamer.RegisterServerFunc("foo", func(p *Peer, t []byte) (Server, error) {
-		return &testOutgoingStreamer{
+		return &testServer{
 			t: t,
 		}, nil
 	})
@@ -188,8 +188,8 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	streamer.RegisterClientFunc("foo", func(p *StreamerPeer, t []byte) (IncomingStreamer, error) {
+	streamer.RegisterClientFunc("foo", func(p *Peer, t []byte) (Client, error) {
-		return &testIncomingStreamer{
+		return &testClient{
 			t: t,
 		}, nil
 	})

swarm/network/stream/syncer_test.go

@@ -26,62 +26,94 @@ import (
 	"time"
 
 	"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/protocols"
 	"github.com/ethereum/go-ethereum/p2p/simulations"
-	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
 	"github.com/ethereum/go-ethereum/swarm/network"
+	streamTesting "github.com/ethereum/go-ethereum/swarm/network/stream/testing"
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
 func TestSyncerSimulation(t *testing.T) {
-	testSimulation(t, testSyncBetweenNodes(2, 1, 81000, true, 1))
+	testSyncBetweenNodes(t, 2, 1, 81000, true, 1)
-	testSimulation(t, testSyncBetweenNodes(3, 1, 81000, true, 1))
+	testSyncBetweenNodes(t, 3, 1, 81000, true, 1)
 }
 
-func testSyncBetweenNodes(nodes, conns, size int, skipCheck bool, po uint8) func(adapter adapters.NodeAdapter) (*simulations.StepResult, error) {
+func testSyncBetweenNodes(t *testing.T, nodes, conns, size int, skipCheck bool, po uint8) {
-	return func(adapter adapters.NodeAdapter) (*simulations.StepResult, error) {
+	toAddr = func(id discover.NodeID) *network.BzzAddr {
-		trigger := func(net *simulations.Network) chan discover.NodeID {
+		addr := network.NewAddrFromNodeID(id)
-			triggerC := make(chan discover.NodeID)
+		addr.OAddr[0] = byte(0)
-			ticker := time.NewTicker(500 * time.Millisecond)
+		return addr
-			go func() {
-				defer ticker.Stop()
-				// we are only testing the pivot node (net.Nodes[0])
-				for range ticker.C {
-					triggerC <- net.Nodes[0].ID()
 	}
-			}()
+	conf := &streamTesting.RunConfig{
-			return triggerC
+		Adapter:   *adapter,
+		NodeCount: nodes,
+		ConnLevel: conns,
+		ToAddr:    toAddr,
+		Services:  services,
 	}
 
-		action := func(net *simulations.Network) func(context.Context) error {
+	sim, teardown, err := streamTesting.NewSimulation(conf)
-			// here we distribute chunks of a random file into localstores of nodes 1 to nodes
+	defer teardown()
-			rrdpa := storage.NewDPA(newRoundRobinStore(testing.LocalStores[1:]...), storage.NewChunkerParams())
-			rrdpa.Start()
-			// create a retriever dpa for the pivot node
-			return func(context.Context) error {
-				defer rrdpa.Stop()
-				// upload an actual random file of size size
-				_, wait, err := rrdpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size))
 	if err != nil {
-					return err
+		t.Fatal(err.Error())
+	}
+	stores = make(map[discover.NodeID]storage.ChunkStore)
+	deliveries = make(map[discover.NodeID]*Delivery)
+	log.Warn("Stores", "len", len(sim.Stores))
+	for i, id := range sim.IDs {
+		stores[id] = sim.Stores[i]
+	}
+
+	// here we distribute chunks of a random file into Stores of nodes 1 to nodes
+	rrdpa := storage.NewDPA(newRoundRobinStore(sim.Stores[1:]...), storage.NewChunkerParams())
+	rrdpa.Start()
+	_, wait, err := rrdpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size))
+	defer rrdpa.Stop()
+	if err != nil {
+		t.Fatal(err.Error())
 	}
 	// wait until all chunks stored
+	// TODO: is wait() necessary?
 	wait()
+	// each node Subscribes to each other's swarmChunkServerStreamName
+	// 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(), swarmChunkServerStreamName, nil, 0, 0, Top, true)
+	if err != nil {
+		t.Fatal(err.Error())
+	}
+	// create a retriever dpa for the pivot node
+	action := func(context.Context) error {
+		for i := 0; i < len(sim.IDs)-1; i++ {
+			id := sim.IDs[i]
+			// if err := streamer.Subscribe(p.ID(), "SYNC", []byte{uint8(1)}, 0, 0, Top, false); err != nil {
+			// 	log.Warn("error in subscribe", "err", err)
+			// }
+			node := sim.Net.GetNode(id)
+			if node == nil {
+				return fmt.Errorf("unknown node: %s", id)
+			}
+			client, err := node.Client()
+			if err != nil {
+				return fmt.Errorf("error getting node client: %s", err)
+			}
+			var n int64
+			sid := sim.IDs[i+1]
+			if err := client.Call(&n, "stream_subscribe", sid, "SYNC", []byte{uint8(1)}, 0, 0, Top, false); err != nil {
+				return fmt.Errorf("error subscribing: %s", err)
+			}
+		}
 		return nil
 	}
-		}
 
-		check := func(net *simulations.Network, dpa *storage.DPA) func(ctx context.Context, id discover.NodeID) (bool, error) {
 	dbs := make([]*storage.DBAPI, nodes)
-
 	for i := 0; i < nodes; i++ {
-				dbs[i] = NewDbAccess(testing.LocalStores[i].(*storage.LocalStore))
+		dbs[i] = storage.NewDBAPI(sim.Stores[i].(*storage.LocalStore))
 	}
-			return func(ctx context.Context, id discover.NodeID) (bool, error) {
+
-				if id != net.Nodes[0].ID() {
+	check := func(ctx context.Context, id discover.NodeID) (bool, error) {
+		if id != sim.Net.Nodes[0].ID() {
 			return true, nil
 		}
 		select {
@@ -92,8 +124,9 @@ func testSyncBetweenNodes(nodes, conns, size int, skipCheck bool, po uint8) func
 
 		var found, total int
 		for i := 1; i < nodes; i++ {
-					dbs[i].iterator(0, math.MaxUint64, po, func(key storage.Key, index uint64) bool {
+
-						_, err := dbs[0].get(key)
+			dbs[i].Iterator(0, math.MaxUint64, po, func(key storage.Key, index uint64) bool {
+				_, err := dbs[0].Get(key)
 				if err == nil {
 					found++
 				}
@@ -104,62 +137,33 @@ func testSyncBetweenNodes(nodes, conns, size int, skipCheck bool, po uint8) func
 		log.Debug("sync check", "bin", po, "found", found, "total", total)
 		return found == total, nil
 	}
-		}
-		toAddr := func(id discover.NodeID) *BzzAddr {
-			addr := network.NewAddrFromNodeID(id)
-			addr.OAddr[0] = byte(0)
-			return addr
-		}
 
-		result, err := runSimulation(nodes, conns, "syncer", toAddr, action, trigger, check, adapter)
+	trigger := make(chan discover.NodeID)
-		if err != nil {
+	ticker := time.NewTicker(500 * time.Millisecond)
-			return nil, fmt.Errorf("Setting up simulation failed: %v", err)
-		}
-		if result.Error != nil {
-			return nil, fmt.Errorf("Simulation failed: %s", result.Error)
-		}
-		return result, err
-	}
-}
-
-func newSyncerService(ctx *adapters.ServiceContext) (node.Service, error) {
-	id := ctx.Config.ID
-	addr := network.NewAddrFromNodeID(id)
-	// 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 := testing.LocalStores[testing.NodeCount]
-	db := NewDbAccess(localStore.(*storage.LocalStore))
-	streamer := NewRegistry(NewDelivery(kad, db))
-	RegisterIncomingSyncer(streamer, db)
-	RegisterOutgoingSyncer(streamer, db)
-
-	testing.NodeCount++
-	return testing.NewTestStreamerService(Spec, makeRunFunc(addr, streamer)), nil
-}
-
-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,
-		}
-		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
+		defer ticker.Stop()
-			// need to wait till an aynchronous process registers the peers in streamer.peers
+		// we are only testing the pivot node (net.Nodes[0])
-			// that is used by Subscribe
+		for range ticker.C {
-			time.Sleep(1 * time.Second)
+			trigger <- sim.Net.Nodes[0].ID()
-			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)
+	conf.Step = &simulations.Step{
+		Action:  action,
+		Trigger: trigger,
+		Expect: &simulations.Expectation{
+			Nodes: sim.IDs[0:1],
+			Check: check,
+		},
 	}
+	startedAt := time.Now()
+	result, err := sim.Run(conf)
+	finishedAt := time.Now()
+	if err != nil {
+		t.Fatalf("Setting up simulation failed: %v", err)
+	}
+	if result.Error != nil {
+		t.Fatalf("Simulation failed: %s", result.Error)
+	}
+	streamTesting.CheckResult(t, result, startedAt, finishedAt)
 }

swarm/network/stream/testing/testing.go

@@ -24,84 +24,76 @@ import (
 	"math/rand"
 	"os"
 	"sync"
-	"sync/atomic"
 	"testing"
 	"time"
 
 	"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/storage"
 )
 
-var (
+type Simulation struct {
-	LocalStores []storage.ChunkStore
+	Net    *simulations.Network
+	Stores []storage.ChunkStore
 	Addrs  []network.Addr
-	NodeCount   int
+	IDs    []discover.NodeID
-)
+}
 
-func setLocalStores(addrs ...network.Addr) (func(), error) {
+func SetStores(addrs ...network.Addr) ([]storage.ChunkStore, func(), error) {
 	var datadirs []string
-	LocalStores = make([]storage.ChunkStore, len(addrs))
+	stores := make([]storage.ChunkStore, len(addrs))
 	var err error
 	for i, addr := range addrs {
-		// TODO: remove temp datadir after test
 		var datadir string
 		datadir, err = ioutil.TempDir("", "streamer")
 		if err != nil {
 			break
 		}
-		var localStore *storage.LocalStore
+		var store storage.ChunkStore
-		localStore, err = storage.NewTestLocalStoreForAddr(datadir, addr.Over())
+		store, err = storage.NewTestLocalStoreForAddr(datadir, addr.Over())
 		if err != nil {
 			break
 		}
 		datadirs = append(datadirs, datadir)
-		LocalStores[i] = localStore
+		stores[i] = store
 	}
 	teardown := func() {
 		for _, datadir := range datadirs {
 			os.RemoveAll(datadir)
 		}
 	}
-	return teardown, err
+	return stores, teardown, err
 }
 
-func testSimulation(t *testing.T, services adapters.Services, adapter string, simf func(adapters.NodeAdapter) (*simulations.StepResult, error)) {
+func NewAdapter(adapterType string, services adapters.Services) (adapter adapters.NodeAdapter, teardown func(), err error) {
-	var err error
+	teardown = func() {}
-	var result *simulations.StepResult
+	switch adapterType {
-	startedAt := time.Now()
-
-	switch adapter {
 	case "sim":
-		t.Logf("simadapter")
+		adapter = adapters.NewSimAdapter(services)
-		result, err = simf(adapters.NewSimAdapter(services))
 	case "socket":
-		result, err = simf(adapters.NewSocketAdapter(services))
+		adapter = adapters.NewSocketAdapter(services)
 	case "exec":
 		baseDir, err0 := ioutil.TempDir("", "swarm-test")
 		if err0 != nil {
-			t.Fatal(err0)
+			return nil, teardown, err0
 		}
-		defer os.RemoveAll(baseDir)
+		teardown = func() { os.RemoveAll(baseDir) }
-		result, err = simf(adapters.NewExecAdapter(baseDir))
+		adapter = adapters.NewExecAdapter(baseDir)
 	case "docker":
-		adapter, err0 := adapters.NewDockerAdapter()
+		adapter, err = adapters.NewDockerAdapter()
-		if err0 != nil {
-			t.Fatal(err0)
-		}
-		result, err = simf(adapter)
-	default:
-		t.Fatal("adapter needs to be one of sim, socket, exec, docker")
-	}
 		if err != nil {
-		t.Fatal(err)
+			return nil, teardown, err
 		}
+	default:
+		return nil, teardown, errors.New("adapter needs to be one of sim, socket, exec, docker")
+	}
+	return adapter, teardown, nil
+}
+
+func CheckResult(t *testing.T, result *simulations.StepResult, startedAt, finishedAt time.Time) {
 	t.Logf("Simulation with %d nodes passed in %s", len(result.Passes), result.FinishedAt.Sub(result.StartedAt))
 	var min, max time.Duration
 	var sum int
@@ -116,148 +108,101 @@ func testSimulation(t *testing.T, services adapters.Services, adapter string, si
 		sum += int(duration.Nanoseconds())
 	}
 	t.Logf("Min: %s, Max: %s, Average: %s", min, max, time.Duration(sum/len(result.Passes))*time.Nanosecond)
-	finishedAt := time.Now()
+	t.Logf("Setup: %s, Shutdown: %s", result.StartedAt.Sub(startedAt), finishedAt.Sub(result.FinishedAt))
-	t.Logf("Setup: %s, shutdown: %s", result.StartedAt.Sub(startedAt), finishedAt.Sub(result.FinishedAt))
 }
 
-func runSimulation(nodes, conns int, serviceName string, toAddr func(discover.NodeID) *network.BzzAddr, action func(*simulations.Network) func(context.Context) error, trigger func(*simulations.Network) chan discover.NodeID, check func(*simulations.Network, *storage.DPA) func(context.Context, discover.NodeID) (bool, error), adapter adapters.NodeAdapter) (*simulations.StepResult, error) {
+type RunConfig struct {
+	Adapter   string
+	Step      *simulations.Step
+	NodeCount int
+	ConnLevel int
+	ToAddr    func(discover.NodeID) *network.BzzAddr
+	Services  adapters.Services
+}
+
+func NewSimulation(conf *RunConfig) (*Simulation, func(), error) {
 	// create network
+	nodes := conf.NodeCount
+	adapter, adapterTeardown, err := NewAdapter(conf.Adapter, conf.Services)
+	if err != nil {
+		return nil, adapterTeardown, err
+	}
 	net := simulations.NewNetwork(adapter, &simulations.NetworkConfig{
 		ID:             "0",
-		DefaultService: serviceName,
+		DefaultService: "streamer",
 	})
-	defer net.Shutdown()
+	teardown := func() {
+		adapterTeardown()
+		net.Shutdown()
+	}
 	ids := make([]discover.NodeID, nodes)
-	NodeCount = 0
+	addrs := make([]network.Addr, nodes)
-	Addrs = make([]network.Addr, nodes)
 	// start nodes
 	for i := 0; i < nodes; i++ {
 		node, err := net.NewNode()
 		if err != nil {
-			return nil, fmt.Errorf("error creating node: %s", err)
+			return nil, teardown, fmt.Errorf("error creating node: %s", err)
 		}
 		ids[i] = node.ID()
-		Addrs[i] = toAddr(ids[i])
+		addrs[i] = conf.ToAddr(ids[i])
+	}
+	// set nodes number of Stores available
+	stores, storeTeardown, err := SetStores(addrs...)
+	teardown = func() {
+		storeTeardown()
+		adapterTeardown()
+		net.Shutdown()
 	}
-	// set nodes number of localstores globally available
-	teardown, err := setLocalStores(Addrs...)
-	defer teardown()
 	if err != nil {
-		return nil, err
+		return nil, teardown, err
+	}
+	s := &Simulation{
+		Net:    net,
+		Stores: stores,
+		IDs:    ids,
+		Addrs:  addrs,
+	}
+	return s, teardown, nil
 }
 
+func (s *Simulation) Run(conf *RunConfig) (*simulations.StepResult, error) {
+	// bring up nodes, launch the servive
+	nodes := conf.NodeCount
+	conns := conf.ConnLevel
 	for i := 0; i < nodes; i++ {
-		if err := net.Start(ids[i]); err != nil {
+		if err := s.Net.Start(s.IDs[i]); err != nil {
-			return nil, fmt.Errorf("error starting node %s: %s", ids[i].TerminalString(), err)
+			return nil, fmt.Errorf("error starting node %s: %s", s.IDs[i].TerminalString(), err)
 		}
 	}
-
 	// run a simulation which connects the 10 nodes in a chain
 	wg := sync.WaitGroup{}
-	for i := range ids {
+	for i := range s.IDs {
 		// collect the overlay addresses, to
 		for j := 0; j < conns; j++ {
 			var k int
 			if j == 0 {
 				k = i - 1
 			} else {
-				k = rand.Intn(len(ids))
+				k = rand.Intn(len(s.IDs))
 			}
 			if i > 0 {
 				wg.Add(1)
 				go func(i, k int) {
 					defer wg.Done()
-					net.Connect(ids[i], ids[k])
+					s.Net.Connect(s.IDs[i], s.IDs[k])
 				}(i, k)
 			}
 		}
 	}
 	wg.Wait()
 
-	log.Debug(fmt.Sprintf("nodes: %v", len(Addrs)))
+	log.Debug(fmt.Sprintf("nodes: %v", len(s.Addrs)))
 
 	// create an only locally retrieving dpa for the pivot node to test
 	// if retriee requests have arrived
-	dpa := storage.NewDPA(LocalStores[0], storage.NewChunkerParams())
-	dpa.Start()
-	defer dpa.Stop()
 	timeout := 300 * time.Second
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()
-	result := simulations.NewSimulation(net).Run(ctx, &simulations.Step{
+	result := simulations.NewSimulation(s.Net).Run(ctx, conf.Step)
-		Action:  action(net),
-		Trigger: trigger(net),
-		Expect: &simulations.Expectation{
-			Nodes: ids[0:1],
-			Check: check(net, dpa),
-		},
-	})
 	return result, nil
 }
-
-type roundRobinStore struct {
-	index  uint32
-	stores []storage.ChunkStore
-}
-
-func newRoundRobinStore(stores ...storage.ChunkStore) *roundRobinStore {
-	return &roundRobinStore{
-		stores: stores,
-	}
-}
-
-func (rrs *roundRobinStore) Get(key storage.Key) (*storage.Chunk, error) {
-	return nil, errors.New("get not well defined on round robin store")
-}
-
-func (rrs *roundRobinStore) Put(chunk *storage.Chunk) {
-	i := atomic.AddUint32(&rrs.index, 1)
-	idx := int(i) % len(rrs.stores)
-	rrs.stores[idx].Put(chunk)
-}
-
-func (rrs *roundRobinStore) Close() {
-	for _, store := range rrs.stores {
-		store.Close()
-	}
-}
-
-type TestStreamerService struct {
-	// index    int
-	// addr     *network.BzzAddr
-	// // streamer *stream.Registry
-	run  func(p *p2p.Peer, rw p2p.MsgReadWriter) error
-	spec *protocols.Spec
-}
-
-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:    tds.spec.Name,
-			Version: tds.spec.Version,
-			Length:  tds.spec.Length(),
-			Run:     tds.run,
-			// NodeInfo: ,
-			// PeerInfo: ,
-		},
-	}
-}
-
-func (b *TestStreamerService) APIs() []rpc.API {
-	return []rpc.API{}
-}
-
-func (b *TestStreamerService) Start(server *p2p.Server) error {
-	return nil
-}
-
-func (b *TestStreamerService) Stop() error {
-	return nil
-}

swarm/swarm.go

@@ -132,7 +132,7 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, ensClient *e
 
 	db := storage.NewDBAPI(self.lstore)
 	delivery := stream.NewDelivery(to, db)
-	self.streamer = stream.NewRegistry(delivery)
+	self.streamer = stream.NewRegistry(addr, delivery)
 	stream.RegisterSwarmSyncerServer(self.streamer, db)
 	stream.RegisterSwarmSyncerClient(self.streamer, db)
 
@@ -289,7 +289,7 @@ func (self *Swarm) RegisterPssProtocol(spec *protocols.Spec, targetprotocol *p2p
 }
 
 // implements node.Service
-// Apis returns the RPC Api descriptors the Swarm implementation offers
+// APIs returns the RPC Api descriptors the Swarm implementation offers
 func (self *Swarm) APIs() []rpc.API {
 
 	apis := []rpc.API{