diff --git a/swarm/network/stream/common_test.go b/swarm/network/stream/common_test.go
index 1deb6ffba9..46d6314624 100644
--- a/swarm/network/stream/common_test.go
+++ b/swarm/network/stream/common_test.go
@@ -30,6 +30,7 @@ import (
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
"github.com/ethereum/go-ethereum/swarm/network"
+ "github.com/ethereum/go-ethereum/swarm/network/stream/intervals"
"github.com/ethereum/go-ethereum/swarm/storage"
)
@@ -68,7 +69,7 @@ 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() {
@@ -98,7 +99,7 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *stora
db := storage.NewDBAPI(localStore)
delivery := NewDelivery(to, db)
- streamer := NewRegistry(addr, delivery, localStore, defaultSkipCheck)
+ streamer := NewRegistry(addr, delivery, localStore, intervals.NewMemStore(), defaultSkipCheck)
protocolTester := p2ptest.NewProtocolTester(t, network.NewNodeIDFromAddr(addr), 1, streamer.runProtocol)
err = waitForPeers(streamer, 1*time.Second, 1)
diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go
index 3ef991158e..53101b4329 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 183ea2b9e9..b51e48dde9 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,
})
@@ -172,9 +174,9 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) {
Hashes: hash,
From: 0,
// TODO: why is this 32???
- To: 32,
- Key: []byte{},
- Stream: swarmChunkServerStreamName,
+ To: 32,
+ Stream: NewStream(swarmChunkServerStreamName, nil, false),
+ Initial: true,
},
Peer: peerID,
},
@@ -227,7 +229,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 +237,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 +262,11 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) {
p2ptest.Expect{
Code: 4,
Msg: &SubscribeMsg{
- Stream: "foo",
- Key: []byte{},
- From: 5,
- To: 8,
+ Stream: stream,
+ History: &Range{
+ From: 5,
+ To: 8,
+ },
Priority: Top,
},
Peer: peerID,
@@ -388,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
@@ -561,7 +565,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/intervals.go b/swarm/network/stream/intervals/intervals.go
new file mode 100644
index 0000000000..4a53e1a9e7
--- /dev/null
+++ b/swarm/network/stream/intervals/intervals.go
@@ -0,0 +1,154 @@
+// 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 (
+ "fmt"
+ "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)
+}
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..87b06c7ed4
--- /dev/null
+++ b/swarm/network/stream/intervals/store.go
@@ -0,0 +1,84 @@
+// 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 TODO: implement LevelDB based Store.
+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 persistance 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)
+}
+
+// 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
+}
diff --git a/swarm/network/stream/intervals/store_test.go b/swarm/network/stream/intervals/store_test.go
new file mode 100644
index 0000000000..9a30b5d2e0
--- /dev/null
+++ b/swarm/network/stream/intervals/store_test.go
@@ -0,0 +1,69 @@
+// 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) {
+ s := NewMemStore()
+
+ 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/messages.go b/swarm/network/stream/messages.go
index 22592d288c..dbc3fc24d3 100644
--- a/swarm/network/stream/messages.go
+++ b/swarm/network/stream/messages.go
@@ -23,14 +23,42 @@ import (
"github.com/ethereum/go-ethereum/log"
bv "github.com/ethereum/go-ethereum/swarm/network/bitvector"
+ "github.com/ethereum/go-ethereum/swarm/network/stream/intervals"
"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
+ Stream Stream
+ History *Range
Priority uint8 // delivered on priority channel
}
@@ -45,24 +73,58 @@ 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, true); err != nil {
p.Drop(err)
}
}()
+
+ if req.Stream.Live && req.History != nil {
+ // subscribe to the history stream as well
+ s, err := f(p, req.Stream.Key, false)
+ if err != nil {
+ return err
+ }
+ historyStream := NewStream(req.Stream.Name, req.Stream.Key, false)
+ priority := req.Priority
+ if priority > 0 {
+ // decrement history stream priority
+ priority--
+ }
+ os, err := p.setServer(historyStream, s, priority)
+ if err != nil {
+ return err
+ }
+ go func() {
+ if err := p.SendOfferedHashes(os, req.History.From, req.History.To, true); err != nil {
+ p.Drop(err)
+ }
+ }()
+ }
+
return nil
}
@@ -75,23 +137,22 @@ 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
+ Initial bool
+ *HandoverProof // HandoverProof
}
// String pretty prints OfferedHashesMsg
@@ -102,9 +163,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.getClient(req.Stream)
if err != nil {
return err
}
@@ -117,7 +176,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 +201,27 @@ 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
+ if req.Initial {
+ // create initial intervals for live stream starting from the first From value
+ if err := c.intervalsStore.Put(peerStreamIntervalsKey(p, req.Stream), intervals.NewIntervals(req.From)); err != nil {
+ return err
+ }
+ }
}
- 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)
+ 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 +231,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 +249,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,15 +263,15 @@ 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
}
hashes := s.currentBatch
// launch in go routine since GetBatch blocks until new hashes arrive
go func() {
- if err := p.SendOfferedHashes(s, req.From, req.To); err != nil {
+ if err := p.SendOfferedHashes(s, req.From, req.To, false); err != nil {
p.Drop(err)
}
}()
@@ -237,7 +300,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 12810789d9..35375f5c34 100644
--- a/swarm/network/stream/peer.go
+++ b/swarm/network/stream/peer.go
@@ -26,6 +26,7 @@ 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"
)
@@ -84,7 +85,7 @@ func (p *Peer) SendPriority(msg interface{}, priority uint8) error {
}
// SendOfferedHashes sends OfferedHashesMsg protocol msg
-func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error {
+func (p *Peer) SendOfferedHashes(s *server, f, t uint64, initial bool) error {
hashes, from, to, proof, err := s.SetNextBatch(f, t)
if err != nil {
return err
@@ -105,57 +106,56 @@ func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error {
From: from,
To: to,
Stream: s.stream,
- Key: s.key,
+ Initial: initial,
}
- 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) getClient(s Stream) (*client, error) {
p.clientMu.RLock()
defer p.clientMu.RUnlock()
- client := p.clients[s]
+ client := p.clients[s.String()]
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
@@ -165,39 +165,63 @@ func (p *Peer) removeServer(s string, key []byte) error {
return nil
}
-func (p *Peer) setClient(s string, key []byte, i Client, priority uint8, live bool) error {
+func (p *Peer) setClient(s Stream, i Client, priority uint8, intervalsStore intervals.Store) error {
p.clientMu.Lock()
defer p.clientMu.Unlock()
- sk := s + keyToString(key)
+ sk := s.String()
if p.clients[sk] != nil {
return fmt.Errorf("client %v already registered", sk)
}
+
+ 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 := intervalsStore.Get(historyKey)
+ switch err {
+ case nil:
+ liveIntervals, err := intervalsStore.Get(intervalsKey)
+ switch err {
+ case nil:
+ historyIntervals.Merge(liveIntervals)
+ if err := 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)
+ }
+ } else {
+ // create intervals for history stream
+ // live stream can create intervals when the first sessionAt is known
+ if err := intervalsStore.Put(intervalsKey, intervals.NewIntervals(0)); err != nil {
+ return 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,
+ Client: i,
+ stream: s,
+ priority: priority,
+ next: next,
+ intervalsStore: intervalsStore,
+ intervalsKey: intervalsKey,
}
next <- nil // this is to allow wantedKeysMsg before first batch arrives
return nil
}
-func (p *Peer) removeClient(s string, key []byte) error {
+func (p *Peer) removeClient(s Stream) error {
p.clientMu.Lock()
defer p.clientMu.Unlock()
- sk := s + keyToString(key)
- client, ok := p.clients[sk]
+ client, ok := p.clients[s.String()]
if !ok {
return errClientNotFound
}
diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go
index 5770f679c3..17838e7a5c 100644
--- a/swarm/network/stream/stream.go
+++ b/swarm/network/stream/stream.go
@@ -30,6 +30,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 +46,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 +92,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 +100,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 +112,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,8 +124,8 @@ 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)
+func (r *Registry) Subscribe(peerId discover.NodeID, s Stream, h *Range, priority uint8) error {
+ f, err := r.GetClientFunc(s.Name)
if err != nil {
return err
}
@@ -132,29 +135,42 @@ 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)
+ is, err := f(peer, s.Key, s.Live)
if err != nil {
return err
}
- err = peer.setClient(s, t, is, priority, live)
+ err = peer.setClient(s, is, priority, r.intervalsStore)
if err != nil {
return err
}
+ if s.Live && h != nil {
+ is, err := f(peer, s.Key, false)
+ if err != nil {
+ return err
+ }
+ p := priority
+ if p > 0 {
+ p--
+ }
+ historyStream := NewStream(s.Name, s.Key, false)
+ err = peer.setClient(historyStream, is, p, r.intervalsStore)
+ if 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 +178,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 {
@@ -261,20 +276,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]), uint8(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 +292,59 @@ type Server interface {
type client struct {
Client
+ stream Stream
priority uint8
sessionAt uint64
- live bool
- 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.live {
- if len(intervals) == 0 {
- intervals = []uint64{c.sessionAt, from}
- } else {
- intervals[1] = from
- }
- 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
+ if c.stream.Live {
+ return from, 0
+ } else if from >= c.sessionAt {
+ return from, math.MaxUint64
+ }
+ nextFrom, nextTo, err := c.NextInterval()
+ if err != nil {
+ log.Error("next intervals", "stream", c.stream)
+ return
+ }
+ 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,6 +353,10 @@ func (c *client) batchDone(p *Peer, req *OfferedHashesMsg, hashes []byte) error
if err != nil {
return err
}
+ // 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
+ }
return p.SendPriority(tp, c.priority)
}
return nil
@@ -399,6 +418,10 @@ func (r *Registry) Stop() error {
return nil
}
+type Range struct {
+ From, To uint64
+}
+
type API struct {
streamer *Registry
dpa *storage.DPA
@@ -432,10 +455,10 @@ 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) SubscribeStream(peerId discover.NodeID, s Stream, history *Range, priority uint8) error {
+ return api.streamer.Subscribe(peerId, s, history, priority)
}
-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 951e008a53..da27f5907c 100644
--- a/swarm/network/stream/streamer_test.go
+++ b/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)
}
@@ -72,7 +73,7 @@ func (self *testClient) NeedData(hash []byte) func() {
return nil
}
-func (self *testClient) BatchDone(string, uint64, []byte, []byte) func() (*TakeoverProof, error) {
+func (self *testClient) BatchDone(Stream, uint64, []byte, []byte) func() (*TakeoverProof, error) {
close(batchDone)
return nil
}
@@ -97,7 +98,7 @@ 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{
t: t,
}, nil
@@ -105,7 +106,8 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(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)
}
@@ -116,10 +118,11 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
p2ptest.Expect{
Code: 4,
Msg: &SubscribeMsg{
- Stream: "foo",
- Key: []byte{},
- From: 5,
- To: 8,
+ Stream: stream,
+ History: &Range{
+ From: 5,
+ To: 8,
+ },
Priority: Top,
},
Peer: peerID,
@@ -131,7 +134,7 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
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 +145,7 @@ func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
p2ptest.Expect{
Code: 0,
Msg: &UnsubscribeMsg{
- Stream: "foo",
- Key: []byte{},
+ Stream: stream,
},
Peer: peerID,
},
@@ -162,7 +164,9 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
t.Fatal(err)
}
- streamer.RegisterServerFunc("foo", func(p *Peer, t []byte) (Server, error) {
+ stream := NewStream("foo", nil, false)
+
+ streamer.RegisterServerFunc("foo", func(p *Peer, t []byte, live bool) (Server, error) {
return &testServer{
t: t,
}, nil
@@ -176,10 +180,11 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
p2ptest.Trigger{
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,14 +194,14 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
p2ptest.Expect{
Code: 1,
Msg: &OfferedHashesMsg{
- Stream: "foo",
- Key: []byte{},
+ Stream: stream,
HandoverProof: &HandoverProof{
Handover: &Handover{},
},
- Hashes: make([]byte, HashSize),
- From: 6,
- To: 9,
+ Hashes: make([]byte, HashSize),
+ From: 6,
+ To: 9,
+ Initial: true,
},
Peer: peerID,
},
@@ -213,8 +218,7 @@ func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) {
p2ptest.Trigger{
Code: 0,
Msg: &UnsubscribeMsg{
- Stream: "foo",
- Key: []byte{},
+ Stream: stream,
},
Peer: peerID,
},
@@ -233,12 +237,14 @@ 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{
t: t,
}, nil
})
+ stream := NewStream("bar", nil, true)
+
peerID := tester.IDs[0]
err = tester.TestExchanges(p2ptest.Exchange{
@@ -247,10 +253,11 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) {
p2ptest.Trigger{
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 +279,78 @@ func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) {
}
}
+// TODO: fix: tests with TestExchanges are inconsistent because Expects check
+// ordering is not guarrantied but fails if the order is wrong.
+// 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,
+// Initial: true,
+// },
+// Peer: peerID,
+// },
+// {
+// Code: 1,
+// Msg: &OfferedHashesMsg{
+// Stream: stream,
+// HandoverProof: &HandoverProof{
+// Handover: &Handover{},
+// },
+// From: 1,
+// To: 1,
+// Hashes: make([]byte, HashSize),
+// Initial: true,
+// },
+// Peer: peerID,
+// },
+// },
+// })
+
+// if err != nil {
+// t.Fatal(err)
+// }
+// }
+
func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
tester, streamer, _, teardown, err := newStreamerTester(t)
defer teardown()
@@ -279,7 +358,9 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
t.Fatal(err)
}
- streamer.RegisterClientFunc("foo", func(p *Peer, t []byte) (Client, error) {
+ stream := NewStream("foo", nil, true)
+
+ streamer.RegisterClientFunc("foo", func(p *Peer, t []byte, live bool) (Client, error) {
return &testClient{
t: t,
}, nil
@@ -287,7 +368,7 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
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 +379,11 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
p2ptest.Expect{
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 +402,7 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
Hashes: hashes,
From: 5,
To: 8,
- Stream: "foo",
+ Stream: stream,
},
Peer: peerID,
},
@@ -329,7 +411,7 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
p2ptest.Expect{
Code: 2,
Msg: &WantedHashesMsg{
- Stream: "foo",
+ Stream: stream,
Want: []byte{5},
From: 8,
To: 0,
diff --git a/swarm/network/stream/syncer.go b/swarm/network/stream/syncer.go
index 6d8473afc9..252b7432ba 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 := uint8(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 58d780c36f..c8f4a0deca 100644
--- a/swarm/network/stream/syncer_test.go
+++ b/swarm/network/stream/syncer_test.go
@@ -160,7 +160,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