swarm: Add base api for mutable resources

This commit is contained in:
lash 2018-01-18 05:09:49 +01:00
parent cf191d30a9
commit 13543f40c9
8 changed files with 224 additions and 19 deletions

View file

@ -46,15 +46,17 @@ on top of the dpa
it is the public interface of the dpa which is included in the ethereum stack it is the public interface of the dpa which is included in the ethereum stack
*/ */
type Api struct { type Api struct {
dpa *storage.DPA dpa *storage.DPA
dns Resolver dns Resolver
resource *storage.ResourceHandler
} }
//the api constructor initialises //the api constructor initialises
func NewApi(dpa *storage.DPA, dns Resolver) (self *Api) { func NewApi(dpa *storage.DPA, dns Resolver, resourceHandler *storage.ResourceHandler) (self *Api) {
self = &Api{ self = &Api{
dpa: dpa, dpa: dpa,
dns: dns, dns: dns,
resource: resourceHandler,
} }
return return
} }
@ -361,3 +363,21 @@ func (self *Api) BuildDirectoryTree(mhash string, nameresolver bool) (key storag
} }
return key, manifestEntryMap, nil return key, manifestEntryMap, nil
} }
func (self *Api) DbLookupLatest(name string) (io.ReadSeeker, error) {
_, err := self.resource.LookupLatest(name, true)
if err != nil {
return nil, err
}
return bytes.NewReader(self.resource.GetData(name)), nil
}
func (self *Api) DbCreate(name string, frequency uint64) (err error) {
_, err = self.resource.NewResource(name, frequency)
return err
}
func (self *Api) DbUpdate(name string, data []byte) (storage.Key, uint32, uint32, error) {
key, err := self.resource.Update(name, data)
return key, self.resource.GetLastPeriod(name), self.resource.GetVersion(name), err
}

View file

@ -40,7 +40,7 @@ func testApi(t *testing.T, f func(*Api)) {
if err != nil { if err != nil {
return return
} }
api := NewApi(dpa, nil) api := NewApi(dpa, nil, nil)
dpa.Start() dpa.Start()
f(api) f(api)
dpa.Stop() dpa.Stop()

View file

@ -55,10 +55,6 @@ func TestConfig(t *testing.T) {
t.Fatal("Failed to correctly initialize SwapParams") t.Fatal("Failed to correctly initialize SwapParams")
} }
if one.HiveParams.MaxPeersPerRequest != 5 {
t.Fatal("Failed to correctly initialize HiveParams")
}
if one.StoreParams.ChunkDbPath == one.Path { if one.StoreParams.ChunkDbPath == one.Path {
t.Fatal("Failed to correctly initialize StoreParams") t.Fatal("Failed to correctly initialize StoreParams")
} }

View file

@ -21,6 +21,7 @@ package http
import ( import (
"archive/tar" "archive/tar"
"bytes"
"encoding/json" "encoding/json"
"errors" "errors"
"fmt" "fmt"
@ -290,6 +291,56 @@ func (s *Server) HandleDelete(w http.ResponseWriter, r *Request) {
fmt.Fprint(w, newKey) fmt.Fprint(w, newKey)
} }
func (s *Server) HandlePostDb(w http.ResponseWriter, r *Request) {
if r.ContentLength == 0 {
frequency, err := strconv.ParseUint(r.uri.Path, 10, 64)
if err != nil {
w.WriteHeader(http.StatusBadRequest)
http.ServeContent(w, &r.Request, "", time.Now(), bytes.NewReader([]byte(err.Error())))
return
}
err = s.api.DbCreate(r.uri.Addr, frequency)
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
return
}
} else {
data, err := ioutil.ReadAll(r.Body)
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
return
}
_, _, _, err = s.api.DbUpdate(r.uri.Addr, data)
if err != nil {
w.WriteHeader(http.StatusUnauthorized)
http.ServeContent(w, &r.Request, "", time.Now(), bytes.NewReader([]byte(err.Error())))
return
}
}
w.WriteHeader(http.StatusOK)
}
func (s *Server) HandleGetDb(w http.ResponseWriter, r *Request) {
w.Header().Set("Content-Type", "application/octet-stream")
var params []string
if len(r.uri.Path) > 0 {
params = strings.Split(r.uri.Path, "/")
}
switch len(params) {
case 0:
data, err := s.api.DbLookupLatest(r.uri.Addr)
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
break
}
http.ServeContent(w, &r.Request, "", time.Now(), data)
break
default:
w.WriteHeader(http.StatusBadRequest)
}
}
// HandleGet handles a GET request to // HandleGet handles a GET request to
// - bzz-raw://<key> and responds with the raw content stored at the // - bzz-raw://<key> and responds with the raw content stored at the
// given storage key // given storage key
@ -604,6 +655,8 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
case "POST": case "POST":
if uri.Raw() || uri.DeprecatedRaw() { if uri.Raw() || uri.DeprecatedRaw() {
s.HandlePostRaw(w, req) s.HandlePostRaw(w, req)
} else if uri.Db() {
s.HandlePostDb(w, req)
} else { } else {
s.HandlePostFiles(w, req) s.HandlePostFiles(w, req)
} }
@ -644,6 +697,11 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
return return
} }
if uri.Db() {
s.HandleGetDb(w, req)
return
}
s.HandleGetFile(w, req) s.HandleGetFile(w, req)
default: default:

View file

@ -22,17 +22,57 @@ import (
"fmt" "fmt"
"io/ioutil" "io/ioutil"
"net/http" "net/http"
"os"
"strings" "strings"
"sync" "sync"
"testing" "testing"
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/swarm/api" "github.com/ethereum/go-ethereum/swarm/api"
swarm "github.com/ethereum/go-ethereum/swarm/api/client" swarm "github.com/ethereum/go-ethereum/swarm/api/client"
"github.com/ethereum/go-ethereum/swarm/storage" "github.com/ethereum/go-ethereum/swarm/storage"
"github.com/ethereum/go-ethereum/swarm/testutil" "github.com/ethereum/go-ethereum/swarm/testutil"
) )
func init() {
log.Root().SetHandler(log.CallerFileHandler(log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat(true)))))
}
func TestBzzGetDb(t *testing.T) {
srv := testutil.NewTestSwarmServer(t)
defer srv.Close()
url := srv.URL + "/bzz-db:/foo/42"
resp, err := http.Post(url, "application/octet-stream", nil)
if err != nil {
fmt.Printf("err: %v\n", err)
return
}
b, err := ioutil.ReadAll(resp.Body)
fmt.Printf("Create: %s : %s\n", resp.Status, b)
url = srv.URL + "/bzz-db:/foo"
data := []byte("foo")
resp, err = http.Post(url, "application/octet-stream", bytes.NewReader(data))
if err != nil {
fmt.Printf("err: %v\n", err)
return
}
b, err = ioutil.ReadAll(resp.Body)
fmt.Printf("Update: %s : %s\n", resp.Status, b)
url = srv.URL + "/bzz-db:/foo"
resp, err = http.Get(url)
if err != nil {
fmt.Printf("err: %v\n", err)
return
}
b, err = ioutil.ReadAll(resp.Body)
fmt.Printf("Get: %s : %s\n", resp.Status, b)
}
func TestBzzGetPath(t *testing.T) { func TestBzzGetPath(t *testing.T) {
var err error var err error
@ -258,7 +298,6 @@ func TestBzzGetPath(t *testing.T) {
t.Fatalf("Non-Hash response body does not match, expected: %v, got: %v", nonhashresponses[i], string(respbody)) t.Fatalf("Non-Hash response body does not match, expected: %v, got: %v", nonhashresponses[i], string(respbody))
} }
} }
} }
// TestBzzRootRedirect tests that getting the root path of a manifest without // TestBzzRootRedirect tests that getting the root path of a manifest without

View file

@ -69,7 +69,7 @@ func Parse(rawuri string) (*URI, error) {
// check the scheme is valid // check the scheme is valid
switch uri.Scheme { switch uri.Scheme {
case "bzz", "bzz-raw", "bzz-immutable", "bzz-list", "bzz-hash", "bzzr", "bzzi": case "bzz", "bzz-raw", "bzz-immutable", "bzz-list", "bzz-hash", "bzzr", "bzzi", "bzz-db":
default: default:
return nil, fmt.Errorf("unknown scheme %q", u.Scheme) return nil, fmt.Errorf("unknown scheme %q", u.Scheme)
} }
@ -92,6 +92,10 @@ func Parse(rawuri string) (*URI, error) {
return uri, nil return uri, nil
} }
func (u *URI) Db() bool {
return u.Scheme == "bzz-db"
}
func (u *URI) Raw() bool { func (u *URI) Raw() bool {
return u.Scheme == "bzz-raw" return u.Scheme == "bzz-raw"
} }

View file

@ -44,8 +44,8 @@ type resource struct {
} }
// TODO Expire content after a defined period (to force resync) // TODO Expire content after a defined period (to force resync)
func (r *resource) isSynced() bool { func (self *resource) isSynced() bool {
return !r.updated.IsZero() return !self.updated.IsZero()
} }
// Implement to activate validation of resource updates // Implement to activate validation of resource updates
@ -166,6 +166,30 @@ func NewResourceHandler(datadir string, cloudStore CloudStore, rpcClient *rpc.Cl
return rh, nil return rh, nil
} }
func (self *ResourceHandler) GetData(name string) []byte {
rsrc := self.getResource(name)
if rsrc == nil {
return nil
}
return rsrc.data
}
func (self *ResourceHandler) GetLastPeriod(name string) uint32 {
rsrc := self.getResource(name)
if rsrc == nil {
return 0
}
return rsrc.lastPeriod
}
func (self *ResourceHandler) GetVersion(name string) uint32 {
rsrc := self.getResource(name)
if rsrc == nil {
return 0
}
return rsrc.version
}
// \TODO should be hashsize * branches from the chosen chunker, implement with dpa // \TODO should be hashsize * branches from the chosen chunker, implement with dpa
func (self *ResourceHandler) chunkSize() int64 { func (self *ResourceHandler) chunkSize() int64 {
return chunkSize return chunkSize

View file

@ -17,11 +17,15 @@
package testutil package testutil
import ( import (
"crypto/ecdsa"
"io/ioutil" "io/ioutil"
"net/http/httptest" "net/http/httptest"
"os" "os"
"path/filepath"
"strconv"
"testing" "testing"
"github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/swarm/api" "github.com/ethereum/go-ethereum/swarm/api"
httpapi "github.com/ethereum/go-ethereum/swarm/api/http" httpapi "github.com/ethereum/go-ethereum/swarm/api/http"
"github.com/ethereum/go-ethereum/swarm/storage" "github.com/ethereum/go-ethereum/swarm/storage"
@ -38,7 +42,7 @@ func NewTestSwarmServer(t *testing.T) *TestSwarmServer {
CacheCapacity: 5000, CacheCapacity: 5000,
Radius: 0, Radius: 0,
} }
localStore, err := storage.NewLocalStore(storage.MakeHashFunc("SHA3"), storeparams, nil) localStore, err := storage.NewLocalStore(storage.MakeHashFunc(storage.SHA3Hash), storeparams)
if err != nil { if err != nil {
os.RemoveAll(dir) os.RemoveAll(dir)
t.Fatal(err) t.Fatal(err)
@ -49,20 +53,59 @@ func NewTestSwarmServer(t *testing.T) *TestSwarmServer {
ChunkStore: localStore, ChunkStore: localStore,
} }
dpa.Start() dpa.Start()
a := api.NewApi(dpa, nil)
// mutable resources test setup
resourcedir, err := ioutil.TempDir("", "swarm-resource-test")
if err != nil {
t.Fatal(err)
}
ipcpath := filepath.Join(resourcedir, "test.ipc")
ipcl, err := rpc.CreateIPCListener(ipcpath)
if err != nil {
t.Fatal(err)
}
rpcserver := rpc.NewServer()
rpcserver.RegisterName("eth", &FakeRPC{})
go func() {
rpcserver.ServeListener(ipcl)
}()
rpcClean := func() {
rpcserver.Stop()
}
// connect to fake rpc
rpcclient, err := rpc.Dial(ipcpath)
if err != nil {
t.Fatal(err)
}
rh, err := storage.NewResourceHandler(resourcedir, &testCloudStore{}, rpcclient, nil)
if err != nil {
t.Fatal(err)
}
a := api.NewApi(dpa, nil, rh)
srv := httptest.NewServer(httpapi.NewServer(a)) srv := httptest.NewServer(httpapi.NewServer(a))
return &TestSwarmServer{ return &TestSwarmServer{
Server: srv, Server: srv,
Dpa: dpa, Dpa: dpa,
dir: dir, dir: dir,
hasher: storage.MakeHashFunc("SHA3")(),
cleanup: func() {
rh.Close()
rpcClean()
os.RemoveAll(dir)
os.RemoveAll(resourcedir)
},
} }
} }
type TestSwarmServer struct { type TestSwarmServer struct {
*httptest.Server *httptest.Server
hasher storage.SwarmHash
Dpa *storage.DPA privatekey *ecdsa.PrivateKey
dir string Dpa *storage.DPA
dir string
cleanup func()
} }
func (t *TestSwarmServer) Close() { func (t *TestSwarmServer) Close() {
@ -70,3 +113,24 @@ func (t *TestSwarmServer) Close() {
t.Dpa.Stop() t.Dpa.Stop()
os.RemoveAll(t.dir) os.RemoveAll(t.dir)
} }
type testCloudStore struct {
}
func (c *testCloudStore) Store(*storage.Chunk) {
}
func (c *testCloudStore) Deliver(*storage.Chunk) {
}
func (c *testCloudStore) Retrieve(*storage.Chunk) {
}
// for faking the rpc service, since we don't need the whole node stack
type FakeRPC struct {
blocknumber uint64
}
func (r *FakeRPC) BlockNumber() (string, error) {
return strconv.FormatUint(r.blocknumber, 10), nil
}