From 0265566534bb180a70819583c7cb2b20174e1161 Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Wed, 17 Jan 2018 18:51:08 +0100 Subject: [PATCH] Fix swarm api tests and rename dpaChunkStore to NetStore --- swarm/api/api_test.go | 5 +- swarm/api/config_test.go | 12 +- swarm/api/filesystem.go | 4 +- swarm/api/filesystem_test.go | 8 +- swarm/api/http/server_test.go | 2 +- swarm/api/manifest.go | 7 +- swarm/api/storage.go | 4 +- swarm/fuse/swarmfs_test.go | 2 +- swarm/network/protocol.go | 9 +- swarm/network/request_test.go | 2 +- .../simulations/discovery/discovery_test.go | 2 +- swarm/network/streamer.go | 4 + swarm/pss/client/client_test.go | 4 +- swarm/pss/pss_test.go | 6 +- swarm/storage/chunker_test.go | 8 +- swarm/storage/dpa.go | 55 +------ swarm/storage/localstore.go | 26 ++- swarm/storage/netstore.go | 148 ++++++------------ swarm/storage/resource.go | 2 +- swarm/swarm.go | 5 +- swarm/testutil/http.go | 3 +- 21 files changed, 115 insertions(+), 203 deletions(-) diff --git a/swarm/api/api_test.go b/swarm/api/api_test.go index 44bf8aadc2..da1d8bcf23 100644 --- a/swarm/api/api_test.go +++ b/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) }) } diff --git a/swarm/api/config_test.go b/swarm/api/config_test.go index 5636b6dafb..993388686b 100644 --- a/swarm/api/config_test.go +++ b/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() - two := NewDefaultConfig() + one := NewConfig() + 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") } diff --git a/swarm/api/filesystem.go b/swarm/api/filesystem.go index b6a2de8862..0074ea167d 100644 --- a/swarm/api/filesystem.go +++ b/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 diff --git a/swarm/api/filesystem_test.go b/swarm/api/filesystem_test.go index 8a15e735dc..6f1594e991 100644 --- a/swarm/api/filesystem_test.go +++ b/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, err := api.Store(bytes.NewReader(index), int64(len(index)), wg) - wg.Wait() + hash, wait, err := api.Store(bytes.NewReader(index), int64(len(index))) + 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") diff --git a/swarm/api/http/server_test.go b/swarm/api/http/server_test.go index 6a35d2c785..bd05d77fcc 100644 --- a/swarm/api/http/server_test.go +++ b/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 { diff --git a/swarm/api/manifest.go b/swarm/api/manifest.go index f6279a4ced..b8b64caa89 100644 --- a/swarm/api/manifest.go +++ b/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 } diff --git a/swarm/api/storage.go b/swarm/api/storage.go index ae94e15cb9..4679fabad3 100644 --- a/swarm/api/storage.go +++ b/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 } diff --git a/swarm/fuse/swarmfs_test.go b/swarm/fuse/swarmfs_test.go index 42af36345f..e768f86495 100644 --- a/swarm/fuse/swarmfs_test.go +++ b/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) } diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go index 9374d844c8..4f5906b4ae 100644 --- a/swarm/network/protocol.go +++ b/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 diff --git a/swarm/network/request_test.go b/swarm/network/request_test.go index 12e820b2ad..8ae1d3c344 100644 --- a/swarm/network/request_test.go +++ b/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 { diff --git a/swarm/network/simulations/discovery/discovery_test.go b/swarm/network/simulations/discovery/discovery_test.go index fc8d6f70ca..33a1c04868 100644 --- a/swarm/network/simulations/discovery/discovery_test.go +++ b/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 } diff --git a/swarm/network/streamer.go b/swarm/network/streamer.go index 66abf82f3a..b0e8b9eb3b 100644 --- a/swarm/network/streamer.go +++ b/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() diff --git a/swarm/pss/client/client_test.go b/swarm/pss/client/client_test.go index e773018a16..fe11e37c91 100644 --- a/swarm/pss/client/client_test.go +++ b/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 }, } } diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go index 57d4a79170..bda1490c9d 100644 --- a/swarm/pss/pss_test.go +++ b/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) diff --git a/swarm/storage/chunker_test.go b/swarm/storage/chunker_test.go index fb66b7c756..cb9d73a93e 100644 --- a/swarm/storage/chunker_test.go +++ b/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") } diff --git a/swarm/storage/dpa.go b/swarm/storage/dpa.go index b54c63804c..b0aafe0343 100644 --- a/swarm/storage/dpa.go +++ b/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() { -} diff --git a/swarm/storage/localstore.go b/swarm/storage/localstore.go index c12c706af5..893670b232 100644 --- a/swarm/storage/localstore.go +++ b/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) { - dbStore, err := NewDbStore(params.ChunkDbPath, hash, params.DbCapacity, func(k Key) (ret uint8) { return uint8(Proximity(basekey[:], k[:])) }) +func NewLocalStore(hash SwarmHasher, params *StoreParams, basekey []byte, mockStore *mock.NodeStore) (*LocalStore, error) { + 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 } diff --git a/swarm/storage/netstore.go b/swarm/storage/netstore.go index 4a7caf7c2e..f01ffe4a69 100644 --- a/swarm/storage/netstore.go +++ b/swarm/storage/netstore.go @@ -17,116 +17,58 @@ package storage import ( - "path/filepath" + "encoding/binary" + "fmt" "time" + + "github.com/ethereum/go-ethereum/log" ) -// import ( -// "fmt" -// "path/filepath" -// "time" - -// "github.com/ethereum/go-ethereum/log" -// ) - -// /* -// 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 +// NetStore 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 NetStore struct { + localStore *LocalStore + retrieve func(chunk *Chunk) error } -//create params with default values -func NewDefaultStoreParams() (self *StoreParams) { - return &StoreParams{ - DbCapacity: defaultDbCapacity, - CacheCapacity: defaultCacheCapacity, - Radius: defaultRadius, +func NewNetStore(localStore *LocalStore, retrieve func(chunk *Chunk) error) *NetStore { + return &NetStore{localStore, retrieve} +} + +// Get is the entrypoint for local retrieve requests +// 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) -//have been evaluated -func (self *StoreParams) Init(path string) { - self.ChunkDbPath = filepath.Join(path, "chunks") +// Put is the entrypoint for local store requests coming from storeLoop +func (self *NetStore) Put(chunk *Chunk) { + self.localStore.Put(chunk) } -// // netstore contructor, takes path argument that is used to initialise dbStore, -// // the persistent (disk) storage component of LocalStore -// // 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() {} +// Close chunk store +func (self *NetStore) Close() {} diff --git a/swarm/storage/resource.go b/swarm/storage/resource.go index e285962591..3f27de5e3e 100644 --- a/swarm/storage/resource.go +++ b/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), } } diff --git a/swarm/swarm.go b/swarm/swarm.go index 3657088383..bc6533875b 100644 --- a/swarm/swarm.go +++ b/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) diff --git a/swarm/testutil/http.go b/swarm/testutil/http.go index f2922fab00..32c83abc4e 100644 --- a/swarm/testutil/http.go +++ b/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)