diff --git a/swarm/network/stream/common_test.go b/swarm/network/stream/common_test.go
index 46d6314624..102faa8287 100644
--- a/swarm/network/stream/common_test.go
+++ b/swarm/network/stream/common_test.go
@@ -88,18 +88,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, 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)
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
index 4a53e1a9e7..5fd820da87 100644
--- a/swarm/network/stream/intervals/intervals.go
+++ b/swarm/network/stream/intervals/intervals.go
@@ -17,7 +17,9 @@
package intervals
import (
+ "bytes"
"fmt"
+ "strconv"
"sync"
)
@@ -152,3 +154,53 @@ func (i *Intervals) Last() (end uint64) {
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/store.go b/swarm/network/stream/intervals/store.go
index 87b06c7ed4..9b56f8b585 100644
--- a/swarm/network/stream/intervals/store.go
+++ b/swarm/network/stream/intervals/store.go
@@ -14,7 +14,6 @@
// 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 (
@@ -33,6 +32,7 @@ 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
@@ -82,3 +82,8 @@ func (s *MemStore) Delete(key string) (err error) {
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
index 9a30b5d2e0..0b7344345f 100644
--- a/swarm/network/stream/intervals/store_test.go
+++ b/swarm/network/stream/intervals/store_test.go
@@ -20,8 +20,11 @@ import "testing"
// TestMemStore tests basic functionality of MemStore.
func TestMemStore(t *testing.T) {
- s := NewMemStore()
+ 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)
diff --git a/swarm/network/stream/peer.go b/swarm/network/stream/peer.go
index 65dbefea5d..c9b51c43ba 100644
--- a/swarm/network/stream/peer.go
+++ b/swarm/network/stream/peer.go
@@ -240,6 +240,7 @@ func (p *Peer) setClientNolock(s Stream, from, to uint64) (c *client, err error)
Client: is,
stream: s,
priority: cp.priority,
+ to: to,
next: next,
intervalsStore: p.streamer.intervalsStore,
intervalsKey: intervalsKey,
diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go
index 85c9a79355..90bf23c641 100644
--- a/swarm/network/stream/stream.go
+++ b/swarm/network/stream/stream.go
@@ -190,6 +190,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()
@@ -286,6 +291,7 @@ type client struct {
stream Stream
priority uint8
sessionAt uint64
+ to uint64
next chan error
intervalsKey string
@@ -348,7 +354,13 @@ func (c *client) batchDone(p *Peer, req *OfferedHashesMsg, hashes []byte) error
if err := c.AddInterval(tp.Takeover.Start, tp.Takeover.End); err != nil {
return err
}
- return p.SendPriority(tp, c.priority)
+ 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
}