swarm/network/stream: start of TestIntervals implementation

swarm/network/stream/common_test.go

@@ -17,18 +17,24 @@
 package stream
 
 import (
+	"context"
 	"errors"
 	"flag"
+	"fmt"
+	"io"
 	"io/ioutil"
 	"os"
 	"sync/atomic"
 	"testing"
 	"time"
 
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/node"
+	"github.com/ethereum/go-ethereum/p2p/discover"
 	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
 	p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
+	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/ethereum/go-ethereum/swarm/network"
 	"github.com/ethereum/go-ethereum/swarm/network/stream/intervals"
 	"github.com/ethereum/go-ethereum/swarm/storage"
@@ -46,7 +52,8 @@ var (
 )
 
 var services = adapters.Services{
-	"streamer": NewStreamerService,
+	"streamer":          NewStreamerService,
+	"intervalsStreamer": newIntervalsStreamerService,
 }
 
 func init() {
@@ -75,7 +82,7 @@ func NewStreamerService(ctx *adapters.ServiceContext) (node.Service, error) {
 	go func() {
 		waitPeerErrC <- waitForPeers(r, 1*time.Second, peerCount(id))
 	}()
-	return r, nil
+	return &TestRegistry{Registry: r}, nil
 }
 
 func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *storage.LocalStore, func(), error) {
@@ -155,3 +162,142 @@ func (rrs *roundRobinStore) Close() {
 		store.Close()
 	}
 }
+
+type TestRegistry struct {
+	*Registry
+}
+
+func (r *TestRegistry) APIs() []rpc.API {
+	a := r.Registry.APIs()
+	a = append(a, rpc.API{
+		Namespace: "stream",
+		Version:   "0.1",
+		Service:   r,
+		Public:    true,
+	})
+	return a
+}
+
+func readAll(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 (r *TestRegistry) ReadAll(hash common.Hash) (int64, error) {
+	return readAll(r.api.dpa, hash[:])
+}
+
+type TestExternalRegistry struct {
+	*Registry
+	hashesChan chan []byte
+}
+
+func (r *TestExternalRegistry) APIs() []rpc.API {
+	a := r.Registry.APIs()
+	a = append(a, rpc.API{
+		Namespace: "stream",
+		Version:   "0.1",
+		Service:   r,
+		Public:    true,
+	})
+	return a
+}
+
+func (r *TestExternalRegistry) GetHashes(ctx context.Context, peerId discover.NodeID, s Stream) (*rpc.Subscription, error) {
+
+	peer := r.getPeer(peerId)
+
+	client, err := peer.getClient(s)
+	if err != nil {
+		return nil, err
+	}
+
+	c := client.Client.(*testExternalClient)
+
+	notifier, supported := rpc.NotifierFromContext(ctx)
+	if !supported {
+		return nil, fmt.Errorf("Subscribe not supported")
+	}
+
+	sub := notifier.CreateSubscription()
+
+	go func() {
+		for {
+			select {
+			case h := <-c.hashes:
+				if err := notifier.Notify(sub.ID, h); err != nil {
+					log.Warn(fmt.Sprintf("rpc sub notifier notify stream %s: %v", s, err))
+				}
+			case err := <-sub.Err():
+				log.Warn(fmt.Sprintf("caught subscription error in stream %s: %v", s, err))
+			case <-notifier.Closed():
+				log.Warn(fmt.Sprintf("rpc sub notifier closed"))
+			}
+		}
+	}()
+
+	return sub, nil
+}
+
+// TODO: merge functionalities of testExternalClient and testExternalServer
+// with testClient and testServer.
+
+type testExternalClient struct {
+	t []byte
+	// wait0     chan bool
+	// batchDone chan bool
+	hashes chan []byte
+}
+
+func newTestExternalClient(t []byte, hashesChan chan []byte) *testExternalClient {
+	return &testExternalClient{
+		t: t,
+		// wait0:     make(chan bool),
+		// batchDone: make(chan bool),
+		hashes: hashesChan,
+	}
+}
+
+func (self *testExternalClient) NeedData(hash []byte) func() {
+	self.hashes <- hash
+	return func() {}
+}
+
+func (self *testExternalClient) BatchDone(Stream, uint64, []byte, []byte) func() (*TakeoverProof, error) {
+	// close(self.batchDone)
+	return nil
+}
+
+func (self *testExternalClient) Close() {}
+
+type testExternalServer struct {
+	t []byte
+}
+
+func newTestExternalServer(t []byte) *testExternalServer {
+	return &testExternalServer{
+		t: t,
+	}
+}
+
+func (self *testExternalServer) SetNextBatch(from uint64, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) {
+	return make([]byte, HashSize), from + 1, to + 1, nil, nil
+}
+
+func (self *testExternalServer) GetData([]byte) ([]byte, error) {
+	return nil, nil
+}
+
+func (self *testExternalServer) Close() {
+}

swarm/network/stream/intervals_test.go

@@ -0,0 +1,188 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package stream
+
+import (
+	"context"
+	crand "crypto/rand"
+	"encoding/binary"
+	"fmt"
+	"io"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/node"
+	"github.com/ethereum/go-ethereum/p2p/discover"
+	"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/intervals"
+	streamTesting "github.com/ethereum/go-ethereum/swarm/network/stream/testing"
+	"github.com/ethereum/go-ethereum/swarm/storage"
+)
+
+var externalStreamName = "externalStream"
+
+func newIntervalsStreamerService(ctx *adapters.ServiceContext) (node.Service, error) {
+	id := ctx.Config.ID
+	addr := toAddr(id)
+	kad := network.NewKademlia(addr.Over(), network.NewKadParams())
+	store := stores[id].(*storage.LocalStore)
+	db := storage.NewDBAPI(store)
+	delivery := NewDelivery(kad, db)
+	deliveries[id] = delivery
+	netStore := storage.NewNetStore(store, nil)
+	hashesChan := make(chan []byte) // this chanel is only for one client, in need for more clients, create a map
+	r := NewRegistry(addr, delivery, netStore, intervals.NewMemStore(), defaultSkipCheck)
+
+	r.RegisterClientFunc(externalStreamName, func(p *Peer, t []byte, live bool) (Client, error) {
+		return newTestExternalClient(t, hashesChan), nil
+	})
+	r.RegisterServerFunc(externalStreamName, func(p *Peer, t []byte, live bool) (Server, error) {
+		return newTestExternalServer(t), nil
+	})
+
+	go func() {
+		waitPeerErrC <- waitForPeers(r, 1*time.Second, peerCount(id))
+	}()
+	return &TestExternalRegistry{r, hashesChan}, nil
+}
+
+func XTestIntervals(t *testing.T) {
+	nodes := 2
+	chunkCount := dataChunkCount
+	skipCheck := false
+
+	defaultSkipCheck = skipCheck
+	toAddr = network.NewAddrFromNodeID
+	conf := &streamTesting.RunConfig{
+		Adapter:   *adapter,
+		NodeCount: nodes,
+		ConnLevel: 1,
+		ToAddr:    toAddr,
+		Services:  services,
+	}
+
+	sim, teardown, err := streamTesting.NewSimulation(conf)
+	defer teardown()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	peerCount = func(id discover.NodeID) int {
+		return 1
+	}
+
+	dpa := storage.NewDPA(sim.Stores[0], storage.NewChunkerParams())
+	dpa.Start()
+	size := chunkCount * chunkSize
+	_, wait, err := dpa.Store(io.LimitReader(crand.Reader, int64(size)), int64(size))
+	wait()
+	defer dpa.Stop()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	errc := make(chan error, 1)
+	waitPeerErrC = make(chan error)
+	quitC := make(chan struct{})
+
+	action := func(ctx context.Context) error {
+		i := 0
+		for err := range waitPeerErrC {
+			if err != nil {
+				return fmt.Errorf("error waiting for peers: %s", err)
+			}
+			i++
+			if i == nodes {
+				break
+			}
+		}
+
+		liveHashesChan := make(chan []byte)
+		historyHashesChan := make(chan []byte)
+		id := sim.IDs[1]
+		err := sim.CallClient(id, func(client *rpc.Client) error {
+			err := streamTesting.WatchDisconnections(id, client, errc, quitC)
+			if err != nil {
+				return err
+			}
+			ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
+			defer cancel()
+			sid := sim.IDs[0]
+			err = client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream(externalStreamName, nil, true), &Range{From: 0, To: 5}, Top)
+
+			if err != nil {
+				return err
+			}
+			// live stream
+			_, err = client.Subscribe(ctx, "stream_getHashes", liveHashesChan, sid, NewStream(externalStreamName, nil, true))
+			if err != nil {
+				return err
+			}
+			// history stream
+			_, err = client.Subscribe(ctx, "stream_getHashes", historyHashesChan, sid, NewStream(externalStreamName, nil, false))
+			return err
+		})
+		if err != nil {
+			return err
+		}
+
+		go func() {
+			for i := uint64(0); i < 5; i++ {
+				h := binary.BigEndian.Uint64(<-historyHashesChan)
+				if h != i {
+					errc <- fmt.Errorf("")
+				}
+			}
+		}()
+		return nil
+	}
+	check := func(ctx context.Context, id discover.NodeID) (bool, error) {
+		select {
+		case err := <-errc:
+			return false, err
+		case <-ctx.Done():
+			return false, ctx.Err()
+		default:
+		}
+		return true, nil
+	}
+
+	conf.Step = &simulations.Step{
+		Action:  action,
+		Trigger: streamTesting.Trigger(10*time.Millisecond, quitC, sim.IDs[0]),
+		Expect: &simulations.Expectation{
+			Nodes: sim.IDs[0:1],
+			Check: check,
+		},
+	}
+	startedAt := time.Now()
+	timeout := 300 * time.Second
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+	result, err := sim.Run(ctx, 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,11 +18,9 @@ package stream
 
 import (
 	"fmt"
-	"io"
 	"math"
 	"sync"
 
-	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/rpc"
 
@@ -455,26 +453,6 @@ func NewAPI(r *Registry, store storage.ChunkStore) *API {
 	}
 }
 
-func readAll(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 common.Hash) (int64, error) {
-	return readAll(api.dpa, hash[:])
-}
-
 func (api *API) SubscribeStream(peerId discover.NodeID, s Stream, history *Range, priority uint8) error {
 	return api.streamer.Subscribe(peerId, s, history, priority)
 }

swarm/network/stream/streamer_test.go

@@ -65,16 +65,6 @@ func newTestClient(t []byte) *testClient {
 	}
 }
 
-type testServer struct {
-	t []byte
-}
-
-func newTestServer(t []byte) *testServer {
-	return &testServer{
-		t: t,
-	}
-}
-
 func (self *testClient) NeedData(hash []byte) func() {
 	self.receivedHashes[string(hash)] = hash
 	if bytes.Equal(hash, hash0[:]) {
@@ -96,6 +86,16 @@ func (self *testClient) BatchDone(Stream, uint64, []byte, []byte) func() (*Takeo
 
 func (self *testClient) Close() {}
 
+type testServer struct {
+	t []byte
+}
+
+func newTestServer(t []byte) *testServer {
+	return &testServer{
+		t: t,
+	}
+}
+
 func (self *testServer) SetNextBatch(from uint64, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) {
 	return make([]byte, HashSize), from + 1, to + 1, nil, nil
 }