Merge pull request #275 from ethersphere/swarm-network-rewrite-syncer-intervals

Swarm network rewrite syncer intervals

swarm/network/light/lightnode.go

@@ -59,7 +59,7 @@ func (r *RemoteSectionReader) NeedData(key []byte) func() {
 	}
 }
 
-func (r *RemoteSectionReader) BatchDone(s string, from uint64, hashes []byte, root []byte) func() (*stream.TakeoverProof, error) {
+func (r *RemoteSectionReader) BatchDone(s stream.Stream, from uint64, hashes []byte, root []byte) func() (*stream.TakeoverProof, error) {
 	r.hashes <- hashes
 	return nil
 }
@@ -158,7 +158,7 @@ func (s *RemoteSectionServer) Close() {}
 
 // RegisterRemoteSectionReader registers RemoteSectionReader on light downstream node
 func RegisterRemoteSectionReader(s *stream.Registry, db *storage.DBAPI) {
-	s.RegisterClientFunc("REMOTE_SECTION", func(p *stream.Peer, t []byte) (stream.Client, error) {
+	s.RegisterClientFunc("REMOTE_SECTION", func(p *stream.Peer, t []byte, live bool) (stream.Client, error) {
 		return NewRemoteSectionReader(t, db), nil
 	})
 }
@@ -166,7 +166,7 @@ func RegisterRemoteSectionReader(s *stream.Registry, db *storage.DBAPI) {
 // RegisterRemoteSectionServer registers RemoteSectionServer outgoing streamer on
 // upstream light server node
 func RegisterRemoteSectionServer(s *stream.Registry, db *storage.DBAPI, rf func([]byte) *storage.LazyChunkReader) {
-	s.RegisterServerFunc("REMOTE_SECTION", func(p *stream.Peer, t []byte) (stream.Server, error) {
+	s.RegisterServerFunc("REMOTE_SECTION", func(p *stream.Peer, t []byte, live bool) (stream.Server, error) {
 		r := rf(t)
 		return NewRemoteSectionServer(db, r), nil
 	})

swarm/network/stream/common_test.go

@@ -17,19 +17,27 @@
 package stream
 
 import (
+	"context"
+	"encoding/binary"
 	"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,6 +54,7 @@ var (
 
 var services = adapters.Services{
 	"streamer":          NewStreamerService,
+	"intervalsStreamer": newIntervalsStreamerService,
 }
 
 func init() {
@@ -68,13 +77,13 @@ func NewStreamerService(ctx *adapters.ServiceContext) (node.Service, error) {
 	delivery := NewDelivery(kad, db)
 	deliveries[id] = delivery
 	netStore := storage.NewNetStore(store, nil)
-	r := NewRegistry(addr, delivery, netStore, defaultSkipCheck)
+	r := NewRegistry(addr, delivery, netStore, intervals.NewMemStore(), defaultSkipCheck)
 	RegisterSwarmSyncerServer(r, db)
 	RegisterSwarmSyncerClient(r, db)
 	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) {
@@ -87,18 +96,22 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *stora
 	if err != nil {
 		return nil, nil, nil, func() {}, err
 	}
-	teardown := func() {
+	removeDataDir := func() {
 		os.RemoveAll(datadir)
 	}
 
 	localStore, err := storage.NewTestLocalStoreForAddr(datadir, addr.Over())
 	if err != nil {
-		return nil, nil, nil, teardown, err
+		return nil, nil, nil, removeDataDir, err
 	}
 
 	db := storage.NewDBAPI(localStore)
 	delivery := NewDelivery(to, db)
-	streamer := NewRegistry(addr, delivery, localStore, defaultSkipCheck)
+	streamer := NewRegistry(addr, delivery, localStore, intervals.NewMemStore(), defaultSkipCheck)
+	teardown := func() {
+		streamer.Close()
+		removeDataDir()
+	}
 	protocolTester := p2ptest.NewProtocolTester(t, network.NewNodeIDFromAddr(addr), 1, streamer.runProtocol)
 
 	err = waitForPeers(streamer, 1*time.Second, 1)
@@ -150,3 +163,202 @@ 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
+}
+
+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(ctx, 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() {
+		// if we begin sending event immediately some events
+		// will probably be dropped since the subscription ID might not be send to
+		// the client.
+		// ref: rpc/subscription_test.go#L65
+		time.Sleep(1 * time.Second)
+		for {
+			select {
+			case h := <-c.hashes:
+				<-c.enableNotificationsC // wait for notification subscription to complete
+				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():
+				if err != nil {
+					log.Warn(fmt.Sprintf("caught subscription error in stream %s: %v", s, err))
+				}
+			case <-notifier.Closed():
+				log.Trace(fmt.Sprintf("rpc sub notifier closed"))
+				return
+			}
+		}
+	}()
+
+	return sub, nil
+}
+
+func (r *TestExternalRegistry) EnableNotifications(peerId discover.NodeID, s Stream) error {
+	peer := r.getPeer(peerId)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	client, err := peer.getClient(ctx, s)
+	if err != nil {
+		return err
+	}
+
+	close(client.Client.(*testExternalClient).enableNotificationsC)
+
+	return 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
+	db                   *storage.DBAPI
+	enableNotificationsC chan struct{}
+}
+
+func newTestExternalClient(t []byte, db *storage.DBAPI) *testExternalClient {
+	return &testExternalClient{
+		t: t,
+		// wait0:     make(chan bool),
+		// batchDone: make(chan bool),
+		hashes:               make(chan []byte),
+		db:                   db,
+		enableNotificationsC: make(chan struct{}),
+	}
+}
+
+func (c *testExternalClient) NeedData(hash []byte) func() {
+	chunk, _ := c.db.GetOrCreateRequest(hash)
+	if chunk.ReqC == nil {
+		return nil
+	}
+	c.hashes <- hash
+	return func() {
+		chunk.WaitToStore()
+	}
+}
+
+func (c *testExternalClient) BatchDone(Stream, uint64, []byte, []byte) func() (*TakeoverProof, error) {
+	return nil
+}
+
+func (c *testExternalClient) Close() {}
+
+const testExternalServerBatchSize = 10
+
+type testExternalServer struct {
+	t         []byte
+	keyFunc   func(key []byte, index uint64)
+	sessionAt uint64
+	maxKeys   uint64
+	streamer  *TestExternalRegistry
+}
+
+func newTestExternalServer(t []byte, sessionAt, maxKeys uint64, keyFunc func(key []byte, index uint64)) *testExternalServer {
+	if keyFunc == nil {
+		keyFunc = binary.BigEndian.PutUint64
+	}
+	return &testExternalServer{
+		t:         t,
+		keyFunc:   keyFunc,
+		sessionAt: sessionAt,
+		maxKeys:   maxKeys,
+	}
+}
+
+func (s *testExternalServer) SetNextBatch(from uint64, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) {
+	if from == 0 && to == 0 {
+		from = s.sessionAt
+		to = s.sessionAt + testExternalServerBatchSize
+	}
+	if to-from > testExternalServerBatchSize {
+		to = from + testExternalServerBatchSize - 1
+	}
+	if from >= s.maxKeys && to > s.maxKeys {
+		return nil, 0, 0, nil, io.EOF
+	}
+	if to > s.maxKeys {
+		to = s.maxKeys
+	}
+	b := make([]byte, HashSize*(to-from+1))
+	for i := from; i <= to; i++ {
+		s.keyFunc(b[(i-from)*HashSize:(i-from+1)*HashSize], i)
+	}
+	return b, from, to, nil, nil
+}
+
+func (s *testExternalServer) GetData([]byte) ([]byte, error) {
+	return make([]byte, 4096), nil
+}
+
+func (s *testExternalServer) Close() {}

swarm/network/stream/delivery.go

@@ -129,7 +129,7 @@ type RetrieveRequestMsg struct {
 
 func (d *Delivery) handleRetrieveRequestMsg(sp *Peer, req *RetrieveRequestMsg) error {
 	log.Debug("received request", "peer", sp.ID(), "hash", req.Key)
-	s, err := sp.getServer(swarmChunkServerStreamName)
+	s, err := sp.getServer(NewStream(swarmChunkServerStreamName, nil, false))
 	if err != nil {
 		return err
 	}

swarm/network/stream/delivery_test.go

@@ -87,10 +87,11 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) {
 	peer := streamer.getPeer(peerID)
 
 	peer.handleSubscribeMsg(&SubscribeMsg{
-		Stream:   swarmChunkServerStreamName,
+		Stream: NewStream(swarmChunkServerStreamName, nil, false),
-		Key:      nil,
+		History: &Range{
 			From: 0,
 			To:   0,
+		},
 		Priority: Top,
 	})
 
@@ -138,10 +139,11 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
 	peer := streamer.getPeer(peerID)
 
 	peer.handleSubscribeMsg(&SubscribeMsg{
-		Stream:   swarmChunkServerStreamName,
+		Stream: NewStream(swarmChunkServerStreamName, nil, false),
-		Key:      nil,
+		History: &Range{
 			From: 0,
 			To:   0,
+		},
 		Priority: Top,
 	})
 
@@ -173,8 +175,7 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
 					From:   0,
 					// TODO: why is this 32???
 					To:     32,
-					Key:    []byte{},
+					Stream: NewStream(swarmChunkServerStreamName, nil, false),
-					Stream: swarmChunkServerStreamName,
 				},
 				Peer: peerID,
 			},
@@ -227,7 +228,7 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	streamer.RegisterClientFunc("foo", func(p *Peer, t []byte) (Client, error) {
+	streamer.RegisterClientFunc("foo", func(p *Peer, t []byte, live bool) (Client, error) {
 		return &testClient{
 			t: t,
 		}, nil
@@ -235,7 +236,8 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) {
 
 	peerID := tester.IDs[0]
 
-	err = streamer.Subscribe(peerID, "foo", []byte{}, 5, 8, Top, true)
+	stream := NewStream("foo", nil, true)
+	err = streamer.Subscribe(peerID, stream, &Range{From: 5, To: 8}, Top)
 	if err != nil {
 		t.Fatalf("Expected no error, got %v", err)
 	}
@@ -259,10 +261,11 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) {
 			{
 				Code: 4,
 				Msg: &SubscribeMsg{
-					Stream:   "foo",
+					Stream: stream,
-					Key:      []byte{},
+					History: &Range{
 						From: 5,
 						To:   8,
+					},
 					Priority: Top,
 				},
 				Peer: peerID,
@@ -389,7 +392,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 				ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
 				defer cancel()
 				sid := sim.IDs[j+1]
-				return client.CallContext(ctx, nil, "stream_subscribeStream", sid, swarmChunkServerStreamName, nil, 0, 0, Top, false)
+				return client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream(swarmChunkServerStreamName, nil, false), &Range{From: 0, To: 0}, Top)
 			})
 			if err != nil {
 				return err
@@ -563,7 +566,7 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip
 				ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
 				defer cancel()
 				sid := sim.IDs[j+1] // the upstream peer's id
-				return client.CallContext(ctx, nil, "stream_subscribeStream", sid, swarmChunkServerStreamName, nil, 0, 0, Top, false)
+				return client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream(swarmChunkServerStreamName, nil, false), &Range{From: 0, To: 0}, Top)
 			})
 			if err != nil {
 				break

swarm/network/stream/intervals/dbstore.go

@@ -0,0 +1,78 @@
+// 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 intervals
+
+import (
+	"github.com/syndtr/goleveldb/leveldb"
+)
+
+// DBStore uses LevelDB to store intervals.
+type DBStore struct {
+	db *leveldb.DB
+}
+
+// NewDBStore creates a new instance of DBStore.
+func NewDBStore(path string) (s *DBStore, err error) {
+	db, err := leveldb.OpenFile(path, nil)
+	if err != nil {
+		return nil, err
+	}
+	return &DBStore{
+		db: db,
+	}, nil
+}
+
+// Get retrieves Intervals for a specific key. If there is no Intervals
+// ErrNotFound is returned.
+func (s *DBStore) Get(key string) (i *Intervals, err error) {
+	k := []byte(key)
+	has, err := s.db.Has(k, nil)
+	if err != nil {
+		return nil, ErrNotFound
+	}
+	if !has {
+		return nil, ErrNotFound
+	}
+	data, err := s.db.Get(k, nil)
+	if err == leveldb.ErrNotFound {
+		err = ErrNotFound
+	}
+	i = &Intervals{}
+	if err = i.UnmarshalBinary(data); err != nil {
+		return nil, err
+	}
+	return i, err
+}
+
+// Put stores Intervals for a specific key.
+func (s *DBStore) Put(key string, i *Intervals) (err error) {
+	data, err := i.MarshalBinary()
+	if err != nil {
+		return err
+	}
+	return s.db.Put([]byte(key), data, nil)
+}
+
+// Delete removes Intervals stored under a specific key.
+func (s *DBStore) Delete(key string) (err error) {
+	return s.db.Delete([]byte(key), nil)
+}
+
+// Close releases the resources used by the underlying LevelDB.
+func (s *DBStore) Close() error {
+	return s.db.Close()
+}

swarm/network/stream/intervals/dbstore_test.go

@@ -0,0 +1,40 @@
+// 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 intervals
+
+import (
+	"io/ioutil"
+	"os"
+	"testing"
+)
+
+// TestDBStore tests basic functionality of DBStore.
+func TestDBStore(t *testing.T) {
+	dir, err := ioutil.TempDir("", "intervals_test_db_store")
+	if err != nil {
+		panic(err)
+	}
+	defer os.RemoveAll(dir)
+
+	store, err := NewDBStore(dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer store.Close()
+
+	testStore(t, store)
+}

swarm/network/stream/intervals/intervals.go

@@ -0,0 +1,206 @@
+// 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 intervals
+
+import (
+	"bytes"
+	"fmt"
+	"strconv"
+	"sync"
+)
+
+// Intervals store a list of intervals. Its purpose is to provide
+// methods to add new intervals and retrieve missing intervals that
+// need to be added.
+// It may be used in synchronization of streaming data to persist
+// retrieved data ranges between sessions.
+type Intervals struct {
+	start  uint64
+	ranges [][2]uint64
+	mu     sync.RWMutex
+}
+
+// New creates a new instance of Intervals.
+// Start argument limits the lower bound of intervals.
+// No range bellow start bound will be added by Add method or
+// returned by Next method. This limit may be used for
+// tracking "live" synchronization, where the sync session
+// starts from a specific value, and if "live" sync intervals
+// need to be merged with historical ones, it can be safely done.
+func NewIntervals(start uint64) *Intervals {
+	return &Intervals{
+		start: start,
+	}
+}
+
+// Add adds a new range to intervals. Range start and end are values
+// are both inclusive.
+func (i *Intervals) Add(start, end uint64) {
+	i.mu.Lock()
+	defer i.mu.Unlock()
+
+	i.add(start, end)
+}
+
+func (i *Intervals) add(start, end uint64) {
+	if start < i.start {
+		start = i.start
+	}
+	if end < i.start {
+		return
+	}
+	minStartJ := -1
+	maxEndJ := -1
+	j := 0
+	for ; j < len(i.ranges); j++ {
+		if minStartJ < 0 {
+			if (start <= i.ranges[j][0] && end+1 >= i.ranges[j][0]) || (start <= i.ranges[j][1]+1 && end+1 >= i.ranges[j][1]) {
+				if i.ranges[j][0] < start {
+					start = i.ranges[j][0]
+				}
+				minStartJ = j
+			}
+		}
+		if (start <= i.ranges[j][1] && end+1 >= i.ranges[j][1]) || (start <= i.ranges[j][0] && end+1 >= i.ranges[j][0]) {
+			if i.ranges[j][1] > end {
+				end = i.ranges[j][1]
+			}
+			maxEndJ = j
+		}
+		if end+1 <= i.ranges[j][0] {
+			break
+		}
+	}
+	if minStartJ < 0 && maxEndJ < 0 {
+		i.ranges = append(i.ranges[:j], append([][2]uint64{{start, end}}, i.ranges[j:]...)...)
+		return
+	}
+	if minStartJ >= 0 {
+		i.ranges[minStartJ][0] = start
+	}
+	if maxEndJ >= 0 {
+		i.ranges[maxEndJ][1] = end
+	}
+	if minStartJ >= 0 && maxEndJ >= 0 && minStartJ != maxEndJ {
+		i.ranges[maxEndJ][0] = start
+		i.ranges = append(i.ranges[:minStartJ], i.ranges[maxEndJ:]...)
+	}
+}
+
+// Merge adds all the intervals from the the m Interval to current one.
+func (i *Intervals) Merge(m *Intervals) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	i.mu.Lock()
+	defer i.mu.Unlock()
+
+	for _, r := range m.ranges {
+		i.add(r[0], r[1])
+	}
+}
+
+// Next returns the first range interval that is not fulfilled. Returned
+// start and end values are both inclusive, meaning that the whole range
+// including start and end need to be added in order to full the gap
+// in intervals.
+// Returned value for end is 0 if the next interval is after the whole
+// range that is stored in Intervals. Zero end value represents no limit
+// on the next interval length.
+func (i *Intervals) Next() (start, end uint64) {
+	i.mu.RLock()
+	defer i.mu.RUnlock()
+
+	l := len(i.ranges)
+	if l == 0 {
+		return i.start, 0
+	}
+	if i.ranges[0][0] != i.start {
+		return i.start, i.ranges[0][0] - 1
+	}
+	if l == 1 {
+		return i.ranges[0][1] + 1, 0
+	}
+	return i.ranges[0][1] + 1, i.ranges[1][0] - 1
+}
+
+// Last returns the value that is at the end of the last interval.
+func (i *Intervals) Last() (end uint64) {
+	i.mu.RLock()
+	defer i.mu.RUnlock()
+
+	l := len(i.ranges)
+	if l == 0 {
+		return 0
+	}
+	return i.ranges[l-1][1]
+}
+
+// String returns a descriptive representation of range intervals
+// in [] notation, as a list of two element vectors.
+func (i *Intervals) String() string {
+	return fmt.Sprint(i.ranges)
+}
+
+// MarshalBinary encodes Intervals parameters into a semicolon separated list.
+// The first element in the list is base36-encoded start value. The following
+// elements are two base36-encoded value ranges separated by comma.
+func (i *Intervals) MarshalBinary() (data []byte, err error) {
+	d := make([][]byte, len(i.ranges)+1)
+	d[0] = []byte(strconv.FormatUint(i.start, 36))
+	for j := range i.ranges {
+		r := i.ranges[j]
+		d[j+1] = []byte(strconv.FormatUint(r[0], 36) + "," + strconv.FormatUint(r[1], 36))
+	}
+	return bytes.Join(d, []byte(";")), nil
+}
+
+// UnmarshalBinary decodes data according to the Intervals.MarshalBinary format.
+func (i *Intervals) UnmarshalBinary(data []byte) (err error) {
+	d := bytes.Split(data, []byte(";"))
+	l := len(d)
+	if l == 0 {
+		return nil
+	}
+	if l >= 1 {
+		i.start, err = strconv.ParseUint(string(d[0]), 36, 64)
+		if err != nil {
+			return err
+		}
+	}
+	if l == 1 {
+		return nil
+	}
+
+	i.ranges = make([][2]uint64, 0, l-1)
+	for j := 1; j < l; j++ {
+		r := bytes.SplitN(d[j], []byte(","), 2)
+		if len(r) < 2 {
+			return fmt.Errorf("range %d has less then 2 elements", j)
+		}
+		start, err := strconv.ParseUint(string(r[0]), 36, 64)
+		if err != nil {
+			return fmt.Errorf("parsing the first element in range %d: %v", j, err)
+		}
+		end, err := strconv.ParseUint(string(r[1]), 36, 64)
+		if err != nil {
+			return fmt.Errorf("parsing the second element in range %d: %v", j, err)
+		}
+		i.ranges = append(i.ranges, [2]uint64{start, end})
+	}
+
+	return nil
+}

swarm/network/stream/intervals/intervals_test.go

@@ -0,0 +1,395 @@
+// 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 intervals
+
+import "testing"
+
+// Test tests Interval methods Add, Next and Last for various
+// initial state.
+func Test(t *testing.T) {
+	for i, tc := range []struct {
+		startLimit uint64
+		initial    [][2]uint64
+		start      uint64
+		end        uint64
+		expected   string
+		nextStart  uint64
+		nextEnd    uint64
+		last       uint64
+	}{
+		{
+			initial:   nil,
+			start:     0,
+			end:       0,
+			expected:  "[[0 0]]",
+			nextStart: 1,
+			nextEnd:   0,
+			last:      0,
+		},
+		{
+			initial:   nil,
+			start:     0,
+			end:       10,
+			expected:  "[[0 10]]",
+			nextStart: 11,
+			nextEnd:   0,
+			last:      10,
+		},
+		{
+			initial:   nil,
+			start:     5,
+			end:       15,
+			expected:  "[[5 15]]",
+			nextStart: 0,
+			nextEnd:   4,
+			last:      15,
+		},
+		{
+			initial:   [][2]uint64{{0, 0}},
+			start:     0,
+			end:       0,
+			expected:  "[[0 0]]",
+			nextStart: 1,
+			nextEnd:   0,
+			last:      0,
+		},
+		{
+			initial:   [][2]uint64{{0, 0}},
+			start:     5,
+			end:       15,
+			expected:  "[[0 0] [5 15]]",
+			nextStart: 1,
+			nextEnd:   4,
+			last:      15,
+		},
+		{
+			initial:   [][2]uint64{{5, 15}},
+			start:     5,
+			end:       15,
+			expected:  "[[5 15]]",
+			nextStart: 0,
+			nextEnd:   4,
+			last:      15,
+		},
+		{
+			initial:   [][2]uint64{{5, 15}},
+			start:     5,
+			end:       20,
+			expected:  "[[5 20]]",
+			nextStart: 0,
+			nextEnd:   4,
+			last:      20,
+		},
+		{
+			initial:   [][2]uint64{{5, 15}},
+			start:     10,
+			end:       20,
+			expected:  "[[5 20]]",
+			nextStart: 0,
+			nextEnd:   4,
+			last:      20,
+		},
+		{
+			initial:   [][2]uint64{{5, 15}},
+			start:     0,
+			end:       20,
+			expected:  "[[0 20]]",
+			nextStart: 21,
+			nextEnd:   0,
+			last:      20,
+		},
+		{
+			initial:   [][2]uint64{{5, 15}},
+			start:     2,
+			end:       10,
+			expected:  "[[2 15]]",
+			nextStart: 0,
+			nextEnd:   1,
+			last:      15,
+		},
+		{
+			initial:   [][2]uint64{{5, 15}},
+			start:     2,
+			end:       4,
+			expected:  "[[2 15]]",
+			nextStart: 0,
+			nextEnd:   1,
+			last:      15,
+		},
+		{
+			initial:   [][2]uint64{{5, 15}},
+			start:     2,
+			end:       5,
+			expected:  "[[2 15]]",
+			nextStart: 0,
+			nextEnd:   1,
+			last:      15,
+		},
+		{
+			initial:   [][2]uint64{{5, 15}},
+			start:     2,
+			end:       3,
+			expected:  "[[2 3] [5 15]]",
+			nextStart: 0,
+			nextEnd:   1,
+			last:      15,
+		},
+		{
+			initial:   [][2]uint64{{5, 15}},
+			start:     2,
+			end:       4,
+			expected:  "[[2 15]]",
+			nextStart: 0,
+			nextEnd:   1,
+			last:      15,
+		},
+		{
+			initial:   [][2]uint64{{0, 1}, {5, 15}},
+			start:     2,
+			end:       4,
+			expected:  "[[0 15]]",
+			nextStart: 16,
+			nextEnd:   0,
+			last:      15,
+		},
+		{
+			initial:   [][2]uint64{{0, 5}, {15, 20}},
+			start:     2,
+			end:       10,
+			expected:  "[[0 10] [15 20]]",
+			nextStart: 11,
+			nextEnd:   14,
+			last:      20,
+		},
+		{
+			initial:   [][2]uint64{{0, 5}, {15, 20}},
+			start:     8,
+			end:       18,
+			expected:  "[[0 5] [8 20]]",
+			nextStart: 6,
+			nextEnd:   7,
+			last:      20,
+		},
+		{
+			initial:   [][2]uint64{{0, 5}, {15, 20}},
+			start:     2,
+			end:       17,
+			expected:  "[[0 20]]",
+			nextStart: 21,
+			nextEnd:   0,
+			last:      20,
+		},
+		{
+			initial:   [][2]uint64{{0, 5}, {15, 20}},
+			start:     2,
+			end:       25,
+			expected:  "[[0 25]]",
+			nextStart: 26,
+			nextEnd:   0,
+			last:      25,
+		},
+		{
+			initial:   [][2]uint64{{0, 5}, {15, 20}},
+			start:     5,
+			end:       14,
+			expected:  "[[0 20]]",
+			nextStart: 21,
+			nextEnd:   0,
+			last:      20,
+		},
+		{
+			initial:   [][2]uint64{{0, 5}, {15, 20}},
+			start:     6,
+			end:       14,
+			expected:  "[[0 20]]",
+			nextStart: 21,
+			nextEnd:   0,
+			last:      20,
+		},
+		{
+			initial:   [][2]uint64{{0, 5}, {15, 20}, {30, 40}},
+			start:     6,
+			end:       29,
+			expected:  "[[0 40]]",
+			nextStart: 41,
+			nextEnd:   0,
+			last:      40,
+		},
+		{
+			initial:   [][2]uint64{{0, 5}, {15, 20}, {30, 40}, {50, 60}},
+			start:     3,
+			end:       55,
+			expected:  "[[0 60]]",
+			nextStart: 61,
+			nextEnd:   0,
+			last:      60,
+		},
+		{
+			initial:   [][2]uint64{{0, 5}, {15, 20}, {30, 40}, {50, 60}},
+			start:     21,
+			end:       49,
+			expected:  "[[0 5] [15 60]]",
+			nextStart: 6,
+			nextEnd:   14,
+			last:      60,
+		},
+		{
+			initial:   [][2]uint64{{0, 5}, {15, 20}, {30, 40}, {50, 60}},
+			start:     0,
+			end:       100,
+			expected:  "[[0 100]]",
+			nextStart: 101,
+			nextEnd:   0,
+			last:      100,
+		},
+		{
+			startLimit: 100,
+			initial:    nil,
+			start:      0,
+			end:        0,
+			expected:   "[]",
+			nextStart:  100,
+			nextEnd:    0,
+			last:       0,
+		},
+		{
+			startLimit: 100,
+			initial:    nil,
+			start:      20,
+			end:        30,
+			expected:   "[]",
+			nextStart:  100,
+			nextEnd:    0,
+			last:       0,
+		},
+		{
+			startLimit: 100,
+			initial:    nil,
+			start:      50,
+			end:        100,
+			expected:   "[[100 100]]",
+			nextStart:  101,
+			nextEnd:    0,
+			last:       100,
+		},
+		{
+			startLimit: 100,
+			initial:    nil,
+			start:      50,
+			end:        110,
+			expected:   "[[100 110]]",
+			nextStart:  111,
+			nextEnd:    0,
+			last:       110,
+		},
+		{
+			startLimit: 100,
+			initial:    nil,
+			start:      120,
+			end:        130,
+			expected:   "[[120 130]]",
+			nextStart:  100,
+			nextEnd:    119,
+			last:       130,
+		},
+		{
+			startLimit: 100,
+			initial:    nil,
+			start:      120,
+			end:        130,
+			expected:   "[[120 130]]",
+			nextStart:  100,
+			nextEnd:    119,
+			last:       130,
+		},
+	} {
+		intervals := NewIntervals(tc.startLimit)
+		intervals.ranges = tc.initial
+		intervals.Add(tc.start, tc.end)
+		got := intervals.String()
+		if got != tc.expected {
+			t.Errorf("interval #%d: expected %s, got %s", i, tc.expected, got)
+		}
+		nextStart, nextEnd := intervals.Next()
+		if nextStart != tc.nextStart {
+			t.Errorf("interval #%d, expected next start %d, got %d", i, tc.nextStart, nextStart)
+		}
+		if nextEnd != tc.nextEnd {
+			t.Errorf("interval #%d, expected next end %d, got %d", i, tc.nextEnd, nextEnd)
+		}
+		last := intervals.Last()
+		if last != tc.last {
+			t.Errorf("interval #%d, expected last %d, got %d", i, tc.last, last)
+		}
+	}
+}
+
+func TestMerge(t *testing.T) {
+	for i, tc := range []struct {
+		initial  [][2]uint64
+		merge    [][2]uint64
+		expected string
+	}{
+		{
+			initial:  nil,
+			merge:    nil,
+			expected: "[]",
+		},
+		{
+			initial:  [][2]uint64{{10, 20}},
+			merge:    nil,
+			expected: "[[10 20]]",
+		},
+		{
+			initial:  nil,
+			merge:    [][2]uint64{{15, 25}},
+			expected: "[[15 25]]",
+		},
+		{
+			initial:  [][2]uint64{{0, 100}},
+			merge:    [][2]uint64{{150, 250}},
+			expected: "[[0 100] [150 250]]",
+		},
+		{
+			initial:  [][2]uint64{{0, 100}},
+			merge:    [][2]uint64{{101, 250}},
+			expected: "[[0 250]]",
+		},
+		{
+			initial:  [][2]uint64{{0, 10}, {30, 40}},
+			merge:    [][2]uint64{{20, 25}, {41, 50}},
+			expected: "[[0 10] [20 25] [30 50]]",
+		},
+		{
+			initial:  [][2]uint64{{0, 5}, {15, 20}, {30, 40}, {50, 60}},
+			merge:    [][2]uint64{{6, 25}},
+			expected: "[[0 25] [30 40] [50 60]]",
+		},
+	} {
+		intervals := NewIntervals(0)
+		intervals.ranges = tc.initial
+		m := NewIntervals(0)
+		m.ranges = tc.merge
+
+		intervals.Merge(m)
+
+		got := intervals.String()
+		if got != tc.expected {
+			t.Errorf("interval #%d: expected %s, got %s", i, tc.expected, got)
+		}
+	}
+}

swarm/network/stream/intervals/store.go

@@ -0,0 +1,89 @@
+// 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 intervals
+
+import (
+	"errors"
+	"sync"
+)
+
+// ErrNotFound is returned by the Store implementation when the Interval
+// for a specific key does not exist.
+var ErrNotFound = errors.New("not found")
+
+// Store defines methods required to get and retrieve Intervals for different keys.
+// It is meant to be used for intervals persistence for different streams in the
+// stream package.
+type Store interface {
+	Get(key string) (i *Intervals, err error)
+	Put(key string, i *Intervals) (err error)
+	Delete(key string) (err error)
+	Close() error
+}
+
+// MemStore is the reference implementation of Store interface that is supposed
+// to be used in tests.
+type MemStore struct {
+	db map[string]*Intervals
+	mu sync.RWMutex
+}
+
+// NewMemStore returns a new instance of MemStore.
+func NewMemStore() *MemStore {
+	return &MemStore{
+		db: make(map[string]*Intervals),
+	}
+}
+
+// Get retrieves Intervals for a specific key. If there is no Intervals
+// ErrNotFound is returned.
+func (s *MemStore) Get(key string) (i *Intervals, err error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	i, ok := s.db[key]
+	if !ok {
+		return nil, ErrNotFound
+	}
+	return i, nil
+}
+
+// Put stores Intervals for a specific key.
+func (s *MemStore) Put(key string, i *Intervals) (err error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	s.db[key] = i
+	return nil
+}
+
+// Delete removes Intervals stored under a specific key.
+func (s *MemStore) Delete(key string) (err error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if _, ok := s.db[key]; !ok {
+		return ErrNotFound
+	}
+	delete(s.db, key)
+	return nil
+}
+
+// Close doesnot do anything.
+func (s *MemStore) Close() error {
+	return nil
+}

swarm/network/stream/intervals/store_test.go

@@ -0,0 +1,72 @@
+// 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 intervals
+
+import "testing"
+
+// TestMemStore tests basic functionality of MemStore.
+func TestMemStore(t *testing.T) {
+	testStore(t, NewMemStore())
+}
+
+// testStore is a helper function to test various Store implementations.
+func testStore(t *testing.T, s Store) {
+	key1 := "key1"
+	i1 := NewIntervals(0)
+	i1.Add(10, 20)
+	if err := s.Put(key1, i1); err != nil {
+		t.Fatal(err)
+	}
+	g, err := s.Get(key1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if g.String() != i1.String() {
+		t.Errorf("expected interval %s, got %s", i1, g)
+	}
+
+	key2 := "key2"
+	i2 := NewIntervals(0)
+	i2.Add(10, 20)
+	if err := s.Put(key2, i2); err != nil {
+		t.Fatal(err)
+	}
+	g, err = s.Get(key2)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if g.String() != i2.String() {
+		t.Errorf("expected interval %s, got %s", i2, g)
+	}
+
+	if err := s.Delete(key1); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.Get(key1); err != ErrNotFound {
+		t.Errorf("expected error %v, got %s", ErrNotFound, err)
+	}
+	if _, err := s.Get(key2); err != nil {
+		t.Errorf("expected error %v, got %s", nil, err)
+	}
+
+	if err := s.Delete(key2); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.Get(key2); err != ErrNotFound {
+		t.Errorf("expected error %v, got %s", ErrNotFound, err)
+	}
+}

swarm/network/stream/intervals_test.go

@@ -0,0 +1,299 @@
+// 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"
+	externalStreamSessionAt uint64 = 50
+	externalStreamMaxKeys   uint64 = 100
+)
+
+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)
+	r := NewRegistry(addr, delivery, netStore, intervals.NewMemStore(), defaultSkipCheck)
+
+	r.RegisterClientFunc(externalStreamName, func(p *Peer, t []byte, live bool) (Client, error) {
+		return newTestExternalClient(t, db), nil
+	})
+	r.RegisterServerFunc(externalStreamName, func(p *Peer, t []byte, live bool) (Server, error) {
+		return newTestExternalServer(t, externalStreamSessionAt, externalStreamMaxKeys, nil), nil
+	})
+
+	go func() {
+		waitPeerErrC <- waitForPeers(r, 1*time.Second, peerCount(id))
+	}()
+	return &TestExternalRegistry{r}, nil
+}
+
+func TestIntervals(t *testing.T) {
+	testIntervals(t, true, nil)
+	testIntervals(t, false, &Range{From: 9, To: 26})
+	testIntervals(t, true, &Range{From: 9, To: 26})
+}
+
+func testIntervals(t *testing.T, live bool, history *Range) {
+	nodes := 2
+	chunkCount := dataChunkCount
+	skipCheck := false
+
+	defaultSkipCheck = skipCheck
+	toAddr = network.NewAddrFromNodeID
+	conf := &streamTesting.RunConfig{
+		Adapter:        *adapter,
+		NodeCount:      nodes,
+		ConnLevel:      1,
+		ToAddr:         toAddr,
+		Services:       services,
+		DefaultService: "intervalsStreamer",
+	}
+
+	sim, teardown, err := streamTesting.NewSimulation(conf)
+	defer teardown()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	stores = make(map[discover.NodeID]storage.ChunkStore)
+	deliveries = make(map[discover.NodeID]*Delivery)
+	for i, id := range sim.IDs {
+		stores[id] = sim.Stores[i]
+	}
+
+	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{})
+	defer close(quitC)
+
+	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
+			}
+		}
+
+		id := sim.IDs[1]
+
+		err := sim.CallClient(id, func(client *rpc.Client) error {
+
+			sid := sim.IDs[0]
+
+			err := streamTesting.WatchDisconnections(id, client, errc, quitC)
+			if err != nil {
+				return err
+			}
+			ctx, cancel := context.WithTimeout(ctx, 100*time.Second)
+			defer cancel()
+
+			err = client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream(externalStreamName, nil, live), history, Top)
+			if err != nil {
+				return err
+			}
+
+			liveErrC := make(chan error)
+			historyErrC := make(chan error)
+
+			go func() {
+				if !live {
+					close(liveErrC)
+					return
+				}
+
+				var err error
+				defer func() {
+					liveErrC <- err
+				}()
+
+				// live stream
+				liveHashesChan := make(chan []byte)
+				liveSubscription, err := client.Subscribe(ctx, "stream", liveHashesChan, "getHashes", sid, NewStream(externalStreamName, nil, true))
+				if err != nil {
+					return
+				}
+				defer liveSubscription.Unsubscribe()
+
+				i := externalStreamSessionAt
+
+				// we have subscribed, enable notifications
+				err = client.CallContext(ctx, nil, "stream_enableNotifications", sid, NewStream(externalStreamName, nil, true))
+				if err != nil {
+					return
+				}
+
+				for {
+					select {
+					case hash := <-liveHashesChan:
+						h := binary.BigEndian.Uint64(hash)
+						if h != i {
+							err = fmt.Errorf("expected live hash %d, got %d", i, h)
+							return
+						}
+						i++
+						if i > externalStreamMaxKeys {
+							return
+						}
+					case err = <-liveSubscription.Err():
+						return
+					case <-ctx.Done():
+						return
+					}
+				}
+			}()
+
+			go func() {
+				if live && history == nil {
+					close(historyErrC)
+					return
+				}
+
+				var err error
+				defer func() {
+					historyErrC <- err
+				}()
+
+				// history stream
+				historyHashesChan := make(chan []byte)
+				historySubscription, err := client.Subscribe(ctx, "stream", historyHashesChan, "getHashes", sid, NewStream(externalStreamName, nil, false))
+				if err != nil {
+					return
+				}
+				defer historySubscription.Unsubscribe()
+
+				var i uint64
+				historyTo := externalStreamMaxKeys
+				if history != nil {
+					i = history.From
+					if history.To != 0 {
+						historyTo = history.To
+					}
+				}
+
+				// we have subscribed, enable notifications
+				err = client.CallContext(ctx, nil, "stream_enableNotifications", sid, NewStream(externalStreamName, nil, false))
+				if err != nil {
+					return
+				}
+
+				for {
+					select {
+					case hash := <-historyHashesChan:
+						h := binary.BigEndian.Uint64(hash)
+						if h != i {
+							err = fmt.Errorf("expected history hash %d, got %d", i, h)
+							return
+						}
+						i++
+						if i > historyTo {
+							return
+						}
+					case err = <-historySubscription.Err():
+						return
+					case <-ctx.Done():
+						return
+					}
+				}
+			}()
+
+			if err := <-liveErrC; err != nil {
+				return err
+			}
+			if err := <-historyErrC; err != nil {
+				return err
+			}
+
+			return nil
+		})
+		return err
+	}
+	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[1: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/messages.go

@@ -26,11 +26,38 @@ import (
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
+// Stream defines a unique stream identifier.
+type Stream struct {
+	// Name is used for Client and Server functions identification.
+	Name string
+	// Key is the name of specific stream data.
+	Key []byte
+	// Live defines whether the stream delivers only new data
+	// for the specific stream.
+	Live bool
+}
+
+func NewStream(name string, key []byte, live bool) Stream {
+	return Stream{
+		Name: name,
+		Key:  key,
+		Live: live,
+	}
+}
+
+// String return a stream id based on all Stream fields.
+func (s Stream) String() string {
+	t := "h"
+	if s.Live {
+		t = "l"
+	}
+	return fmt.Sprintf("%s|%x|%s", s.Name, s.Key, t)
+}
+
 // SubcribeMsg is the protocol msg for requesting a stream(section)
 type SubscribeMsg struct {
-	Stream   string
+	Stream   Stream
-	Key      []byte
+	History  *Range `rlp:"nil"`
-	From, To uint64
 	Priority uint8  // delivered on priority channel
 }
 
@@ -45,24 +72,53 @@ func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) (err error) {
 		}
 	}()
 
-	f, err := p.streamer.GetServerFunc(req.Stream)
+	log.Debug("received subscription", "peer", p.ID(), "stream", req.Stream, "history", req.History)
+
+	f, err := p.streamer.GetServerFunc(req.Stream.Name)
 	if err != nil {
 		return err
 	}
-	s, err := f(p, req.Key)
+
+	s, err := f(p, req.Stream.Key, req.Stream.Live)
 	if err != nil {
 		return err
 	}
-	os, err := p.setServer(req.Stream, req.Key, s, req.Priority)
+	os, err := p.setServer(req.Stream, s, req.Priority)
 	if err != nil {
 		return err
 	}
-	log.Debug("received subscription", "peer", p.ID(), "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To)
+
+	var from uint64
+	var to uint64
+	if !req.Stream.Live && req.History != nil {
+		from = req.History.From
+		to = req.History.To
+	}
+
 	go func() {
-		if err := p.SendOfferedHashes(os, req.From, req.To); err != nil {
+		if err := p.SendOfferedHashes(os, from, to); err != nil {
 			p.Drop(err)
 		}
 	}()
+
+	if req.Stream.Live && req.History != nil {
+		// subscribe to the history stream
+		s, err := f(p, req.Stream.Key, false)
+		if err != nil {
+			return err
+		}
+
+		os, err := p.setServer(getHistoryStream(req.Stream), s, getHistoryPriority(req.Priority))
+		if err != nil {
+			return err
+		}
+		go func() {
+			if err := p.SendOfferedHashes(os, req.History.From, req.History.To); err != nil {
+				p.Drop(err)
+			}
+		}()
+	}
+
 	return nil
 }
 
@@ -75,20 +131,18 @@ func (p *Peer) handleSubscribeErrorMsg(req *SubscribeErrorMsg) (err error) {
 }
 
 type UnsubscribeMsg struct {
-	Stream string
+	Stream Stream
-	Key    []byte
 }
 
 func (p *Peer) handleUnsubscribeMsg(req *UnsubscribeMsg) error {
-	p.removeServer(req.Stream, req.Key)
+	p.removeServer(req.Stream)
 	return nil
 }
 
 // OfferedHashesMsg is the protocol msg for offering to hand over a
 // stream section
 type OfferedHashesMsg struct {
-	Stream         string // name of Stream
+	Stream         Stream // name of Stream
-	Key            []byte // subtype or key
 	From, To       uint64 // peer and db-specific entry count
 	Hashes         []byte // stream of hashes (128)
 	*HandoverProof        // HandoverProof
@@ -102,9 +156,7 @@ func (m OfferedHashesMsg) String() string {
 // handleOfferedHashesMsg protocol msg handler calls the incoming streamer interface
 // Filter method
 func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
-	sk := req.Stream
+	c, _, err := p.getOrSetClient(req.Stream, req.From, req.To)
-	sk += keyToString(req.Key)
-	s, err := p.getClient(sk)
 	if err != nil {
 		return err
 	}
@@ -117,7 +169,7 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
 	for i := 0; i < len(hashes); i += HashSize {
 		hash := hashes[i : i+HashSize]
 
-		if wait := s.NeedData(hash); wait != nil {
+		if wait := c.NeedData(hash); wait != nil {
 			want.Set(i/HashSize, true)
 			wg.Add(1)
 			// create request and wait until the chunk data arrives and is stored
@@ -142,22 +194,21 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
 	// }()
 	go func() {
 		wg.Wait()
-		s.next <- s.batchDone(p, req, hashes)
+		c.next <- c.batchDone(p, req, hashes)
 	}()
 	// only send wantedKeysMsg if all missing chunks of the previous batch arrived
 	// except
-	if s.live {
+	if c.stream.Live {
-		s.sessionAt = req.From
+		c.sessionAt = req.From
 	}
-	from, to := s.nextBatch(req.To)
+	from, to := c.nextBatch(req.To + 1)
-	log.Trace("received offered batch", "peer", p.ID(), "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To)
+	log.Trace("received offered batch", "peer", p.ID(), "stream", req.Stream, "from", req.From, "to", req.To)
 	if from == to {
 		return nil
 	}
 
 	msg := &WantedHashesMsg{
 		Stream: req.Stream,
-		Key:    req.Key,
 		Want:   want.Bytes(),
 		From:   from,
 		To:     to,
@@ -167,14 +218,14 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
 		case <-time.After(30 * time.Second):
 			p.Drop(err)
 			return
-		case err := <-s.next:
+		case err := <-c.next:
 			if err != nil {
 				p.Drop(err)
 				return
 			}
 		}
-		log.Trace("sending want batch", "peer", p.ID(), "stream", msg.Stream, "Key", msg.Key, "from", msg.From, "to", msg.To)
+		log.Trace("sending want batch", "peer", p.ID(), "stream", msg.Stream, "from", msg.From, "to", msg.To)
-		err := p.SendPriority(msg, s.priority)
+		err := p.SendPriority(msg, c.priority)
 		if err != nil {
 			p.Drop(err)
 		}
@@ -185,8 +236,7 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
 // WantedHashesMsg is the protocol msg data for signaling which hashes
 // offered in OfferedHashesMsg downstream peer actually wants sent over
 type WantedHashesMsg struct {
-	Stream   string // name of stream
+	Stream   Stream
-	Key      []byte // subtype or key
 	Want     []byte // bitvector indicating which keys of the batch needed
 	From, To uint64 // next interval offset - empty if not to be continued
 }
@@ -200,8 +250,8 @@ func (m WantedHashesMsg) String() string {
 // * sends the next batch of unsynced keys
 // * sends the actual data chunks as per WantedHashesMsg
 func (p *Peer) handleWantedHashesMsg(req *WantedHashesMsg) error {
-	log.Trace("received wanted batch", "peer", p.ID(), "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To)
+	log.Trace("received wanted batch", "peer", p.ID(), "stream", req.Stream, "from", req.From, "to", req.To)
-	s, err := p.getServer(req.Stream + keyToString(req.Key))
+	s, err := p.getServer(req.Stream)
 	if err != nil {
 		return err
 	}
@@ -237,7 +287,7 @@ func (p *Peer) handleWantedHashesMsg(req *WantedHashesMsg) error {
 
 // Handover represents a statement that the upstream peer hands over the stream section
 type Handover struct {
-	Stream     string // name of stream
+	Stream     Stream // name of stream
 	Start, End uint64 // index of hashes
 	Root       []byte // Root hash for indexed segment inclusion proofs
 }

swarm/network/stream/peer.go

@@ -18,7 +18,6 @@ package stream
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"sync"
 	"time"
@@ -26,15 +25,24 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p/protocols"
 	pq "github.com/ethereum/go-ethereum/swarm/network/priorityqueue"
+	"github.com/ethereum/go-ethereum/swarm/network/stream/intervals"
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
 var sendTimeout = 5 * time.Second
 
-var (
+type notFoundError struct {
-	errServerNotFound = errors.New("server not found")
+	t string
-	errClientNotFound = errors.New("client not found")
+	s Stream
-)
+}
+
+func newNotFoundError(t string, s Stream) *notFoundError {
+	return &notFoundError{t: t, s: s}
+}
+
+func (e *notFoundError) Error() string {
+	return fmt.Sprintf("%s not found for stream %q", e.t, e.s)
+}
 
 // Peer is the Peer extension for the streaming protocol
 type Peer struct {
@@ -42,9 +50,13 @@ type Peer struct {
 	streamer *Registry
 	pq       *pq.PriorityQueue
 	serverMu sync.RWMutex
-	clientMu sync.RWMutex
+	clientMu sync.RWMutex // protects both clients and clientParams
 	servers  map[string]*server
 	clients  map[string]*client
+	// clientParams map keeps required client arguments
+	// that are set on Registry.Subscribe and used
+	// on creating a new client in offered hashes handler.
+	clientParams map[string]*clientParams
 	quit         chan struct{}
 }
 
@@ -56,6 +68,7 @@ func NewPeer(peer *protocols.Peer, streamer *Registry) *Peer {
 		streamer:     streamer,
 		servers:      make(map[string]*server),
 		clients:      make(map[string]*client),
+		clientParams: make(map[string]*clientParams),
 		quit:         make(chan struct{}),
 	}
 	ctx, cancel := context.WithCancel(context.Background())
@@ -105,103 +118,206 @@ func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error {
 		From:          from,
 		To:            to,
 		Stream:        s.stream,
-		Key:           s.key,
 	}
-	log.Trace("Swarm syncer offer batch", "peer", p.ID(), "stream", s.stream, "key", s.key, "len", len(hashes), "from", from, "to", to)
+	log.Trace("Swarm syncer offer batch", "peer", p.ID(), "stream", s.stream, "len", len(hashes), "from", from, "to", to)
 	return p.SendPriority(msg, s.priority)
 }
 
-func (p *Peer) getServer(s string) (*server, error) {
+func (p *Peer) getServer(s Stream) (*server, error) {
 	p.serverMu.RLock()
 	defer p.serverMu.RUnlock()
 
-	server := p.servers[s]
+	server := p.servers[s.String()]
 	if server == nil {
 		return nil, fmt.Errorf("server '%v' not provided to peer %v", s, p.ID())
 	}
 	return server, nil
 }
 
-func (p *Peer) getClient(s string) (*client, error) {
+func (p *Peer) setServer(s Stream, o Server, priority uint8) (*server, error) {
-	p.clientMu.RLock()
-	defer p.clientMu.RUnlock()
-
-	client := p.clients[s]
-	if client == nil {
-		return nil, fmt.Errorf("client '%v' not provided to peer %v", s, p.ID())
-	}
-	return client, nil
-}
-
-func (p *Peer) setServer(s string, key []byte, o Server, priority uint8) (*server, error) {
 	p.serverMu.Lock()
 	defer p.serverMu.Unlock()
 
-	sk := s + keyToString(key)
+	sk := s.String()
 	if p.servers[sk] != nil {
 		return nil, fmt.Errorf("server %v already registered", sk)
 	}
 	os := &server{
 		Server:   o,
-		priority: priority,
 		stream:   s,
-		key:      key,
+		priority: priority,
 	}
 	p.servers[sk] = os
 	return os, nil
 }
 
-func (p *Peer) removeServer(s string, key []byte) error {
+func (p *Peer) removeServer(s Stream) error {
 	p.serverMu.Lock()
 	defer p.serverMu.Unlock()
 
-	sk := s + keyToString(key)
+	sk := s.String()
 	server, ok := p.servers[sk]
 	if !ok {
-		return errServerNotFound
+		return newNotFoundError("server", s)
 	}
 	server.Close()
 	delete(p.servers, sk)
 	return nil
 }
 
-func (p *Peer) setClient(s string, key []byte, i Client, priority uint8, live bool) error {
+func (p *Peer) getClient(ctx context.Context, s Stream) (c *client, err error) {
+	var params *clientParams
+	sk := s.String()
+	func() {
+		p.clientMu.RLock()
+		defer p.clientMu.RUnlock()
+
+		c = p.clients[sk]
+		if c != nil {
+			return
+		}
+		params = p.clientParams[sk]
+	}()
+	if c != nil {
+		return c, nil
+	}
+
+	if params != nil {
+		//debug.PrintStack()
+		if err := params.waitClient(ctx); err != nil {
+			return nil, err
+		}
+	}
+
+	p.clientMu.RLock()
+	defer p.clientMu.RUnlock()
+
+	c = p.clients[sk]
+	if c != nil {
+		return c, nil
+	}
+	return nil, newNotFoundError("client", s)
+}
+
+func (p *Peer) getOrSetClient(s Stream, from, to uint64) (c *client, created bool, err error) {
+	sk := s.String()
+
 	p.clientMu.Lock()
 	defer p.clientMu.Unlock()
 
-	sk := s + keyToString(key)
+	c = p.clients[sk]
-	if p.clients[sk] != nil {
+	if c != nil {
-		return fmt.Errorf("client %v already registered", sk)
+		return c, false, nil
 	}
+
+	f, err := p.streamer.GetClientFunc(s.Name)
+	if err != nil {
+		return nil, false, err
+	}
+
+	is, err := f(p, s.Key, s.Live)
+	if err != nil {
+		return nil, false, err
+	}
+
+	cp, err := p.getClientParams(s)
+	if err != nil {
+		return nil, false, err
+	}
+	defer func() {
+		if err == nil {
+			if err := p.removeClientParams(s); err != nil {
+				log.Error("stream set client: remove client params", "stream", s, "peer", p, "err", err)
+			}
+		}
+	}()
+
+	intervalsKey := peerStreamIntervalsKey(p, s)
+	if s.Live {
+		// try to find previous history and live intervals and merge live into history
+		historyKey := peerStreamIntervalsKey(p, NewStream(s.Name, s.Key, false))
+		historyIntervals, err := p.streamer.intervalsStore.Get(historyKey)
+		switch err {
+		case nil:
+			liveIntervals, err := p.streamer.intervalsStore.Get(intervalsKey)
+			switch err {
+			case nil:
+				historyIntervals.Merge(liveIntervals)
+				if err := p.streamer.intervalsStore.Put(historyKey, historyIntervals); err != nil {
+					log.Error("stream set client: put history intervals", "stream", s, "peer", p, "err", err)
+				}
+			case intervals.ErrNotFound:
+			default:
+				log.Error("stream set client: get live intervals", "stream", s, "peer", p, "err", err)
+			}
+		case intervals.ErrNotFound:
+		default:
+			log.Error("stream set client: get history intervals", "stream", s, "peer", p, "err", err)
+		}
+	}
+
+	if err := p.streamer.intervalsStore.Put(intervalsKey, intervals.NewIntervals(from)); err != nil {
+		return nil, false, err
+	}
+
 	next := make(chan error, 1)
-	// var intervals *Intervals
+	c = &client{
-	// if !live {
+		Client:         is,
-	// key := s + p.ID().String()
-	// intervals = NewIntervals(key, p.streamer)
-	// }
-	p.clients[sk] = &client{
-		Client: i,
-		// intervals:        intervals,
-		live:     live,
-		priority: priority,
-		next:     next,
 		stream:         s,
-		key:      key,
+		priority:       cp.priority,
+		to:             cp.to,
+		next:           next,
+		intervalsStore: p.streamer.intervalsStore,
+		intervalsKey:   intervalsKey,
 	}
+	p.clients[sk] = c
+	cp.clientCreated() // unblock all possible getClient calls that are waiting
 	next <- nil        // this is to allow wantedKeysMsg before first batch arrives
+	return c, true, nil
+}
+
+func (p *Peer) removeClient(s Stream) error {
+	p.clientMu.Lock()
+	defer p.clientMu.Unlock()
+
+	client, ok := p.clients[s.String()]
+	if !ok {
+		return newNotFoundError("client", s)
+	}
+	client.close()
 	return nil
 }
 
-func (p *Peer) removeClient(s string, key []byte) error {
+func (p *Peer) setClientParams(s Stream, params *clientParams) error {
 	p.clientMu.Lock()
 	defer p.clientMu.Unlock()
 
-	sk := s + keyToString(key)
+	sk := s.String()
-	client, ok := p.clients[sk]
+	if p.clients[sk] != nil {
-	if !ok {
+		return fmt.Errorf("client %v already exists", sk)
-		return errClientNotFound
 	}
-	client.close()
+	if p.clientParams[sk] != nil {
+		return fmt.Errorf("client params %v already set", sk)
+	}
+	p.clientParams[sk] = params
+	return nil
+}
+
+func (p *Peer) getClientParams(s Stream) (*clientParams, error) {
+	params := p.clientParams[s.String()]
+	if params == nil {
+		return nil, fmt.Errorf("client params '%v' not provided to peer %v", s, p.ID())
+	}
+	return params, nil
+}
+
+func (p *Peer) removeClientParams(s Stream) error {
+	sk := s.String()
+	_, ok := p.clientParams[sk]
+	if !ok {
+		return newNotFoundError("client params", s)
+	}
+	delete(p.clientParams, sk)
 	return nil
 }
 

swarm/network/stream/stream.go

@@ -17,12 +17,11 @@
 package stream
 
 import (
+	"context"
 	"fmt"
-	"io"
 	"math"
 	"sync"
 
-	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/rpc"
 
@@ -30,6 +29,7 @@ import (
 	"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/network/stream/intervals"
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
@@ -51,37 +51,39 @@ type Registry struct {
 	clientMu       sync.RWMutex
 	serverMu       sync.RWMutex
 	peersMu        sync.RWMutex
-	serverFuncs map[string]func(*Peer, []byte) (Server, error)
+	serverFuncs    map[string]func(*Peer, []byte, bool) (Server, error)
-	clientFuncs map[string]func(*Peer, []byte) (Client, error)
+	clientFuncs    map[string]func(*Peer, []byte, bool) (Client, error)
 	peers          map[discover.NodeID]*Peer
 	delivery       *Delivery
 	store          storage.ChunkStore
+	intervalsStore intervals.Store
 }
 
 // NewRegistry is Streamer constructor
-func NewRegistry(addr *network.BzzAddr, delivery *Delivery, store storage.ChunkStore, skipCheck bool) *Registry {
+func NewRegistry(addr *network.BzzAddr, delivery *Delivery, store storage.ChunkStore, intervalsStore intervals.Store, skipCheck bool) *Registry {
 	streamer := &Registry{
 		addr:           addr,
 		skipCheck:      skipCheck,
 		store:          store,
-		serverFuncs: make(map[string]func(*Peer, []byte) (Server, error)),
+		serverFuncs:    make(map[string]func(*Peer, []byte, bool) (Server, error)),
-		clientFuncs: make(map[string]func(*Peer, []byte) (Client, error)),
+		clientFuncs:    make(map[string]func(*Peer, []byte, bool) (Client, error)),
 		peers:          make(map[discover.NodeID]*Peer),
 		delivery:       delivery,
+		intervalsStore: intervalsStore,
 	}
 	streamer.api = NewAPI(streamer, streamer.store)
 	delivery.getPeer = streamer.getPeer
-	streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, t []byte) (Server, error) {
+	streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, _ []byte, _ bool) (Server, error) {
 		return NewSwarmChunkServer(delivery.db), nil
 	})
-	streamer.RegisterClientFunc(swarmChunkServerStreamName, func(p *Peer, t []byte) (Client, error) {
+	streamer.RegisterClientFunc(swarmChunkServerStreamName, func(p *Peer, _ []byte, _ bool) (Client, error) {
 		return NewSwarmSyncerClient(p, delivery.db, nil)
 	})
 	return streamer
 }
 
 // RegisterClient registers an incoming streamer constructor
-func (r *Registry) RegisterClientFunc(stream string, f func(*Peer, []byte) (Client, error)) {
+func (r *Registry) RegisterClientFunc(stream string, f func(*Peer, []byte, bool) (Client, error)) {
 	r.clientMu.Lock()
 	defer r.clientMu.Unlock()
 
@@ -89,7 +91,7 @@ func (r *Registry) RegisterClientFunc(stream string, f func(*Peer, []byte) (Clie
 }
 
 // RegisterServer registers an outgoing streamer constructor
-func (r *Registry) RegisterServerFunc(stream string, f func(*Peer, []byte) (Server, error)) {
+func (r *Registry) RegisterServerFunc(stream string, f func(*Peer, []byte, bool) (Server, error)) {
 	r.serverMu.Lock()
 	defer r.serverMu.Unlock()
 
@@ -97,7 +99,7 @@ func (r *Registry) RegisterServerFunc(stream string, f func(*Peer, []byte) (Serv
 }
 
 // GetClient accessor for incoming streamer constructors
-func (r *Registry) GetClientFunc(stream string) (func(*Peer, []byte) (Client, error), error) {
+func (r *Registry) GetClientFunc(stream string) (func(*Peer, []byte, bool) (Client, error), error) {
 	r.clientMu.RLock()
 	defer r.clientMu.RUnlock()
 
@@ -109,7 +111,7 @@ func (r *Registry) GetClientFunc(stream string) (func(*Peer, []byte) (Client, er
 }
 
 // GetServer accessor for incoming streamer constructors
-func (r *Registry) GetServerFunc(stream string) (func(*Peer, []byte) (Server, error), error) {
+func (r *Registry) GetServerFunc(stream string) (func(*Peer, []byte, bool) (Server, error), error) {
 	r.serverMu.RLock()
 	defer r.serverMu.RUnlock()
 
@@ -121,9 +123,9 @@ func (r *Registry) GetServerFunc(stream string) (func(*Peer, []byte) (Server, er
 }
 
 // Subscribe initiates the streamer
-func (r *Registry) Subscribe(peerId discover.NodeID, s string, t []byte, from, to uint64, priority uint8, live bool) error {
+func (r *Registry) Subscribe(peerId discover.NodeID, s Stream, h *Range, priority uint8) error {
-	f, err := r.GetClientFunc(s)
+	// check if the stream is registered
-	if err != nil {
+	if _, err := r.GetClientFunc(s.Name); err != nil {
 		return err
 	}
 
@@ -132,29 +134,36 @@ func (r *Registry) Subscribe(peerId discover.NodeID, s string, t []byte, from, t
 		return fmt.Errorf("peer not found %v", peerId)
 	}
 
-	is, err := f(peer, t)
+	var to uint64
+	if !s.Live && h != nil {
+		to = h.To
+	}
+
+	err := peer.setClientParams(s, newClientParams(priority, to))
 	if err != nil {
 		return err
 	}
-	err = peer.setClient(s, t, is, priority, live)
+
-	if err != nil {
+	if s.Live && h != nil {
+		if err := peer.setClientParams(
+			getHistoryStream(s),
+			newClientParams(getHistoryPriority(priority), h.To),
+		); err != nil {
 			return err
 		}
+	}
 
 	msg := &SubscribeMsg{
 		Stream:   s,
-		Key:    t,
+		History:  h,
-		// Live:     live,
-		From:     from,
-		To:       to,
 		Priority: priority,
 	}
-	log.Debug("Subscribe ", "peer", peerId, "stream", s, "key", t, "from", from, "to", to)
+	log.Debug("Subscribe ", "peer", peerId, "stream", s, "history", h)
 
 	return peer.SendPriority(msg, priority)
 }
 
-func (r *Registry) Unsubscribe(peerId discover.NodeID, s string, t []byte) error {
+func (r *Registry) Unsubscribe(peerId discover.NodeID, s Stream) error {
 	peer := r.getPeer(peerId)
 	if peer == nil {
 		return fmt.Errorf("peer not found %v", peerId)
@@ -162,14 +171,13 @@ func (r *Registry) Unsubscribe(peerId discover.NodeID, s string, t []byte) error
 
 	msg := &UnsubscribeMsg{
 		Stream: s,
-		Key:    t,
 	}
-	log.Debug("Unsubscribe ", "peer", peerId, "stream", s, "key", t)
+	log.Debug("Unsubscribe ", "peer", peerId, "stream", s)
 
 	if err := peer.Send(msg); err != nil {
 		return err
 	}
-	return peer.removeClient(s, t)
+	return peer.removeClient(s)
 }
 
 func (r *Registry) Retrieve(chunk *storage.Chunk) error {
@@ -184,6 +192,11 @@ func (r *Registry) PeerInfo(id discover.NodeID) interface{} {
 	return nil
 }
 
+func (r *Registry) Close() error {
+	r.store.Close()
+	return r.intervalsStore.Close()
+}
+
 func (r *Registry) getPeer(peerId discover.NodeID) *Peer {
 	r.peersMu.RLock()
 	defer r.peersMu.RUnlock()
@@ -261,20 +274,11 @@ func (p *Peer) HandleMsg(msg interface{}) error {
 	}
 }
 
-func keyToString(key []byte) string {
-	l := len(key)
-	if l == 0 {
-		return ""
-	}
-	return fmt.Sprintf("%s-%d", string(key[:l-1]), key[l-1])
-}
-
 type server struct {
 	Server
+	stream       Stream
 	priority     uint8
 	currentBatch []byte
-	stream       string
-	key          []byte
 }
 
 // Server interface for outgoing peer Streamer
@@ -286,50 +290,69 @@ type Server interface {
 
 type client struct {
 	Client
+	stream    Stream
 	priority  uint8
 	sessionAt uint64
-	live      bool
+	to        uint64
-	stream    string
-	key       []byte
 	next      chan error
+
+	intervalsKey   string
+	intervalsStore intervals.Store
+}
+
+func peerStreamIntervalsKey(p *Peer, s Stream) string {
+	return p.ID().String() + s.String()
+}
+
+func (c client) AddInterval(start, end uint64) (err error) {
+	i, err := c.intervalsStore.Get(c.intervalsKey)
+	if err != nil {
+		return err
+	}
+	i.Add(start, end)
+	return c.intervalsStore.Put(c.intervalsKey, i)
+}
+
+func (c client) NextInterval() (start, end uint64, err error) {
+	i, err := c.intervalsStore.Get(c.intervalsKey)
+	if err != nil {
+		return 0, 0, err
+	}
+	start, end = i.Next()
+	return start, end, nil
 }
 
 // Client interface for incoming peer Streamer
 type Client interface {
 	NeedData([]byte) func()
-	BatchDone(string, uint64, []byte, []byte) func() (*TakeoverProof, error)
+	BatchDone(Stream, uint64, []byte, []byte) func() (*TakeoverProof, error)
 	Close()
 }
 
-// nextBatch adjusts the indexes by inspecting the intervals
 func (c *client) nextBatch(from uint64) (nextFrom uint64, nextTo uint64) {
-	var intervals []uint64
+	if c.to > 0 && from >= c.to {
-	if c.live {
+		return 0, 0
-		if len(intervals) == 0 {
-			intervals = []uint64{c.sessionAt, from}
-		} else {
-			intervals[1] = from
 	}
-		nextFrom = from
+	if c.stream.Live {
-	} else if from >= c.sessionAt { // history sync complete
+		return from, 0
-		intervals = nil
+	} else if from >= c.sessionAt {
-		nextFrom = from
+		if c.to > 0 {
-		nextTo = math.MaxUint64
+			return from, c.to
-	} else if len(intervals) > 2 && from >= intervals[2] { // filled a gap in the intervals
+		}
-		intervals = append(intervals[:1], intervals[3:]...)
+		return from, math.MaxUint64
-		nextFrom = intervals[1]
+	}
-		if len(intervals) > 2 {
+	nextFrom, nextTo, err := c.NextInterval()
-			nextTo = intervals[2]
+	if err != nil {
-		} else {
+		log.Error("next intervals", "stream", c.stream)
+		return
+	}
+	if nextTo > c.to {
+		nextTo = c.to
+	}
+	if nextTo == 0 {
 		nextTo = c.sessionAt
 	}
-	} else {
+	return
-		nextFrom = from
-		intervals[1] = from
-		nextTo = c.sessionAt
-	}
-	// b.intervals.set(intervals)
-	return nextFrom, nextTo
 }
 
 func (c *client) batchDone(p *Peer, req *OfferedHashesMsg, hashes []byte) error {
@@ -338,7 +361,17 @@ func (c *client) batchDone(p *Peer, req *OfferedHashesMsg, hashes []byte) error
 		if err != nil {
 			return err
 		}
-		return p.SendPriority(tp, c.priority)
+		// TODO: make a test case for testing if the interval is added when the batch is done
+		if err := c.AddInterval(tp.Takeover.Start, tp.Takeover.End); err != nil {
+			return err
+		}
+		if err := p.SendPriority(tp, c.priority); err != nil {
+			return err
+		}
+		if c.to > 0 && tp.Takeover.End >= c.to {
+			return p.streamer.Unsubscribe(p.Peer.ID(), req.Stream)
+		}
+		return nil
 	}
 	return nil
 }
@@ -348,6 +381,36 @@ func (c *client) close() {
 	c.Close()
 }
 
+// clientParams store parameters for the new client
+// between a subscription and initial offered hashes request handling.
+type clientParams struct {
+	priority uint8
+	to       uint64
+	// signal when the client is created
+	clientCreatedC chan struct{}
+}
+
+func newClientParams(priority uint8, to uint64) *clientParams {
+	return &clientParams{
+		priority:       priority,
+		to:             to,
+		clientCreatedC: make(chan struct{}),
+	}
+}
+
+func (c *clientParams) waitClient(ctx context.Context) error {
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-c.clientCreatedC:
+		return nil
+	}
+}
+
+func (c *clientParams) clientCreated() {
+	close(c.clientCreatedC)
+}
+
 // Spec is the spec of the streamer protocol
 var Spec = &protocols.Spec{
 	Name:       "stream",
@@ -399,6 +462,21 @@ func (r *Registry) Stop() error {
 	return nil
 }
 
+type Range struct {
+	From, To uint64
+}
+
+func getHistoryPriority(priority uint8) uint8 {
+	if priority == 0 {
+		return 0
+	}
+	return priority - 1
+}
+
+func getHistoryStream(s Stream) Stream {
+	return NewStream(s.Name, s.Key, false)
+}
+
 type API struct {
 	streamer *Registry
 	dpa      *storage.DPA
@@ -412,30 +490,10 @@ func NewAPI(r *Registry, store storage.ChunkStore) *API {
 	}
 }
 
-func readAll(dpa *storage.DPA, hash []byte) (int64, error) {
+func (api *API) SubscribeStream(peerId discover.NodeID, s Stream, history *Range, priority uint8) error {
-	r := dpa.Retrieve(hash)
+	return api.streamer.Subscribe(peerId, s, history, priority)
-	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) {
+func (api *API) UnsubscribeStream(peerId discover.NodeID, s Stream) error {
-	return readAll(api.dpa, hash[:])
+	return api.streamer.Unsubscribe(peerId, s)
-}
-
-func (api *API) SubscribeStream(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)
-}
-
-func (api *API) UnsubscribeStream(peerId discover.NodeID, s string, t []byte) error {
-	return api.streamer.Unsubscribe(peerId, s, t)
 }

swarm/network/stream/streamer_test.go

@@ -32,7 +32,8 @@ func TestStreamerSubscribe(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	err = streamer.Subscribe(tester.IDs[0], "foo", nil, 0, 0, Top, true)
+	stream := NewStream("foo", nil, true)
+	err = streamer.Subscribe(tester.IDs[0], stream, &Range{From: 0, To: 0}, Top)
 	if err == nil || err.Error() != "stream foo not registered" {
 		t.Fatalf("Expected error %v, got %v", "stream foo not registered", err)
 	}
@@ -44,40 +45,56 @@ var (
 	hash2     = sha3.Sum256([]byte{2})
 	hashesTmp = append(hash0[:], hash1[:]...)
 	hashes    = append(hashesTmp, hash2[:]...)
-	receivedHashes map[string][]byte = make(map[string][]byte)
-	wait0                            = make(chan bool)
-	wait2                            = make(chan bool)
-	batchDone                        = make(chan bool)
 )
 
 type testClient struct {
 	t              []byte
+	wait0          chan bool
+	wait2          chan bool
+	batchDone      chan bool
+	receivedHashes map[string][]byte
 }
 
+func newTestClient(t []byte) *testClient {
+	return &testClient{
+		t:              t,
+		wait0:          make(chan bool),
+		wait2:          make(chan bool),
+		batchDone:      make(chan bool),
+		receivedHashes: make(map[string][]byte),
+	}
+}
+
+func (self *testClient) NeedData(hash []byte) func() {
+	self.receivedHashes[string(hash)] = hash
+	if bytes.Equal(hash, hash0[:]) {
+		return func() {
+			<-self.wait0
+		}
+	} else if bytes.Equal(hash, hash2[:]) {
+		return func() {
+			<-self.wait2
+		}
+	}
+	return nil
+}
+
+func (self *testClient) BatchDone(Stream, uint64, []byte, []byte) func() (*TakeoverProof, error) {
+	close(self.batchDone)
+	return nil
+}
+
+func (self *testClient) Close() {}
+
 type testServer struct {
 	t []byte
 }
 
-func (self *testClient) NeedData(hash []byte) func() {
+func newTestServer(t []byte) *testServer {
-	receivedHashes[string(hash)] = hash
+	return &testServer{
-	if bytes.Equal(hash, hash0[:]) {
+		t: t,
-		return func() {
-			<-wait0
-		}
-	} else if bytes.Equal(hash, hash2[:]) {
-		return func() {
-			<-wait2
 	}
 }
-	return nil
-}
-
-func (self *testClient) BatchDone(string, uint64, []byte, []byte) func() (*TakeoverProof, error) {
-	close(batchDone)
-	return nil
-}
-
-func (self *testClient) Close() {}
 
 func (self *testServer) SetNextBatch(from uint64, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) {
 	return make([]byte, HashSize), from + 1, to + 1, nil, nil
@@ -97,41 +114,73 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	streamer.RegisterClientFunc("foo", func(p *Peer, t []byte) (Client, error) {
+	streamer.RegisterClientFunc("foo", func(p *Peer, t []byte, live bool) (Client, error) {
-		return &testClient{
+		return newTestClient(t), nil
-			t: t,
-		}, nil
 	})
 
 	peerID := tester.IDs[0]
 
-	err = streamer.Subscribe(peerID, "foo", []byte{}, 5, 8, Top, true)
+	stream := NewStream("foo", nil, true)
+	err = streamer.Subscribe(peerID, stream, &Range{From: 5, To: 8}, Top)
 	if err != nil {
 		t.Fatalf("Expected no error, got %v", err)
 	}
 
-	err = tester.TestExchanges(p2ptest.Exchange{
+	err = tester.TestExchanges(
+		p2ptest.Exchange{
 			Label: "Subscribe message",
 			Expects: []p2ptest.Expect{
 				{
 					Code: 4,
 					Msg: &SubscribeMsg{
-					Stream:   "foo",
+						Stream: stream,
-					Key:      []byte{},
+						History: &Range{
 							From: 5,
 							To:   8,
+						},
 						Priority: Top,
 					},
 					Peer: peerID,
 				},
 			},
-	})
+		},
-
+		// trigger OfferedHashesMsg to actually create the client
+		p2ptest.Exchange{
+			Label: "OfferedHashes message",
+			Triggers: []p2ptest.Trigger{
+				{
+					Code: 1,
+					Msg: &OfferedHashesMsg{
+						HandoverProof: &HandoverProof{
+							Handover: &Handover{},
+						},
+						Hashes: hashes,
+						From:   5,
+						To:     8,
+						Stream: stream,
+					},
+					Peer: peerID,
+				},
+			},
+			Expects: []p2ptest.Expect{
+				{
+					Code: 2,
+					Msg: &WantedHashesMsg{
+						Stream: stream,
+						Want:   []byte{5},
+						From:   9,
+						To:     0,
+					},
+					Peer: peerID,
+				},
+			},
+		},
+	)
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	err = streamer.Unsubscribe(peerID, "foo", []byte{})
+	err = streamer.Unsubscribe(peerID, stream)
 	if err != nil {
 		t.Fatalf("Expected no error, got %v", err)
 	}
@@ -142,8 +191,7 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 			{
 				Code: 0,
 				Msg: &UnsubscribeMsg{
-					Stream: "foo",
+					Stream: stream,
-					Key:    []byte{},
 				},
 				Peer: peerID,
 			},
@@ -162,10 +210,10 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	streamer.RegisterServerFunc("foo", func(p *Peer, t []byte) (Server, error) {
+	stream := NewStream("foo", nil, false)
-		return &testServer{
+
-			t: t,
+	streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) {
-		}, nil
+		return newTestServer(t), nil
 	})
 
 	peerID := tester.IDs[0]
@@ -176,10 +224,11 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 			{
 				Code: 4,
 				Msg: &SubscribeMsg{
-					Stream:   "foo",
+					Stream: stream,
-					Key:      []byte{},
+					History: &Range{
 						From: 5,
 						To:   8,
+					},
 					Priority: Top,
 				},
 				Peer: peerID,
@@ -189,8 +238,7 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 			{
 				Code: 1,
 				Msg: &OfferedHashesMsg{
-					Stream: "foo",
+					Stream: stream,
-					Key:    []byte{},
 					HandoverProof: &HandoverProof{
 						Handover: &Handover{},
 					},
@@ -213,8 +261,73 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
 			{
 				Code: 0,
 				Msg: &UnsubscribeMsg{
-					Stream: "foo",
+					Stream: stream,
-					Key:    []byte{},
+				},
+				Peer: peerID,
+			},
+		},
+	})
+
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestStreamerUpstreamSubscribeUnsubscribeMsgExchangeLive(t *testing.T) {
+	tester, streamer, _, teardown, err := newStreamerTester(t)
+	defer teardown()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	stream := NewStream("foo", nil, true)
+
+	streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) {
+		return newTestServer(t), nil
+	})
+
+	peerID := tester.IDs[0]
+
+	err = tester.TestExchanges(p2ptest.Exchange{
+		Label: "Subscribe message",
+		Triggers: []p2ptest.Trigger{
+			{
+				Code: 4,
+				Msg: &SubscribeMsg{
+					Stream:   stream,
+					Priority: Top,
+				},
+				Peer: peerID,
+			},
+		},
+		Expects: []p2ptest.Expect{
+			{
+				Code: 1,
+				Msg: &OfferedHashesMsg{
+					Stream: stream,
+					HandoverProof: &HandoverProof{
+						Handover: &Handover{},
+					},
+					Hashes: make([]byte, HashSize),
+					From:   1,
+					To:     1,
+				},
+				Peer: peerID,
+			},
+		},
+	})
+
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = tester.TestExchanges(p2ptest.Exchange{
+		Label: "unsubscribe message",
+		Triggers: []p2ptest.Trigger{
+			{
+				Code: 0,
+				Msg: &UnsubscribeMsg{
+					Stream: stream,
 				},
 				Peer: peerID,
 			},
@@ -233,12 +346,12 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	streamer.RegisterServerFunc("foo", func(p *Peer, t []byte) (Server, error) {
+	streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) {
-		return &testServer{
+		return newTestServer(t), nil
-			t: t,
-		}, nil
 	})
 
+	stream := NewStream("bar", nil, true)
+
 	peerID := tester.IDs[0]
 
 	err = tester.TestExchanges(p2ptest.Exchange{
@@ -247,10 +360,11 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) {
 			{
 				Code: 4,
 				Msg: &SubscribeMsg{
-					Stream:   "bar",
+					Stream: stream,
-					Key:      []byte{},
+					History: &Range{
 						From: 5,
 						To:   8,
+					},
 					Priority: Top,
 				},
 				Peer: peerID,
@@ -272,6 +386,74 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) {
 	}
 }
 
+func TestStreamerUpstreamSubscribeLiveAndHistory(t *testing.T) {
+	tester, streamer, _, teardown, err := newStreamerTester(t)
+	defer teardown()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	stream := NewStream("foo", nil, true)
+
+	streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) {
+		return &testServer{
+			t: t,
+		}, nil
+	})
+
+	peerID := tester.IDs[0]
+
+	err = tester.TestExchanges(p2ptest.Exchange{
+		Label: "Subscribe message",
+		Triggers: []p2ptest.Trigger{
+			{
+				Code: 4,
+				Msg: &SubscribeMsg{
+					Stream: stream,
+					History: &Range{
+						From: 5,
+						To:   8,
+					},
+					Priority: Top,
+				},
+				Peer: peerID,
+			},
+		},
+		Expects: []p2ptest.Expect{
+			{
+				Code: 1,
+				Msg: &OfferedHashesMsg{
+					Stream: NewStream("foo", nil, false),
+					HandoverProof: &HandoverProof{
+						Handover: &Handover{},
+					},
+					Hashes: make([]byte, HashSize),
+					From:   6,
+					To:     9,
+				},
+				Peer: peerID,
+			},
+			{
+				Code: 1,
+				Msg: &OfferedHashesMsg{
+					Stream: stream,
+					HandoverProof: &HandoverProof{
+						Handover: &Handover{},
+					},
+					From:   1,
+					To:     1,
+					Hashes: make([]byte, HashSize),
+				},
+				Peer: peerID,
+			},
+		},
+	})
+
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
 func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
 	tester, streamer, _, teardown, err := newStreamerTester(t)
 	defer teardown()
@@ -279,15 +461,18 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	streamer.RegisterClientFunc("foo", func(p *Peer, t []byte) (Client, error) {
+	stream := NewStream("foo", nil, true)
-		return &testClient{
+
-			t: t,
+	var tc *testClient
-		}, nil
+
+	streamer.RegisterClientFunc("foo", func(p *Peer, t []byte, live bool) (Client, error) {
+		tc = newTestClient(t)
+		return tc, nil
 	})
 
 	peerID := tester.IDs[0]
 
-	err = streamer.Subscribe(peerID, "foo", []byte{}, 5, 8, Top, true)
+	err = streamer.Subscribe(peerID, stream, &Range{From: 5, To: 8}, Top)
 	if err != nil {
 		t.Fatalf("Expected no error, got %v", err)
 	}
@@ -298,10 +483,11 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
 			{
 				Code: 4,
 				Msg: &SubscribeMsg{
-					Stream:   "foo",
+					Stream: stream,
-					Key:      []byte{},
+					History: &Range{
 						From: 5,
 						To:   8,
+					},
 					Priority: Top,
 				},
 				Peer: peerID,
@@ -320,7 +506,7 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
 						Hashes: hashes,
 						From:   5,
 						To:     8,
-						Stream: "foo",
+						Stream: stream,
 					},
 					Peer: peerID,
 				},
@@ -329,9 +515,9 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
 				{
 					Code: 2,
 					Msg: &WantedHashesMsg{
-						Stream: "foo",
+						Stream: stream,
 						Want:   []byte{5},
-						From:   8,
+						From:   9,
 						To:     0,
 					},
 					Peer: peerID,
@@ -342,28 +528,28 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	if len(receivedHashes) != 3 {
+	if len(tc.receivedHashes) != 3 {
-		t.Fatalf("Expected number of received hashes %v, got %v", 3, len(receivedHashes))
+		t.Fatalf("Expected number of received hashes %v, got %v", 3, len(tc.receivedHashes))
 	}
 
-	close(wait0)
+	close(tc.wait0)
 
 	timeout := time.NewTimer(100 * time.Millisecond)
 	defer timeout.Stop()
 
 	select {
-	case <-batchDone:
+	case <-tc.batchDone:
 		t.Fatal("batch done early")
 	case <-timeout.C:
 	}
 
-	close(wait2)
+	close(tc.wait2)
 
 	timeout2 := time.NewTimer(10000 * time.Millisecond)
 	defer timeout2.Stop()
 
 	select {
-	case <-batchDone:
+	case <-tc.batchDone:
 	case <-timeout2.C:
 		t.Fatal("timeout waiting batchdone call")
 	}

swarm/network/stream/syncer.go

@@ -64,10 +64,9 @@ func NewSwarmSyncerServer(live bool, po uint8, db *storage.DBAPI) (*SwarmSyncerS
 const maxPO = 32
 
 func RegisterSwarmSyncerServer(streamer *Registry, db *storage.DBAPI) {
-	streamer.RegisterServerFunc("SYNC", func(p *Peer, t []byte) (Server, error) {
+	streamer.RegisterServerFunc("SYNC", func(p *Peer, t []byte, live bool) (Server, error) {
 		po := t[0]
-		// TODO: make this work for HISTORY too
+		return NewSwarmSyncerServer(live, po, db)
-		return NewSwarmSyncerServer(false, po, db)
 	})
 	// streamer.RegisterServerFunc(stream, func(p *Peer) (Server, error) {
 	// 	return NewOutgoingProvableSwarmSyncer(po, db)
@@ -188,7 +187,7 @@ func NewSwarmSyncerClient(_ *Peer, db *storage.DBAPI, chunker storage.Chunker) (
 // RegisterSwarmSyncerClient registers the client constructor function for
 // to handle incoming sync streams
 func RegisterSwarmSyncerClient(streamer *Registry, db *storage.DBAPI) {
-	streamer.RegisterClientFunc("SYNC", func(p *Peer, t []byte) (Client, error) {
+	streamer.RegisterClientFunc("SYNC", func(p *Peer, t []byte, love bool) (Client, error) {
 		return NewSwarmSyncerClient(p, db, nil)
 	})
 }
@@ -207,14 +206,14 @@ func (s *SwarmSyncerClient) NeedData(key []byte) (wait func()) {
 }
 
 // BatchDone
-func (s *SwarmSyncerClient) BatchDone(streamName string, from uint64, hashes []byte, root []byte) func() (*TakeoverProof, error) {
+func (s *SwarmSyncerClient) BatchDone(stream Stream, from uint64, hashes []byte, root []byte) func() (*TakeoverProof, error) {
 	if s.chunker != nil {
-		return func() (*TakeoverProof, error) { return s.TakeoverProof(streamName, from, hashes, root) }
+		return func() (*TakeoverProof, error) { return s.TakeoverProof(stream, from, hashes, root) }
 	}
 	return nil
 }
 
-func (s *SwarmSyncerClient) TakeoverProof(streamName string, from uint64, hashes []byte, root storage.Key) (*TakeoverProof, error) {
+func (s *SwarmSyncerClient) TakeoverProof(stream Stream, from uint64, hashes []byte, root storage.Key) (*TakeoverProof, error) {
 	// for provable syncer currentRoot is non-zero length
 	if s.chunker != nil {
 		if from > s.sessionAt { // for live syncing currentRoot is always updated
@@ -241,8 +240,7 @@ func (s *SwarmSyncerClient) TakeoverProof(streamName string, from uint64, hashes
 	}
 	s.end += uint64(len(hashes)) / HashSize
 	takeover := &Takeover{
-		Stream: streamName,
+		Stream: stream,
-		// Key:    s.Key,
 		Start:  s.start,
 		End:    s.end,
 		Root:   root,

swarm/network/stream/syncer_test.go

@@ -161,7 +161,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 				defer cancel()
 				// start syncing, i.e., subscribe to upstream peers po 1 bin
 				sid := sim.IDs[j+1]
-				return client.CallContext(ctx, nil, "stream_subscribeStream", sid, "SYNC", []byte{1}, 0, 0, Top, false)
+				return client.CallContext(ctx, nil, "stream_subscribeStream", sid, NewStream("SYNC", []byte{1}, false), &Range{From: 0, To: 0}, Top)
 			})
 			if err != nil {
 				return err

swarm/network/stream/testing/testing.go

@@ -123,6 +123,7 @@ type RunConfig struct {
 	ConnLevel       int
 	ToAddr          func(discover.NodeID) *network.BzzAddr
 	Services        adapters.Services
+	DefaultService  string
 	EnableMsgEvents bool
 }
 
@@ -133,9 +134,13 @@ func NewSimulation(conf *RunConfig) (*Simulation, func(), error) {
 	if err != nil {
 		return nil, adapterTeardown, err
 	}
+	defaultService := "streamer"
+	if conf.DefaultService != "" {
+		defaultService = conf.DefaultService
+	}
 	net := simulations.NewNetwork(adapter, &simulations.NetworkConfig{
 		ID:             "0",
-		DefaultService: "streamer",
+		DefaultService: defaultService,
 	})
 	teardown := func() {
 		adapterTeardown()

swarm/swarm.go

@@ -23,6 +23,7 @@ import (
 	"fmt"
 	"math/big"
 	"net"
+	"path/filepath"
 	"strings"
 	"time"
 	"unicode"
@@ -46,6 +47,7 @@ import (
 	"github.com/ethereum/go-ethereum/swarm/fuse"
 	"github.com/ethereum/go-ethereum/swarm/network"
 	"github.com/ethereum/go-ethereum/swarm/network/stream"
+	"github.com/ethereum/go-ethereum/swarm/network/stream/intervals"
 	"github.com/ethereum/go-ethereum/swarm/pss"
 	"github.com/ethereum/go-ethereum/swarm/storage"
 	"github.com/ethereum/go-ethereum/swarm/storage/mock"
@@ -146,7 +148,12 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api.
 
 	db := storage.NewDBAPI(self.lstore)
 	delivery := stream.NewDelivery(to, db)
-	self.streamer = stream.NewRegistry(addr, delivery, self.lstore, false)
+	// TODO: decide on intervals store file location
+	intervalsStore, err := intervals.NewDBStore(filepath.Join(config.Path, "stream-intervals.db"))
+	if err != nil {
+		return
+	}
+	self.streamer = stream.NewRegistry(addr, delivery, self.lstore, intervalsStore, false)
 	stream.RegisterSwarmSyncerServer(self.streamer, db)
 	stream.RegisterSwarmSyncerClient(self.streamer, db)