Fix swarm api tests and rename dpaChunkStore to NetStore

swarm/api/api_test.go

@@ -34,9 +34,8 @@ func testApi(t *testing.T, f func(*Api)) {
 	if err != nil {
 		t.Fatalf("unable to create temp dir: %v", err)
 	}
-	os.RemoveAll(datadir)
 	defer os.RemoveAll(datadir)
-	dpa, err := storage.NewLocalDPA(datadir)
+	dpa, err := storage.NewLocalDPA(datadir, make([]byte, 32))
 	if err != nil {
 		return
 	}
@@ -114,7 +113,7 @@ func TestApiPut(t *testing.T) {
 		if err != nil {
 			t.Fatalf("unexpected error: %v", err)
 		}
-		resp := testGet(t, api, key.String(), "")
+		resp := testGet(t, api, key.Hex(), "")
 		checkResponse(t, resp, exp)
 	})
 }

swarm/api/config_test.go

@@ -33,8 +33,8 @@ func TestConfig(t *testing.T) {
 		t.Fatalf("failed to load private key: %v", err)
 	}
 
-	one := NewDefaultConfig()
+	one := NewConfig()
-	two := NewDefaultConfig()
+	two := NewConfig()
 
 	if equal := reflect.DeepEqual(one, two); !equal {
 		t.Fatal("Two default configs are not equal")
@@ -55,14 +55,6 @@ func TestConfig(t *testing.T) {
 		t.Fatal("Failed to correctly initialize SwapParams")
 	}
 
-	if one.SyncParams.RequestDbPath == one.Path {
-		t.Fatal("Failed to correctly initialize SyncParams")
-	}
-
-	if one.HiveParams.KadDbPath == one.Path {
-		t.Fatal("Failed to correctly initialize HiveParams")
-	}
-
 	if one.StoreParams.ChunkDbPath == one.Path {
 		t.Fatal("Failed to correctly initialize StoreParams")
 	}

swarm/api/filesystem.go

@@ -116,7 +116,7 @@ func (self *FileSystem) Upload(lpath, index string) (string, error) {
 				var wait func()
 				hash, wait, err = self.api.dpa.Store(f, stat.Size())
 				if hash != nil {
-					list[i].Hash = hash.String()
+					list[i].Hash = hash.Hex()
 				}
 				wait()
 				awg.Done()
@@ -164,7 +164,7 @@ func (self *FileSystem) Upload(lpath, index string) (string, error) {
 	err2 := trie.recalcAndStore()
 	var hs string
 	if err2 == nil {
-		hs = trie.hash.String()
+		hs = trie.hash.Hex()
 	}
 	awg.Wait()
 	return hs, err2

swarm/api/filesystem_test.go

@@ -21,7 +21,6 @@ import (
 	"io/ioutil"
 	"os"
 	"path/filepath"
-	"sync"
 	"testing"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -105,9 +104,8 @@ func TestApiDirUploadModify(t *testing.T) {
 			t.Errorf("unexpected error: %v", err)
 			return
 		}
-		wg := &sync.WaitGroup{}
+		hash, wait, err := api.Store(bytes.NewReader(index), int64(len(index)))
-		hash, err := api.Store(bytes.NewReader(index), int64(len(index)), wg)
+		wait()
-		wg.Wait()
 		if err != nil {
 			t.Errorf("unexpected error: %v", err)
 			return
@@ -122,7 +120,7 @@ func TestApiDirUploadModify(t *testing.T) {
 			t.Errorf("unexpected error: %v", err)
 			return
 		}
-		bzzhash = key.String()
+		bzzhash = key.Hex()
 
 		content := readPath(t, "testdata", "test0", "index.html")
 		resp := testGet(t, api, bzzhash, "index2.html")

swarm/api/http/server_test.go

@@ -120,7 +120,7 @@ func TestBzzGetPath(t *testing.T) {
 			t.Fatalf("Read request body: %v", err)
 		}
 
-		if string(respbody) != key[v].String() {
+		if string(respbody) != key[v].Hex() {
 			isexpectedfailrequest := false
 
 			for _, r := range expectedfailrequests {

swarm/api/manifest.go

@@ -90,7 +90,7 @@ func (m *ManifestWriter) AddEntry(data io.Reader, e *ManifestEntry) (storage.Key
 		return nil, err
 	}
 	entry := newManifestTrieEntry(e, nil)
-	entry.Hash = key.String()
+	entry.Hash = key.Hex()
 	m.trie.addEntry(entry, m.quitC)
 	return key, nil
 }
@@ -338,7 +338,7 @@ func (self *manifestTrie) recalcAndStore() error {
 				if err != nil {
 					return err
 				}
-				entry.Hash = entry.subtrie.hash.String()
+				entry.Hash = entry.subtrie.hash.Hex()
 			}
 			list.Entries = append(list.Entries, entry.ManifestEntry)
 		}
@@ -351,7 +351,8 @@ func (self *manifestTrie) recalcAndStore() error {
 	}
 
 	sr := bytes.NewReader(manifest)
-	key, _, err2 := self.dpa.Store(sr, int64(len(manifest)))
+	key, wait, err2 := self.dpa.Store(sr, int64(len(manifest)))
+	wait()
 	self.hash = key
 	return err2
 }

swarm/api/storage.go

@@ -46,7 +46,7 @@ func (self *Storage) Put(content, contentType string) (string, error) {
 	if err != nil {
 		return "", err
 	}
-	return key.String(), err
+	return key.Hex(), err
 }
 
 // Get retrieves the content from bzzpath and reads the response in full
@@ -100,5 +100,5 @@ func (self *Storage) Modify(rootHash, path, contentHash, contentType string) (ne
 	if err != nil {
 		return "", err
 	}
-	return key.String(), nil
+	return key.Hex(), nil
 }

swarm/fuse/swarmfs_test.go

@@ -808,7 +808,7 @@ func TestFUSE(t *testing.T) {
 	}
 	os.RemoveAll(datadir)
 
-	dpa, err := storage.NewLocalDPA(datadir)
+	dpa, err := storage.NewLocalDPA(datadir, make([]byte, 32))
 	if err != nil {
 		t.Fatal(err)
 	}

swarm/network/protocol.go

@@ -143,7 +143,7 @@ func (b *Bzz) NodeInfo() interface{} {
 // * handshake/hive
 // * discovery
 func (b *Bzz) Protocols() []p2p.Protocol {
-	return []p2p.Protocol{
+	protocols := []p2p.Protocol{
 		{
 			Name:     BzzSpec.Name,
 			Version:  BzzSpec.Version,
@@ -159,15 +159,18 @@ func (b *Bzz) Protocols() []p2p.Protocol {
 			NodeInfo: b.Hive.NodeInfo,
 			PeerInfo: b.Hive.PeerInfo,
 		},
-		{
+	}
+	if b.Streamer != nil {
+		protocols = append(protocols, p2p.Protocol{
 			Name:     StreamerSpec.Name,
 			Version:  StreamerSpec.Version,
 			Length:   StreamerSpec.Length(),
 			Run:      b.RunProtocol(StreamerSpec, b.Streamer.Run),
 			NodeInfo: b.Streamer.NodeInfo,
 			PeerInfo: b.Streamer.PeerInfo,
-		},
+		})
 	}
+	return protocols
 }
 
 // APIs returns the APIs offered by bzz

swarm/network/request_test.go

@@ -326,7 +326,7 @@ func testDeliveryFromNodes(nodes, conns, size int, skipCheck bool) func(adapter
 			rrdpa := storage.NewDPA(newRoundRobinStore(localStores[1:]...), storage.NewChunkerParams())
 			rrdpa.Start()
 			// create a retriever dpa for the pivot node
-			dpacs := storage.NewDpaChunkStore(localStores[0].(*storage.LocalStore), func(chunk *storage.Chunk) error { return delivery.RequestFromPeers(chunk.Key[:], skipCheck) })
+			dpacs := storage.NewNetStore(localStores[0].(*storage.LocalStore), func(chunk *storage.Chunk) error { return delivery.RequestFromPeers(chunk.Key[:], skipCheck) })
 			dpa := storage.NewDPA(dpacs, storage.NewChunkerParams())
 			dpa.Start()
 			return func(context.Context) error {

swarm/network/simulations/discovery/discovery_test.go

@@ -319,5 +319,5 @@ func newService(ctx *adapters.ServiceContext) (node.Service, error) {
 		HiveParams:   hp,
 	}
 
-	return network.NewBzz(config, kad, nil), nil
+	return network.NewBzz(config, kad, nil, nil), nil
 }

swarm/network/streamer.go

@@ -147,6 +147,10 @@ func NewStreamer(delivery *Delivery) *Streamer {
 	return streamer
 }
 
+func (self *Streamer) Retrieve(chunk *storage.Chunk) error {
+	return self.delivery.RequestFromPeers(chunk.Key[:], false)
+}
+
 // RegisterIncomingStreamer registers an incoming streamer constructor
 func (self *Streamer) RegisterIncomingStreamer(stream string, f func(*StreamerPeer, []byte) (IncomingStreamer, error)) {
 	self.incomingLock.Lock()

swarm/pss/client/client_test.go

@@ -233,7 +233,7 @@ func newServices() adapters.Services {
 			if err != nil {
 				return nil, fmt.Errorf("create pss cache tmpdir failed", "error", err)
 			}
-			dpa, err := storage.NewLocalDPA(cachedir)
+			dpa, err := storage.NewLocalDPA(cachedir, make([]byte, 32))
 			if err != nil {
 				return nil, fmt.Errorf("local dpa creation failed", "error", err)
 			}
@@ -260,7 +260,7 @@ func newServices() adapters.Services {
 				UnderlayAddr: addr.Under(),
 				HiveParams:   hp,
 			}
-			return network.NewBzz(config, kademlia(ctx.Config.ID), stateStore), nil
+			return network.NewBzz(config, kademlia(ctx.Config.ID), stateStore, nil), nil
 		},
 	}
 }

swarm/pss/pss_test.go

@@ -1123,7 +1123,7 @@ func newServices() adapters.Services {
 			if err != nil {
 				return nil, fmt.Errorf("create pss cache tmpdir failed", "error", err)
 			}
-			dpa, err := storage.NewLocalDPA(cachedir)
+			dpa, err := storage.NewLocalDPA(cachedir, network.NewAddrFromNodeID(ctx.Config.ID).Over())
 			if err != nil {
 				return nil, fmt.Errorf("local dpa creation failed", "error", err)
 			}
@@ -1178,7 +1178,7 @@ func newServices() adapters.Services {
 				UnderlayAddr: addr.Under(),
 				HiveParams:   hp,
 			}
-			return network.NewBzz(config, kademlia(ctx.Config.ID), stateStore), nil
+			return network.NewBzz(config, kademlia(ctx.Config.ID), stateStore, nil), nil
 		},
 	}
 }
@@ -1195,7 +1195,7 @@ func newTestPss(privkey *ecdsa.PrivateKey, overlay network.Overlay, ppextra *Pss
 		log.Error("create pss cache tmpdir failed", "error", err)
 		os.Exit(1)
 	}
-	dpa, err := storage.NewLocalDPA(cachedir)
+	dpa, err := storage.NewLocalDPA(cachedir, addr.Over())
 	if err != nil {
 		log.Error("local dpa creation failed", "error", err)
 		os.Exit(1)

swarm/storage/chunker_test.go

@@ -64,7 +64,7 @@ func (self *chunkerTester) Split(chunker Splitter, data io.Reader, size int64, c
 					return nil
 				case chunk := <-chunkC:
 					// self.chunks = append(self.chunks, chunk)
-					self.chunks[chunk.Key.String()] = chunk
+					self.chunks[chunk.Key.Hex()] = chunk
 					close(chunk.dbStored)
 				}
 
@@ -101,10 +101,10 @@ func (self *chunkerTester) Append(chunker Splitter, rootKey Key, data io.Reader,
 					return nil
 				case chunk := <-chunkC:
 					if chunk != nil {
-						stored, success := self.chunks[chunk.Key.String()]
+						stored, success := self.chunks[chunk.Key.Hex()]
 						if !success {
 							// Requesting data
-							self.chunks[chunk.Key.String()] = chunk
+							self.chunks[chunk.Key.Hex()] = chunk
 							close(chunk.dbStored)
 						} else {
 							// getting data
@@ -151,7 +151,7 @@ func (self *chunkerTester) Join(chunker Chunker, key Key, c int, chunkC chan *Ch
 					return nil
 				}
 				// this just mocks the behaviour of a chunk store retrieval
-				stored, success := self.chunks[chunk.Key.String()]
+				stored, success := self.chunks[chunk.Key.Hex()]
 				if !success {
 					return errors.New("Not found")
 				}

swarm/storage/dpa.go

@@ -17,7 +17,6 @@
 package storage
 
 import (
-	"encoding/binary"
 	"errors"
 	"fmt"
 	"io"
@@ -50,6 +49,9 @@ const (
 
 var (
 	notFound = errors.New("not found")
+
+	// timeout interval before retrieval is timed out
+	searchTimeout = 3 * time.Second
 )
 
 type DPA struct {
@@ -173,54 +175,3 @@ func (self *DPA) storeWorker() {
 		}
 	}
 }
-
-// DpaChunkStore implements the ChunkStore interface,
-// this chunk access layer assumed 2 chunk stores
-// local storage eg. LocalStore and network storage eg., NetStore
-// access by calling network is blocking with a timeout
-
-type dpaChunkStore struct {
-	localStore *LocalStore
-	retrieve   func(chunk *Chunk) error
-}
-
-func NewDpaChunkStore(localStore *LocalStore, retrieve func(chunk *Chunk) error) *dpaChunkStore {
-	return &dpaChunkStore{localStore, retrieve}
-}
-
-// Get is the entrypoint for local retrieve requests
-// waits for response or times out
-func (self *dpaChunkStore) Get(key Key) (chunk *Chunk, err error) {
-	var created bool
-	chunk, created = self.localStore.GetOrCreateRequest(key)
-	if chunk.ReqC == nil {
-		log.Trace(fmt.Sprintf("DPA.Get: %v found locally, %d bytes", key.Log(), len(chunk.SData)))
-		return
-	}
-
-	if created {
-		if err := self.retrieve(chunk); err != nil {
-			return nil, err
-		}
-	}
-	t := time.NewTicker(searchTimeout)
-	defer t.Stop()
-
-	select {
-	case <-t.C:
-		log.Trace(fmt.Sprintf("DPA.Get: %v request time out ", key.Log()))
-		return nil, notFound
-	case <-chunk.ReqC:
-	}
-	chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8]))
-	return chunk, nil
-}
-
-// Put is the entrypoint for local store requests coming from storeLoop
-func (self *dpaChunkStore) Put(chunk *Chunk) {
-	self.localStore.Put(chunk)
-}
-
-// Close chunk store
-func (self *dpaChunkStore) Close() {
-}

swarm/storage/localstore.go

@@ -19,10 +19,32 @@ package storage
 import (
 	"encoding/binary"
 	"fmt"
+	"path/filepath"
 
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/swarm/storage/mock"
 )
 
+type StoreParams struct {
+	ChunkDbPath   string
+	DbCapacity    uint64
+	CacheCapacity uint
+}
+
+//create params with default values
+func NewDefaultStoreParams() (self *StoreParams) {
+	return &StoreParams{
+		DbCapacity:    defaultDbCapacity,
+		CacheCapacity: defaultCacheCapacity,
+	}
+}
+
+//this can only finally be set after all config options (file, cmd line, env vars)
+//have been evaluated
+func (self *StoreParams) Init(path string) {
+	self.ChunkDbPath = filepath.Join(path, "chunks")
+}
+
 // LocalStore is a combination of inmemory db over a disk persisted db
 // implements a Get/Put with fallback (caching) logic using any 2 ChunkStores
 type LocalStore struct {
@@ -31,8 +53,8 @@ type LocalStore struct {
 }
 
 // This constructor uses MemStore and DbStore as components
-func NewLocalStore(hash SwarmHasher, params *StoreParams, basekey []byte) (*LocalStore, error) {
+func NewLocalStore(hash SwarmHasher, params *StoreParams, basekey []byte, mockStore *mock.NodeStore) (*LocalStore, error) {
-	dbStore, err := NewDbStore(params.ChunkDbPath, hash, params.DbCapacity, func(k Key) (ret uint8) { return uint8(Proximity(basekey[:], k[:])) })
+	dbStore, err := NewMockDbStore(params.ChunkDbPath, hash, params.DbCapacity, func(k Key) (ret uint8) { return uint8(Proximity(basekey[:], k[:])) }, mockStore)
 	if err != nil {
 		return nil, err
 	}

swarm/storage/netstore.go

@@ -17,116 +17,58 @@
 package storage
 
 import (
-	"path/filepath"
+	"encoding/binary"
+	"fmt"
 	"time"
+
+	"github.com/ethereum/go-ethereum/log"
 )
 
-// import (
+// NetStore implements the ChunkStore interface,
-// 	"fmt"
+// this chunk access layer assumed 2 chunk stores
-// 	"path/filepath"
+// local storage eg. LocalStore and network storage eg., NetStore
-// 	"time"
+// access by calling network is blocking with a timeout
-
+type NetStore struct {
-// 	"github.com/ethereum/go-ethereum/log"
+	localStore *LocalStore
-// )
+	retrieve   func(chunk *Chunk) error
-
-// /*
-// NetStore is a cloud storage access abstaction layer for swarm
-// it contains the shared logic of network served chunk store/retrieval requests
-// both local (coming from DPA api) and remote (coming from peers via bzz protocol)
-// it implements the ChunkStore interface and embeds LocalStore
-
-// It is called by the bzz protocol instances via Depo (the store/retrieve request handler)
-// a protocol instance is running on each peer, so this is heavily parallelised.
-// NetStore falls back to a backend (CloudStorage interface)
-// implemented by bzz/network/forwarder. forwarder or IPFS or IPΞS
-// */
-// type NetStore struct {
-// 	hashfunc   SwarmHasher
-// 	localStore *LocalStore
-// 	cloud      CloudStore
-// }
-
-// // backend engine for cloud store
-// // It can be aggregate dispatching to several parallel implementations:
-// // bzz/network/forwarder. forwarder or IPFS or IPΞS
-// type CloudStore interface {
-// 	Store(*Chunk)
-// 	Deliver(*Chunk)
-// 	Retrieve(*Chunk)
-// }
-
-type StoreParams struct {
-	ChunkDbPath   string
-	DbCapacity    uint64
-	CacheCapacity uint
-	Radius        int
 }
 
-//create params with default values
+func NewNetStore(localStore *LocalStore, retrieve func(chunk *Chunk) error) *NetStore {
-func NewDefaultStoreParams() (self *StoreParams) {
+	return &NetStore{localStore, retrieve}
-	return &StoreParams{
+}
-		DbCapacity:    defaultDbCapacity,
+
-		CacheCapacity: defaultCacheCapacity,
+// Get is the entrypoint for local retrieve requests
-		Radius:        defaultRadius,
+// waits for response or times out
+func (self *NetStore) Get(key Key) (chunk *Chunk, err error) {
+	var created bool
+	chunk, created = self.localStore.GetOrCreateRequest(key)
+	if chunk.ReqC == nil {
+		log.Trace(fmt.Sprintf("DPA.Get: %v found locally, %d bytes", key.Log(), len(chunk.SData)))
+		return
 	}
+
+	if created {
+		if err := self.retrieve(chunk); err != nil {
+			return nil, err
+		}
+	}
+	t := time.NewTicker(searchTimeout)
+	defer t.Stop()
+
+	select {
+	case <-t.C:
+		log.Trace(fmt.Sprintf("DPA.Get: %v request time out ", key.Log()))
+		return nil, notFound
+	case <-chunk.ReqC:
+	}
+	chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8]))
+	return chunk, nil
 }
 
-//this can only finally be set after all config options (file, cmd line, env vars)
+// Put is the entrypoint for local store requests coming from storeLoop
-//have been evaluated
+func (self *NetStore) Put(chunk *Chunk) {
-func (self *StoreParams) Init(path string) {
+	self.localStore.Put(chunk)
-	self.ChunkDbPath = filepath.Join(path, "chunks")
 }
 
-// // netstore contructor, takes path argument that is used to initialise dbStore,
+// Close chunk store
-// // the persistent (disk) storage component of LocalStore
+func (self *NetStore) Close() {}
-// // the second argument is the hive, the connection/logistics manager for the node
-// func NewNetStore(hash SwarmHasher, lstore *LocalStore, cloud CloudStore, params *StoreParams) *NetStore {
-// 	return &NetStore{
-// 		hashfunc:   hash,
-// 		localStore: lstore,
-// 		cloud:      cloud,
-// 	}
-// }
-
-// const (
-// 	// maximum number of peers that a retrieved message is delivered to
-// 	requesterCount = 3
-// )
-
-var (
-	// timeout interval before retrieval is timed out
-	searchTimeout = 3 * time.Second
-)
-
-// // store logic common to local and network chunk store requests
-// // ~ unsafe put in localdb no check if exists no extra copy no hash validation
-// // the chunk is forced to propagate (Cloud.Store) even if locally found!
-// // caller needs to make sure if that is wanted
-// func (self *NetStore) Put(entry *Chunk) {
-// 	self.localStore.Put(entry)
-
-// 	// handle deliveries
-// 	if entry.ReqC != nil {
-// 		log.Trace(fmt.Sprintf("NetStore.Put: localStore.Put %v hit existing request...delivering", entry.Key.Log()))
-// 		// closing C signals to other routines (local requests)
-// 		// that the chunk is has been retrieved
-// 		close(entry.ReqC)
-// 		// deliver the chunk to requesters upstream
-// 		go self.cloud.Deliver(entry)
-// 	} else {
-// 		log.Trace(fmt.Sprintf("NetStore.Put: localStore.Put %v stored locally", entry.Key.Log()))
-// 		// handle propagating store requests
-// 		// go self.cloud.Store(entry)
-// 		go self.cloud.Store(entry)
-// 	}
-// }
-
-// // retrieve logic common for local and network chunk retrieval requests
-// func (self *NetStore) Get(key Key) (*Chunk, error) {
-// 	chunk, _ := self.localStore.GetOrCreateRequest(key)
-// 	go self.cloud.Retrieve(chunk)
-// 	return chunk, nil
-// }
-
-// // Close netstore
-// func (self *NetStore) Close() {}

swarm/storage/resource.go

@@ -639,7 +639,7 @@ type resourceChunkStore struct {
 func newResourceChunkStore(path string, hasher SwarmHasher, localStore *LocalStore, request func(*Chunk) error) *resourceChunkStore {
 	return &resourceChunkStore{
 		localStore: localStore,
-		netStore:   NewDpaChunkStore(localStore, request),
+		netStore:   NewNetStore(localStore, request),
 	}
 }
 

swarm/swarm.go

@@ -130,14 +130,15 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, ensClient *e
 	}
 
 	dbAccess := network.NewDbAccess(self.lstore)
-	self.streamer = network.NewStreamer(to, dbAccess)
+	delivery := network.NewDelivery(to, dbAccess)
+	self.streamer = network.NewStreamer(delivery)
 	network.RegisterOutgoingSyncer(self.streamer, dbAccess)
 	network.RegisterIncomingSyncer(self.streamer, dbAccess)
 
 	self.bzz = network.NewBzz(bzzconfig, to, nil, self.streamer)
 
 	// set up DPA, the cloud storage local access layer
-	dpaChunkStore := storage.NewDpaChunkStore(self.lstore, self.streamer.Retrieve)
+	dpaChunkStore := storage.NewNetStore(self.lstore, self.streamer.Retrieve)
 	log.Debug(fmt.Sprintf("-> Local Access to Swarm"))
 	// Swarm Hash Merklised Chunking for Arbitrary-length Document/File storage
 	self.dpa = storage.NewDPA(dpaChunkStore, self.config.ChunkerParams)

swarm/testutil/http.go

@@ -36,9 +36,8 @@ func NewTestSwarmServer(t *testing.T) *TestSwarmServer {
 		ChunkDbPath:   dir,
 		DbCapacity:    5000000,
 		CacheCapacity: 5000,
-		Radius:        0,
 	}
-	localStore, err := storage.NewLocalStore(storage.MakeHashFunc("SHA3"), storeparams)
+	localStore, err := storage.NewLocalStore(storage.MakeHashFunc("SHA3"), storeparams, make([]byte, 32), nil)
 	if err != nil {
 		os.RemoveAll(dir)
 		t.Fatal(err)