diff --git a/swarm/network/light/lightnode.go b/swarm/network/light/lightnode.go
index 06a0b4efcd..ad62a88425 100644
--- a/swarm/network/light/lightnode.go
+++ b/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
})
diff --git a/swarm/network/stream/common_test.go b/swarm/network/stream/common_test.go
index 1deb6ffba9..2a8245413b 100644
--- a/swarm/network/stream/common_test.go
+++ b/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"
)
@@ -45,7 +53,8 @@ var (
)
var services = adapters.Services{
- "streamer": NewStreamerService,
+ "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() {}
diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go
index 024d612dcd..7b44d090f8 100644
--- a/swarm/network/stream/delivery.go
+++ b/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
}
diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go
index 73f0caba40..690ae707dd 100644
--- a/swarm/network/stream/delivery_test.go
+++ b/swarm/network/stream/delivery_test.go
@@ -87,10 +87,11 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) {
peer := streamer.getPeer(peerID)
peer.handleSubscribeMsg(&SubscribeMsg{
- Stream: swarmChunkServerStreamName,
- Key: nil,
- From: 0,
- To: 0,
+ Stream: NewStream(swarmChunkServerStreamName, nil, false),
+ 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,
- Key: nil,
- From: 0,
- To: 0,
+ Stream: NewStream(swarmChunkServerStreamName, nil, false),
+ 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: swarmChunkServerStreamName,
+ Stream: NewStream(swarmChunkServerStreamName, nil, false),
},
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",
- Key: []byte{},
- From: 5,
- To: 8,
+ Stream: stream,
+ 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
diff --git a/swarm/network/stream/intervals/dbstore.go b/swarm/network/stream/intervals/dbstore.go
new file mode 100644
index 0000000000..1849c09433
--- /dev/null
+++ b/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 .
+
+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()
+}
diff --git a/swarm/network/stream/intervals/dbstore_test.go b/swarm/network/stream/intervals/dbstore_test.go
new file mode 100644
index 0000000000..75a7ddbfb4
--- /dev/null
+++ b/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 .
+
+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)
+}
diff --git a/swarm/network/stream/intervals/intervals.go b/swarm/network/stream/intervals/intervals.go
new file mode 100644
index 0000000000..5fd820da87
--- /dev/null
+++ b/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 .
+
+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
+}
diff --git a/swarm/network/stream/intervals/intervals_test.go b/swarm/network/stream/intervals/intervals_test.go
new file mode 100644
index 0000000000..b5212f0d91
--- /dev/null
+++ b/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 .
+
+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)
+ }
+ }
+}
diff --git a/swarm/network/stream/intervals/store.go b/swarm/network/stream/intervals/store.go
new file mode 100644
index 0000000000..d5a3bb8437
--- /dev/null
+++ b/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 .
+
+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
+}
diff --git a/swarm/network/stream/intervals/store_test.go b/swarm/network/stream/intervals/store_test.go
new file mode 100644
index 0000000000..0b7344345f
--- /dev/null
+++ b/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 .
+
+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)
+ }
+}
diff --git a/swarm/network/stream/intervals_test.go b/swarm/network/stream/intervals_test.go
new file mode 100644
index 0000000000..bda123e7c0
--- /dev/null
+++ b/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 .
+
+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)
+}
diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go
index 63c8783fdf..0c2ffae6ef 100644
--- a/swarm/network/stream/messages.go
+++ b/swarm/network/stream/messages.go
@@ -26,12 +26,39 @@ 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
- Key []byte
- From, To uint64
- Priority uint8 // delivered on priority channel
+ Stream Stream
+ History *Range `rlp:"nil"`
+ Priority uint8 // delivered on priority channel
}
func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) (err error) {
@@ -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
- Key []byte
+ Stream Stream
}
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
- Key []byte // subtype or key
+ Stream Stream // name of Stream
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
- sk += keyToString(req.Key)
- s, err := p.getClient(sk)
+ c, _, err := p.getOrSetClient(req.Stream, req.From, req.To)
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 {
- s.sessionAt = req.From
+ if c.stream.Live {
+ c.sessionAt = req.From
}
- from, to := s.nextBatch(req.To)
- log.Trace("received offered batch", "peer", p.ID(), "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To)
+ from, to := c.nextBatch(req.To + 1)
+ 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)
- err := p.SendPriority(msg, s.priority)
+ log.Trace("sending want batch", "peer", p.ID(), "stream", msg.Stream, "from", msg.From, "to", msg.To)
+ 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
- Key []byte // subtype or key
+ Stream Stream
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)
- s, err := p.getServer(req.Stream + keyToString(req.Key))
+ log.Trace("received wanted batch", "peer", p.ID(), "stream", req.Stream, "from", req.From, "to", req.To)
+ 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
}
diff --git a/swarm/network/stream/peer.go b/swarm/network/stream/peer.go
index c1e64bd740..c9b623e792 100644
--- a/swarm/network/stream/peer.go
+++ b/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 (
- errServerNotFound = errors.New("server not found")
- errClientNotFound = errors.New("client not found")
-)
+type notFoundError struct {
+ t string
+ s Stream
+}
+
+func newNotFoundError(t string, s Stream) *notFoundError {
+ return ¬FoundError{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,21 +50,26 @@ 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
- quit chan struct{}
+ // 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{}
}
// NewPeer is the constructor for Peer
func NewPeer(peer *protocols.Peer, streamer *Registry) *Peer {
p := &Peer{
- Peer: peer,
- pq: pq.New(int(PriorityQueue), PriorityQueueCap),
- streamer: streamer,
- servers: make(map[string]*server),
- clients: make(map[string]*client),
- quit: make(chan struct{}),
+ Peer: peer,
+ pq: pq.New(int(PriorityQueue), PriorityQueueCap),
+ 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())
go p.pq.Run(ctx, func(i interface{}) { p.Send(i) })
@@ -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) {
- 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) {
+func (p *Peer) setServer(s Stream, 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)
- if p.clients[sk] != nil {
- return fmt.Errorf("client %v already registered", sk)
+ c = p.clients[sk]
+ if c != nil {
+ 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
- // if !live {
- // 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,
+ c = &client{
+ Client: is,
+ stream: s,
+ priority: cp.priority,
+ to: cp.to,
+ next: next,
+ intervalsStore: p.streamer.intervalsStore,
+ intervalsKey: intervalsKey,
}
- next <- nil // this is to allow wantedKeysMsg before first batch arrives
+ 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)
- client, ok := p.clients[sk]
- if !ok {
- return errClientNotFound
+ sk := s.String()
+ if p.clients[sk] != nil {
+ return fmt.Errorf("client %v already exists", sk)
}
- 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
}
diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go
index 24925edd7e..d0e230eb37 100644
--- a/swarm/network/stream/stream.go
+++ b/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"
)
@@ -45,43 +45,45 @@ const (
// Registry registry for outgoing and incoming streamer constructors
type Registry struct {
- api *API
- addr *network.BzzAddr
- skipCheck bool
- clientMu sync.RWMutex
- serverMu sync.RWMutex
- peersMu sync.RWMutex
- serverFuncs map[string]func(*Peer, []byte) (Server, error)
- clientFuncs map[string]func(*Peer, []byte) (Client, error)
- peers map[discover.NodeID]*Peer
- delivery *Delivery
- store storage.ChunkStore
+ api *API
+ addr *network.BzzAddr
+ skipCheck bool
+ clientMu sync.RWMutex
+ serverMu sync.RWMutex
+ peersMu sync.RWMutex
+ serverFuncs map[string]func(*Peer, []byte, bool) (Server, 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)),
- clientFuncs: make(map[string]func(*Peer, []byte) (Client, error)),
- peers: make(map[discover.NodeID]*Peer),
- delivery: delivery,
+ addr: addr,
+ skipCheck: skipCheck,
+ store: store,
+ serverFuncs: make(map[string]func(*Peer, []byte, bool) (Server, 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 {
- f, err := r.GetClientFunc(s)
- if err != nil {
+func (r *Registry) Subscribe(peerId discover.NodeID, s Stream, h *Range, priority uint8) error {
+ // check if the stream is registered
+ 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)
- if err != nil {
- return err
+ var to uint64
+ if !s.Live && h != nil {
+ to = h.To
}
- err = peer.setClient(s, t, is, priority, live)
+
+ err := peer.setClientParams(s, newClientParams(priority, to))
if err != nil {
return err
}
+ 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,
- // Live: live,
- From: from,
- To: to,
+ Stream: s,
+ History: h,
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
- stream string
- key []byte
+ to uint64
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.live {
- if len(intervals) == 0 {
- intervals = []uint64{c.sessionAt, from}
- } else {
- intervals[1] = from
+ if c.to > 0 && from >= c.to {
+ return 0, 0
+ }
+ if c.stream.Live {
+ return from, 0
+ } else if from >= c.sessionAt {
+ if c.to > 0 {
+ return from, c.to
}
- nextFrom = from
- } else if from >= c.sessionAt { // history sync complete
- intervals = nil
- nextFrom = from
- nextTo = math.MaxUint64
- } else if len(intervals) > 2 && from >= intervals[2] { // filled a gap in the intervals
- intervals = append(intervals[:1], intervals[3:]...)
- nextFrom = intervals[1]
- if len(intervals) > 2 {
- nextTo = intervals[2]
- } else {
- nextTo = c.sessionAt
- }
- } else {
- nextFrom = from
- intervals[1] = from
+ return from, math.MaxUint64
+ }
+ nextFrom, nextTo, err := c.NextInterval()
+ if err != nil {
+ log.Error("next intervals", "stream", c.stream)
+ return
+ }
+ if nextTo > c.to {
+ nextTo = c.to
+ }
+ if nextTo == 0 {
nextTo = c.sessionAt
}
- // b.intervals.set(intervals)
- return nextFrom, nextTo
+ return
}
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) {
- 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) SubscribeStream(peerId discover.NodeID, s Stream, history *Range, priority uint8) error {
+ return api.streamer.Subscribe(peerId, s, history, priority)
}
-func (api *API) ReadAll(hash common.Hash) (int64, error) {
- return readAll(api.dpa, hash[:])
-}
-
-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)
+func (api *API) UnsubscribeStream(peerId discover.NodeID, s Stream) error {
+ return api.streamer.Unsubscribe(peerId, s)
}
diff --git a/swarm/network/stream/streamer_test.go b/swarm/network/stream/streamer_test.go
index a2aabc7f82..44175b254f 100644
--- a/swarm/network/stream/streamer_test.go
+++ b/swarm/network/stream/streamer_test.go
@@ -32,53 +32,70 @@ 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)
}
}
var (
- hash0 = sha3.Sum256([]byte{0})
- hash1 = sha3.Sum256([]byte{1})
- 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)
+ hash0 = sha3.Sum256([]byte{0})
+ hash1 = sha3.Sum256([]byte{1})
+ hash2 = sha3.Sum256([]byte{2})
+ hashesTmp = append(hash0[:], hash1[:]...)
+ hashes = append(hashesTmp, hash2[:]...)
)
type testClient struct {
- t []byte
+ t []byte
+ wait0 chan bool
+ wait2 chan bool
+ batchDone chan bool
+ receivedHashes map[string][]byte
}
-type testServer struct {
- t []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() {
- receivedHashes[string(hash)] = hash
+ self.receivedHashes[string(hash)] = hash
if bytes.Equal(hash, hash0[:]) {
return func() {
- <-wait0
+ <-self.wait0
}
} else if bytes.Equal(hash, hash2[:]) {
return func() {
- <-wait2
+ <-self.wait2
}
}
return nil
}
-func (self *testClient) BatchDone(string, uint64, []byte, []byte) func() (*TakeoverProof, error) {
- close(batchDone)
+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 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
}
@@ -97,41 +114,73 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
t.Fatal(err)
}
- streamer.RegisterClientFunc("foo", func(p *Peer, t []byte) (Client, error) {
- return &testClient{
- t: t,
- }, nil
+ streamer.RegisterClientFunc("foo", func(p *Peer, t []byte, live bool) (Client, error) {
+ return newTestClient(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{
- Label: "Subscribe message",
- Expects: []p2ptest.Expect{
- {
- Code: 4,
- Msg: &SubscribeMsg{
- Stream: "foo",
- Key: []byte{},
- From: 5,
- To: 8,
- Priority: Top,
+ err = tester.TestExchanges(
+ p2ptest.Exchange{
+ Label: "Subscribe message",
+ Expects: []p2ptest.Expect{
+ {
+ Code: 4,
+ Msg: &SubscribeMsg{
+ Stream: stream,
+ History: &Range{
+ From: 5,
+ To: 8,
+ },
+ Priority: Top,
+ },
+ Peer: peerID,
},
- 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",
- Key: []byte{},
+ Stream: stream,
},
Peer: peerID,
},
@@ -162,10 +210,10 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
t.Fatal(err)
}
- streamer.RegisterServerFunc("foo", func(p *Peer, t []byte) (Server, error) {
- return &testServer{
- t: t,
- }, nil
+ stream := NewStream("foo", nil, false)
+
+ streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) {
+ return newTestServer(t), nil
})
peerID := tester.IDs[0]
@@ -176,10 +224,11 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
{
Code: 4,
Msg: &SubscribeMsg{
- Stream: "foo",
- Key: []byte{},
- From: 5,
- To: 8,
+ Stream: stream,
+ 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",
- Key: []byte{},
+ Stream: stream,
HandoverProof: &HandoverProof{
Handover: &Handover{},
},
@@ -213,8 +261,73 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
{
Code: 0,
Msg: &UnsubscribeMsg{
- Stream: "foo",
- Key: []byte{},
+ Stream: stream,
+ },
+ 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) {
- return &testServer{
- t: t,
- }, nil
+ streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) {
+ return newTestServer(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",
- Key: []byte{},
- From: 5,
- To: 8,
+ Stream: stream,
+ 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) {
- return &testClient{
- t: t,
- }, nil
+ stream := NewStream("foo", nil, true)
+
+ var tc *testClient
+
+ 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",
- Key: []byte{},
- From: 5,
- To: 8,
+ Stream: stream,
+ 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 {
- t.Fatalf("Expected number of received hashes %v, got %v", 3, len(receivedHashes))
+ if len(tc.receivedHashes) != 3 {
+ 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")
}
diff --git a/swarm/network/stream/syncer.go b/swarm/network/stream/syncer.go
index 605429693d..ded9163b89 100644
--- a/swarm/network/stream/syncer.go
+++ b/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(false, po, db)
+ return NewSwarmSyncerServer(live, 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,11 +240,10 @@ func (s *SwarmSyncerClient) TakeoverProof(streamName string, from uint64, hashes
}
s.end += uint64(len(hashes)) / HashSize
takeover := &Takeover{
- Stream: streamName,
- // Key: s.Key,
- Start: s.start,
- End: s.end,
- Root: root,
+ Stream: stream,
+ Start: s.start,
+ End: s.end,
+ Root: root,
}
// serialise and sign
return &TakeoverProof{
diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go
index 480bf61eaa..938c33d98c 100644
--- a/swarm/network/stream/syncer_test.go
+++ b/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
diff --git a/swarm/network/stream/testing/testing.go b/swarm/network/stream/testing/testing.go
index 39b2c1df1d..25987f7330 100644
--- a/swarm/network/stream/testing/testing.go
+++ b/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()
diff --git a/swarm/swarm.go b/swarm/swarm.go
index b8b08b6079..91ec805511 100644
--- a/swarm/swarm.go
+++ b/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)