forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-23 05:06:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

bzz API (internal, JS, http)

Browse source 
* SwarmProxyPortFlag --bzzport (8500) to allow multiple instances on a local swarm
* roundtripper proxying to localhost allows bzz protocol uris with test
* bzz/api implements
  * get as per recursive manifest resolution abstracted from httpaccess
  * post
  * getManifest ~ raw
  * download as per full recursive manifest resolution abstracted from httpaccess
  * upload as per bzzup.sh file system directory walkthrough
* bzz/js_api implements js bindings for console delegating to api
* refactor backend - swarm interface: backend simply interacts with bzz.Api
* Api starts http server, resolver is set in JS
This commit is contained in:
zelig 2015-05-25 10:16:16 +01:00
parent a7804860c9
commit acf73a4237
15 changed files with 610 additions and 223 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

188
bzz/api.go Normal file
View file

@ -0,0 +1,188 @@
package bzz
import (
"encoding/json"
"fmt"
"net"
"path/filepath"
"strings"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/resolver"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
)
/*
Api implements webserver/file system related content storage and retrieval
on top of the dpa
*/
type Api struct {
dpa *DPA
netStore *netStore
Resolver *resolver.Resolver
}
func NewApi(datadir, port string) (api *Api, err error) {
api = &Api{}
api.netStore, err = newNetStore(filepath.Join(datadir, "bzz"), filepath.Join(datadir, "bzzpeers.json"))
if err != nil {
return
}
chunker := &TreeChunker{}
chunker.Init()
api.dpa = &DPA{
Chunker: chunker,
ChunkStore: api.netStore,
}
return
}
func (self *Api) Bzz() (p2p.Protocol, error) {
return BzzProtocol(self.netStore)
}
func (self *Api) Start(node *discover.Node, connectPeer func(string) error) {
self.dpa.Start()
self.netStore.Start(node, connectPeer)
}
func (self *Api) Stop() {
self.dpa.Stop()
self.netStore.Stop()
}
// Get uses iterative manifest retrieval and prefix matching
// to resolve path to contenwt using dpa retrieve
func (self *Api) Get(bzzpath string) (string, error) {
return "", nil
}
// Put provides singleton manifest creation and optional name registration
// on top of dpa store
func (self *Api) Put(content, contentType, address, domain string) (string, error) {
return "", nil
}
// Download replicates the manifest path structure on the local filesystem
// under localpath
func (self *Api) Download(bzzpath, localpath string) (string, error) {
return "", nil
}
// Upload replicates a local directory as a manifest file and uploads it
// using dpa store
// TODO: localpath should point to a manifest
func (self *Api) Upload(localpath, address, domain string) (string, error) {
return "", nil
}
type errResolve error
func (self *Api) resolveHost(hostport string) (contentHash Key, errR errResolve) {
var host, port string
var err error
host, port, err = net.SplitHostPort(hostport)
if err != nil {
errR = errResolve(fmt.Errorf("invalid host '%s': %v", hostport, err))
return
}
if hashMatcher.MatchString(host) {
contentHash = Key(host)
} else {
if self.Resolver != nil {
hostHash := common.BytesToHash(crypto.Sha3([]byte(host)))
// TODO: should take port as block number versioning
_ = port
var hash common.Hash
hash, err = self.Resolver.KeyToContentHash(hostHash)
if err != nil {
err = errResolve(fmt.Errorf("unable to resolve '%s': %v", hostport, err))
}
contentHash = Key(hash.Bytes())
} else {
err = errResolve(fmt.Errorf("no resolver '%s': %v", hostport, err))
}
}
return
}
func (self *Api) getPath(uri string) (reader SectionReader, mimeType string, status int, err error) {
parts := strings.SplitAfterN(uri[1:], "/", 2)
hostPort := parts[0]
path := parts[1]
dpaLogger.Debugf("Swarm: host: '%s', path '%s' requested.", hostPort, path)
//resolving host and port
var key Key
key, err = self.resolveHost(hostPort)
if err != nil {
return
}
// retrieve content following path along manifests
var pos int
for {
// retrieve manifest via DPA
manifestReader := self.dpa.Retrieve(key)
// TODO check size for oversized manifests
manifestData := make([]byte, manifestReader.Size())
var size int
size, err = manifestReader.Read(manifestData)
if int64(size) < manifestReader.Size() {
dpaLogger.Debugf("Swarm: Manifest for '%s' not found.", uri)
if err == nil {
err = fmt.Errorf("Manifest retrieval cut short: %v &lt; %v", size, manifestReader.Size())
}
return
}
dpaLogger.Debugf("Swarm: Manifest for '%s' retrieved.", uri)
man := manifest{}
err = json.Unmarshal(manifestData, &man)
if err != nil {
err = fmt.Errorf("Manifest for '%s' is malformed: %v", uri, err)
dpaLogger.Debugf("Swarm: %v", err)
return
}
dpaLogger.Debugf("Swarm: Manifest for '%s' has %d entries.", uri, len(man.Entries))
// retrieve entry that matches path from manifest entries
var entry *manifestEntry
entry, pos = man.getEntry(path)
if entry == nil {
err = fmt.Errorf("Content for '%s' not found.", uri)
return
}
// check hash of entry
if !hashMatcher.MatchString(entry.Hash) {
err = fmt.Errorf("Incorrect hash '%064x' for '%s'", entry.Hash, uri)
return
}
key = common.Hex2Bytes(entry.Hash)
status = entry.Status
// get mime type of entry
mimeType = entry.ContentType
if mimeType != "" {
mimeType = manifestType
}
// if path matched on non-manifest content type, then retrieve reader
// and return
if mimeType != manifestType {
reader = self.dpa.Retrieve(key)
return
}
// otherwise continue along the path with manifest resolution
path = path[pos:]
}
return
}

10
bzz/api_test.go Normal file
View file

@ -0,0 +1,10 @@
package bzz
import (
// "net/http"
// "strings"
// "testing"
// "time"
// "github.com/ethereum/go-ethereum/common/docserver"
)

236
bzz/httpaccess.go
View file

@ -4,9 +4,7 @@ A simple http server interface to Swarm
package bzz
import (
"encoding/json"
"fmt"
"github.com/ethereum/go-ethereum/common"
"io"
"net/http"
"regexp"
@ -15,15 +13,15 @@ import (
)
const (
port = ":8500"
manifestType = "application/bzz-manifest+json"
notFoundStatus = 404
rawType = "application/octet-stream"
)
var (
protocolMatcher = regexp.MustCompile("^/bzz:")
uriMatcher = regexp.MustCompile("^/raw/[0-9A-Fa-f]{64}(?:/[a-z]+/[-+0-9a-z]+)?$")
// protocolMatcher = regexp.MustCompile("^/bzz:")
rawManifestMatcher = regexp.MustCompile("^/raw/")
// rawManifestMatcher = regexp.MustCompile("^/raw/[0-9A-Fa-f]{64}(?:/[a-z]+/[-+0-9a-z]+)?$")
manifestMatcher = regexp.MustCompile("^/[0-9A-Fa-f]{64}")
hashMatcher = regexp.MustCompile("^[0-9A-Fa-f]{64}$")
)
type sequentialReader struct {
@ -33,15 +31,101 @@ type sequentialReader struct {
lock sync.Mutex
}
type manifest struct {
Entries []manifestEntry
// starts up http server
// TODO: started by dpa/api rather than backend
func startHttpServer(api *Api, port string) {
http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
handler(w, r, api)
})
go http.ListenAndServe(port, nil)
dpaLogger.Infof("Swarm HTTP proxy started.")
}
type manifestEntry struct {
Path string
Hash string
ContentType string
Status int16
func handler(w http.ResponseWriter, r *http.Request, api *Api) {
switch {
case r.Method == "POST":
if r.URL.Path == "/raw" {
dpaLogger.Debugf("Swarm: POST request received.")
key, err := api.dpa.Store(io.NewSectionReader(&sequentialReader{
reader: r.Body,
ahead: make(map[int64]chan bool),
}, 0, r.ContentLength), nil)
if err == nil {
fmt.Fprintf(w, "%064x", key)
dpaLogger.Debugf("Swarm: Object %064x stored", key)
} else {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
} else {
http.Error(w, "No POST to "+r.URL.Path+" allowed.", http.StatusBadRequest)
return
}
case r.Method == "GET" || r.Method == "HEAD":
uri := r.URL.Path
dpaLogger.Debugf("request URL Host: '%s', Path: '%s'", r.URL.Host, r.URL.Path)
// raw ,
if rawManifestMatcher.MatchString(uri) {
dpaLogger.Debugf("Swarm: Raw GET request '%s' received", uri)
// resolving host
name := uri[5:]
key, err := api.resolveHost(uri)
if err != nil {
dpaLogger.Debugf("Swarm: %v", err)
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
// retrieving content
reader := api.dpa.Retrieve(key)
dpaLogger.Debugf("Swarm: Reading %d bytes.", reader.Size())
// setting mime type
qv := r.URL.Query()
mimeType := qv.Get("content_type")
if mimeType == "" {
mimeType = rawType
}
w.Header().Set("Content-Type", mimeType)
http.ServeContent(w, r, name, time.Unix(0, 0), reader)
dpaLogger.Debugf("Swarm: Serve raw content '%s' (%d bytes) as '%s'", uri, reader.Size(), mimeType)
// retrieve path via manifest
} else {
dpaLogger.Debugf("Swarm: Structured GET request '%s' received.", uri)
// call to api.getPath on uri
reader, mimeType, status, err := api.getPath(uri)
if err != nil {
var status int
if _, ok := err.(errResolve); ok {
dpaLogger.Debugf("Swarm: %v", err)
status = http.StatusBadRequest
} else {
dpaLogger.Debugf("Swarm: error retrieving '%s': %v", uri, err)
status = http.StatusNotFound
}
http.Error(w, err.Error(), status)
return
}
// set mime type and status headers
w.Header().Set("Content-Type", mimeType)
if status > 0 {
w.WriteHeader(status)
}
http.ServeContent(w, r, uri, time.Unix(0, 0), reader)
dpaLogger.Debugf("Swarm: Served '%s' (%d bytes) as '%s' (status code: %d)", uri, reader.Size(), mimeType, status)
}
default:
http.Error(w, "Method "+r.Method+" is not supported.", http.StatusMethodNotAllowed)
}
}
func (self *sequentialReader) ReadAt(target []byte, off int64) (n int, err error) {
@ -93,127 +177,3 @@ func (self *sequentialReader) ReadAt(target []byte, off int64) (n int, err error
self.lock.Unlock()
return localPos, err
}
func handler(w http.ResponseWriter, r *http.Request, dpa *DPA) {
uri := protocolMatcher.ReplaceAllString(r.RequestURI, "")
switch {
case r.Method == "POST":
if uri == "/raw" {
dpaLogger.Debugf("Swarm: POST request received.")
key, err := dpa.Store(io.NewSectionReader(&sequentialReader{
reader: r.Body,
ahead: make(map[int64]chan bool),
}, 0, r.ContentLength), nil)
if err == nil {
fmt.Fprintf(w, "%064x", key)
dpaLogger.Debugf("Swarm: Object %064x stored", key)
} else {
http.Error(w, err.Error(), http.StatusBadRequest)
}
} else {
http.Error(w, "No POST to "+uri+" allowed.", http.StatusBadRequest)
}
case r.Method == "GET" || r.Method == "HEAD":
if uriMatcher.MatchString(uri) {
dpaLogger.Debugf("Swarm: Raw GET request %s received", uri)
name := uri[5:69]
key := common.Hex2Bytes(name)
reader := dpa.Retrieve(key)
dpaLogger.Debugf("Swarm: Reading %d bytes.", reader.Size())
mimeType := "application/octet-stream"
if len(uri) > 70 {
mimeType = uri[70:]
}
w.Header().Set("Content-Type", mimeType)
http.ServeContent(w, r, name, time.Unix(0, 0), reader)
dpaLogger.Debugf("Swarm: Object %s returned.", name)
} else if manifestMatcher.MatchString(uri) {
dpaLogger.Debugf("Swarm: Structured GET request %s received.", uri)
name := uri[1:65]
path := uri[65:] // typically begins with a /
dpaLogger.Debugf("Swarm: path \"%s\" requested.", path)
key := common.Hex2Bytes(name)
MANIFEST_RESOLUTION:
for {
manifestReader := dpa.Retrieve(key)
// TODO check size for oversized manifests
manifestData := make([]byte, manifestReader.Size())
size, err := manifestReader.Read(manifestData)
if int64(size) < manifestReader.Size() {
dpaLogger.Debugf("Swarm: Manifest %s not found.", name)
if err == nil {
http.Error(w, "Manifest retrieval cut short: "+string(size)+"&lt;"+string(manifestReader.Size()),
http.StatusNotFound)
} else {
http.Error(w, err.Error(), http.StatusNotFound)
}
return
}
dpaLogger.Debugf("Swarm: Manifest %s retrieved.", name)
man := manifest{}
err = json.Unmarshal(manifestData, &man)
if err != nil {
dpaLogger.Debugf("Swarm: Manifest %s is malformed.", name)
http.Error(w, err.Error(), http.StatusNotFound)
return
} else {
dpaLogger.Debugf("Swarm: Manifest %s has %d entries.", name, len(man.Entries))
}
var mimeType string
key = nil
prefix := 0
status := int16(404)
MANIFEST_ENTRIES:
for _, entry := range man.Entries {
if !hashMatcher.MatchString(entry.Hash) {
// hash is mandatory
continue MANIFEST_ENTRIES
}
if entry.ContentType == "" {
// content type defaults to manifest
entry.ContentType = manifestType
}
pathLen := len(entry.Path)
if len(path) >= pathLen && path[:pathLen] == entry.Path && prefix <= pathLen {
dpaLogger.Debugf("Swarm: \"%s\" matches \"%s\".", path, entry.Path)
prefix = pathLen
key = common.Hex2Bytes(entry.Hash)
dpaLogger.Debugf("Swarm: Payload hash %064x", key)
mimeType = entry.ContentType
status = entry.Status
}
}
if key == nil {
http.Error(w, "Object "+uri+" not found.", http.StatusNotFound)
break MANIFEST_RESOLUTION
} else if mimeType != manifestType {
reader := dpa.Retrieve(key)
w.Header().Set("Content-Type", mimeType)
dpaLogger.Debugf("Swarm: HTTP Status %d", status)
if status > 0 {
w.WriteHeader(int(status))
}
dpaLogger.Debugf("Swarm: Reading %d bytes.", reader.Size())
http.ServeContent(w, r, name, time.Unix(0, 0), reader)
dpaLogger.Debugf("Swarm: Served %s as %s.", mimeType, uri)
break MANIFEST_RESOLUTION
} else {
path = path[prefix:]
// continue with manifest resolution
}
}
} else {
http.Error(w, "Object "+uri+" not found.", http.StatusNotFound)
}
default:
http.Error(w, "Method "+r.Method+" is not supported.", http.StatusMethodNotAllowed)
}
}
func StartHttpServer(dpa *DPA) {
http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
handler(w, r, dpa)
})
go http.ListenAndServe(port, nil)
dpaLogger.Infof("Swarm HTTP proxy started.")
}

163
bzz/js_api.go Normal file
View file

@ -0,0 +1,163 @@
package bzz
import (
"fmt"
// "net/http"
"github.com/ethereum/go-ethereum/jsre"
"github.com/robertkrimen/otto"
)
func NewJSApi(vm *jsre.JSRE, api *Api) (jsapi *JSApi) {
jsapi = &JSApi{
vm: vm,
api: api,
}
vm.Set("bzz", struct{}{})
t, _ := vm.Get("bzz")
o := t.Object()
o.Set("download", jsapi.download)
o.Set("upload", jsapi.upload)
o.Set("get", jsapi.get)
o.Set("put", jsapi.put)
return
}
type JSApi struct {
vm *jsre.JSRE
api *Api
}
func (self *JSApi) get(call otto.FunctionCall) otto.Value {
if len(call.ArgumentList) != 1 {
fmt.Println("requires 1 argument: bzz.get(path)")
return otto.UndefinedValue()
}
var err error
var bzzpath, res string
bzzpath, err = call.Argument(0).ToString()
if err != nil {
fmt.Println(err)
return otto.UndefinedValue()
}
res, err = self.api.Get(bzzpath)
if err != nil {
fmt.Println(err)
return otto.UndefinedValue()
}
v, _ := call.Otto.ToValue(res)
return v
}
func (self *JSApi) put(call otto.FunctionCall) otto.Value {
if len(call.ArgumentList) != 2 || len(call.ArgumentList) != 4 {
fmt.Println("requires 2 or 4 arguments: bzz.put(content, content-type[, address, domain])")
return otto.UndefinedValue()
}
var err error
var res, content, contentType, address, domain string
content, err = call.Argument(0).ToString()
if err != nil {
fmt.Println(err)
return otto.UndefinedValue()
}
contentType, err = call.Argument(1).ToString()
if err != nil {
fmt.Println(err)
return otto.UndefinedValue()
}
if len(call.ArgumentList) > 2 {
address, err = call.Argument(2).ToString()
if err != nil {
fmt.Println(err)
return otto.UndefinedValue()
}
domain, err = call.Argument(3).ToString()
if err != nil {
fmt.Println(err)
return otto.UndefinedValue()
}
}
res, err = self.api.Put(content, contentType, address, domain)
if err != nil {
fmt.Println(err)
return otto.UndefinedValue()
}
v, _ := call.Otto.ToValue(res)
return v
}
func (self *JSApi) download(call otto.FunctionCall) otto.Value {
if len(call.ArgumentList) != 2 {
fmt.Println("requires 2 arguments: bzz.download(bzzpath, localpath)")
return otto.UndefinedValue()
}
var err error
var bzzpath, localpath, res string
bzzpath, err = call.Argument(0).ToString()
if err != nil {
fmt.Println(err)
return otto.UndefinedValue()
}
localpath, err = call.Argument(1).ToString()
if err != nil {
fmt.Println(err)
return otto.UndefinedValue()
}
res, err = self.api.Download(bzzpath, localpath)
if err != nil {
fmt.Println(err)
return otto.UndefinedValue()
}
v, _ := call.Otto.ToValue(res)
return v
}
func (self *JSApi) upload(call otto.FunctionCall) otto.Value {
if len(call.ArgumentList) != 1 || len(call.ArgumentList) != 3 {
fmt.Println("requires 1 or 1 arguments: bzz.put(localpath[, address, domain])")
return otto.UndefinedValue()
}
var err error
var localpath, address, domain, res string
localpath, err = call.Argument(0).ToString()
if err != nil {
fmt.Println(err)
return otto.UndefinedValue()
}
if len(call.ArgumentList) > 1 {
address, err = call.Argument(1).ToString()
if err != nil {
fmt.Println(err)
return otto.UndefinedValue()
}
domain, err = call.Argument(2).ToString()
if err != nil {
fmt.Println(err)
return otto.UndefinedValue()
}
}
res, err = self.api.Upload(localpath, address, domain)
if err != nil {
fmt.Println(err)
return otto.UndefinedValue()
}
v, _ := call.Otto.ToValue(res)
return v
}
// http.PostForm("http://example.com/form",
// url.Values{"key": {"Value"}, "id": {"123"}})

70
bzz/netstore.go
View file

@ -10,7 +10,7 @@ import (
"github.com/ethereum/go-ethereum/p2p/discover"
)
type NetStore struct {
type netStore struct {
localStore *localStore
lock sync.Mutex
hive *hive
@ -44,13 +44,13 @@ type requestStatus struct {
C chan bool
}
func NewNetStore(path, hivepath string) (netstore *NetStore, err error) {
func newNetStore(path, hivepath string) (netstore *netStore, err error) {
dbStore, err := newDbStore(path)
if err != nil {
return
}
hive := newHive(hivepath)
netstore = &NetStore{
netstore = &netStore{
localStore: &localStore{
memStore: newMemStore(dbStore),
dbStore: dbStore,
@ -61,7 +61,7 @@ func NewNetStore(path, hivepath string) (netstore *NetStore, err error) {
return
}
func (self *NetStore) Start(node *discover.Node, connectPeer func(string) error) (err error) {
func (self *netStore) Start(node *discover.Node, connectPeer func(string) error) (err error) {
self.self = node
err = self.hive.start(kademlia.Address(node.Sha()), connectPeer)
if err != nil {
@ -70,13 +70,13 @@ func (self *NetStore) Start(node *discover.Node, connectPeer func(string) error)
return
}
func (self *NetStore) Stop() (err error) {
func (self *netStore) Stop() (err error) {
return self.hive.stop()
}
func (self *NetStore) Put(entry *Chunk) {
func (self *netStore) Put(entry *Chunk) {
chunk, err := self.localStore.Get(entry.Key)
dpaLogger.Debugf("NetStore.Put: localStore.Get returned with %v.", err)
dpaLogger.Debugf("netStore.Put: localStore.Get returned with %v.", err)
if err != nil {
chunk = entry
} else if chunk.SData == nil {
@ -88,9 +88,9 @@ func (self *NetStore) Put(entry *Chunk) {
self.put(chunk)
}
func (self *NetStore) put(entry *Chunk) {
func (self *netStore) put(entry *Chunk) {
self.localStore.Put(entry)
dpaLogger.Debugf("NetStore.put: localStore.Put of %064x completed, %d bytes (%p).", entry.Key, len(entry.SData), entry)
dpaLogger.Debugf("netStore.put: localStore.Put of %064x completed, %d bytes (%p).", entry.Key, len(entry.SData), entry)
if entry.req != nil {
if entry.req.status == reqSearching {
entry.req.status = reqFound
@ -102,7 +102,7 @@ func (self *NetStore) put(entry *Chunk) {
}
}
func (self *NetStore) store(chunk *Chunk) {
func (self *netStore) store(chunk *Chunk) {
for _, peer := range self.hive.getPeers(chunk.Key, 0) {
if chunk.source == nil || peer.Addr() != chunk.source.Addr() {
@ -111,12 +111,12 @@ func (self *NetStore) store(chunk *Chunk) {
}
}
func (self *NetStore) addStoreRequest(req *storeRequestMsgData) {
func (self *netStore) addStoreRequest(req *storeRequestMsgData) {
self.lock.Lock()
defer self.lock.Unlock()
dpaLogger.Debugf("NetStore.addStoreRequest: req = %v", req)
dpaLogger.Debugf("netStore.addStoreRequest: req = %v", req)
chunk, err := self.localStore.Get(req.Key)
dpaLogger.Debugf("NetStore.addStoreRequest: chunk reference %p", chunk)
dpaLogger.Debugf("netStore.addStoreRequest: chunk reference %p", chunk)
// we assume that a returned chunk is the one stored in the memory cache
if err != nil {
chunk = &Chunk{
@ -135,7 +135,7 @@ func (self *NetStore) addStoreRequest(req *storeRequestMsgData) {
}
// waits for response or times out
func (self *NetStore) Get(key Key) (chunk *Chunk, err error) {
func (self *netStore) Get(key Key) (chunk *Chunk, err error) {
chunk = self.get(key)
id := generateId()
timeout := time.Now().Add(searchTimeout)
@ -148,18 +148,18 @@ func (self *NetStore) Get(key Key) (chunk *Chunk, err error) {
timer := time.After(searchTimeout)
select {
case <-timer:
dpaLogger.Debugf("NetStore.Get: %064x request time out ", key)
dpaLogger.Debugf("netStore.Get: %064x request time out ", key)
err = notFound
case <-chunk.req.C:
dpaLogger.Debugf("NetStore.Get: %064x retrieved, %d bytes (%p)", key, len(chunk.SData), chunk)
dpaLogger.Debugf("netStore.Get: %064x retrieved, %d bytes (%p)", key, len(chunk.SData), chunk)
}
return
}
func (self *NetStore) get(key Key) (chunk *Chunk) {
func (self *netStore) get(key Key) (chunk *Chunk) {
var err error
chunk, err = self.localStore.Get(key)
dpaLogger.Debugf("NetStore.get: localStore.Get of %064x returned with %v.", key, err)
dpaLogger.Debugf("netStore.get: localStore.Get of %064x returned with %v.", key, err)
// we assume that a returned chunk is the one stored in the memory cache
if err != nil {
// no data and no request status
@ -182,7 +182,7 @@ func newRequestStatus() *requestStatus {
}
}
func (self *NetStore) addRetrieveRequest(req *retrieveRequestMsgData) {
func (self *netStore) addRetrieveRequest(req *retrieveRequestMsgData) {
self.lock.Lock()
defer self.lock.Unlock()
@ -198,28 +198,28 @@ func (self *NetStore) addRetrieveRequest(req *retrieveRequestMsgData) {
timeout := self.strategyUpdateRequest(chunk.req, req) // may change req status
if timeout == nil {
dpaLogger.Debugf("NetStore.addRetrieveRequest: %064x - content found, delivering...", req.Key)
dpaLogger.Debugf("netStore.addRetrieveRequest: %064x - content found, delivering...", req.Key)
self.deliver(req, chunk)
} else {
// we might need chunk.req to cache relevant peers response, or would it expire?
self.peers(req, chunk, timeout)
dpaLogger.Debugf("NetStore.addRetrieveRequest: %064x - searching.... responding with peers...", req.Key)
dpaLogger.Debugf("netStore.addRetrieveRequest: %064x - searching.... responding with peers...", req.Key)
self.startSearch(chunk, int64(req.Id), timeout)
}
}
// it's assumed that caller holds the lock
func (self *NetStore) startSearch(chunk *Chunk, id int64, timeout *time.Time) {
func (self *netStore) startSearch(chunk *Chunk, id int64, timeout *time.Time) {
chunk.req.status = reqSearching
peers := self.hive.getPeers(chunk.Key, 0)
dpaLogger.Debugf("NetStore.startSearch: %064x - received %d peers from KΛÐΞMLIΛ...", chunk.Key, len(peers))
dpaLogger.Debugf("netStore.startSearch: %064x - received %d peers from KΛÐΞMLIΛ...", chunk.Key, len(peers))
req := &retrieveRequestMsgData{
Key: chunk.Key,
Id: uint64(id),
timeout: timeout,
}
for _, peer := range peers {
dpaLogger.Debugf("NetStore.startSearch: sending retrieveRequests to peer [%064x]", req.Key)
dpaLogger.Debugf("netStore.startSearch: sending retrieveRequests to peer [%064x]", req.Key)
dpaLogger.Debugf("req.requesters: %v", chunk.req.requesters)
var requester bool
OUT:
@ -247,8 +247,8 @@ adds a new peer to an existing open request
only add if less than requesterCount peers forwarded the same request id so far
note this is done irrespective of status (searching or found)
*/
func (self *NetStore) addRequester(rs *requestStatus, req *retrieveRequestMsgData) {
dpaLogger.Debugf("NetStore.addRequester: key %064x - add peer [%v] to req.Id %064x", req.Key, req.peer, req.Id)
func (self *netStore) addRequester(rs *requestStatus, req *retrieveRequestMsgData) {
dpaLogger.Debugf("netStore.addRequester: key %064x - add peer [%v] to req.Id %064x", req.Key, req.peer, req.Id)
list := rs.requesters[int64(req.Id)]
rs.requesters[int64(req.Id)] = append(list, req)
}
@ -265,8 +265,8 @@ this is the most simplistic implementation:
- respond with peers and timeout if still searching
! in the last case as well, we should respond with reject if already got requesterCount peers with that exact id
*/
func (self *NetStore) strategyUpdateRequest(rs *requestStatus, req *retrieveRequestMsgData) (timeout *time.Time) {
dpaLogger.Debugf("NetStore.strategyUpdateRequest: key %064x", req.Key)
func (self *netStore) strategyUpdateRequest(rs *requestStatus, req *retrieveRequestMsgData) (timeout *time.Time) {
dpaLogger.Debugf("netStore.strategyUpdateRequest: key %064x", req.Key)
self.addRequester(rs, req)
if rs.status == reqSearching {
timeout = self.searchTimeout(rs, req)
@ -275,11 +275,11 @@ func (self *NetStore) strategyUpdateRequest(rs *requestStatus, req *retrieveRequ
}
func (self *NetStore) propagateResponse(chunk *Chunk) {
dpaLogger.Debugf("NetStore.propagateResponse: key %064x", chunk.Key)
func (self *netStore) propagateResponse(chunk *Chunk) {
dpaLogger.Debugf("netStore.propagateResponse: key %064x", chunk.Key)
for id, requesters := range chunk.req.requesters {
counter := requesterCount
dpaLogger.Debugf("NetStore.propagateResponse id %064x", id)
dpaLogger.Debugf("netStore.propagateResponse id %064x", id)
msg := &storeRequestMsgData{
Key: chunk.Key,
SData: chunk.SData,
@ -287,7 +287,7 @@ func (self *NetStore) propagateResponse(chunk *Chunk) {
}
for _, req := range requesters {
if req.timeout.After(time.Now()) {
dpaLogger.Debugf("NetStore.propagateResponse store -> %064x with %v", req.Id, req.peer)
dpaLogger.Debugf("netStore.propagateResponse store -> %064x with %v", req.Id, req.peer)
go req.peer.store(msg)
counter--
if counter <= 0 {
@ -298,7 +298,7 @@ func (self *NetStore) propagateResponse(chunk *Chunk) {
}
}
func (self *NetStore) deliver(req *retrieveRequestMsgData, chunk *Chunk) {
func (self *netStore) deliver(req *retrieveRequestMsgData, chunk *Chunk) {
storeReq := &storeRequestMsgData{
Key: req.Key,
Id: req.Id,
@ -310,7 +310,7 @@ func (self *NetStore) deliver(req *retrieveRequestMsgData, chunk *Chunk) {
req.peer.store(storeReq)
}
func (self *NetStore) peers(req *retrieveRequestMsgData, chunk *Chunk, timeout *time.Time) {
func (self *netStore) peers(req *retrieveRequestMsgData, chunk *Chunk, timeout *time.Time) {
var addrs []*peerAddr
for _, peer := range self.hive.getPeers(req.Key, int(req.MaxPeers)) {
addrs = append(addrs, peer.peerAddr())
@ -324,7 +324,7 @@ func (self *NetStore) peers(req *retrieveRequestMsgData, chunk *Chunk, timeout *
req.peer.peers(peersData)
}
func (self *NetStore) searchTimeout(rs *requestStatus, req *retrieveRequestMsgData) (timeout *time.Time) {
func (self *netStore) searchTimeout(rs *requestStatus, req *retrieveRequestMsgData) (timeout *time.Time) {
t := time.Now().Add(searchTimeout)
if req.timeout != nil && req.timeout.Before(t) {
return req.timeout

12
bzz/protocol.go
View file

@ -62,7 +62,7 @@ var errorToString = map[int]string{
// instance is running on each peer
type bzzProtocol struct {
node *discover.Node
netStore *NetStore
netStore *netStore
peer *p2p.Peer
rw p2p.MsgReadWriter
errors *errs.Errors
@ -216,9 +216,9 @@ main entrypoint, wrappers starting a server running the bzz protocol
use this constructor to attach the protocol ("class") to server caps
the Dev p2p layer then runs the protocol instance on each peer
*/
func BzzProtocol(netStore *NetStore) (p2p.Protocol, error) {
func BzzProtocol(netstore *netStore) (p2p.Protocol, error) {
db, err := NewLDBDatabase(path.Join(netStore.path, "requests"))
db, err := NewLDBDatabase(path.Join(netstore.path, "requests"))
if err != nil {
return p2p.Protocol{}, err
}
@ -227,17 +227,17 @@ func BzzProtocol(netStore *NetStore) (p2p.Protocol, error) {
Version: Version,
Length: ProtocolLength,
Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
return runBzzProtocol(db, netStore, p, rw)
return runBzzProtocol(db, netstore, p, rw)
},
}, nil
}
// the main loop that handles incoming messages
// note RemovePeer in the post-disconnect hook
func runBzzProtocol(db *LDBDatabase, netStore *NetStore, p *p2p.Peer, rw p2p.MsgReadWriter) (err error) {
func runBzzProtocol(db *LDBDatabase, netstore *netStore, p *p2p.Peer, rw p2p.MsgReadWriter) (err error) {
self := &bzzProtocol{
netStore: netStore,
netStore: netstore,
rw: rw,
peer: p,
errors: &errs.Errors{

19
bzz/roundtripper.go Normal file
View file

@ -0,0 +1,19 @@
package bzz
import (
"fmt"
"net/http"
// "github.com/ethereum/go-ethereum/common/docserver"
// "github.com/ethereum/go-ethereum/jsre"
)
type RoundTripper struct {
Port string
}
func (self *RoundTripper) RoundTrip(req *http.Request) (resp *http.Response, err error) {
url := fmt.Sprintf("http://localhost:%s/%s/%s", self.Port, req.URL.Host, req.URL.Path)
dpaLogger.Infof("roundtripper: proxying request '%s' to '%s'", req.RequestURI, url)
return http.Get(url)
}

51
bzz/roundtripper_test.go Normal file
View file

@ -0,0 +1,51 @@
package bzz
import (
"io/ioutil"
"net/http"
"strings"
"testing"
"time"
"github.com/ethereum/go-ethereum/common/docserver"
)
const port = "8500"
func TestRoundTripper(t *testing.T) {
http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
if r.Method == "GET" {
w.Header().Set("Content-Type", "text/plain")
http.ServeContent(w, r, "", time.Unix(0, 0), strings.NewReader(r.RequestURI))
} else {
http.Error(w, "Method "+r.Method+" is not supported.", http.StatusMethodNotAllowed)
}
})
go http.ListenAndServe(":"+port, nil)
rt := &RoundTripper{port}
ds := docserver.New("/")
ds.RegisterProtocol("bzz", rt)
resp, err := ds.Client().Get("bzz://test.com/path")
if err != nil {
t.Errorf("expected no error, got %v", err)
return
}
defer func() {
if resp != nil {
resp.Body.Close()
}
}()
content, err := ioutil.ReadAll(resp.Body)
if err != nil {
t.Errorf("expected no error, got %v", err)
return
}
if string(content) != "/test.com/path" {
t.Errorf("incorrect response from http server: expected '%v', got '%v'", "/test.com/path", string(content))
}
}

5
cmd/geth/admin.go
View file

@ -27,10 +27,7 @@ import (
"gopkg.in/fatih/set.v0"
)
/*
node admin bindings
*/
// node admin bindings
func (js *jsre) adminBindings() {
ethO, _ := js.re.Get("eth")
eth := ethO.Object()

11
cmd/geth/js.go
View file

@ -25,6 +25,7 @@ import (
"path/filepath"
"strings"
"github.com/ethereum/go-ethereum/bzz"
"github.com/ethereum/go-ethereum/cmd/utils"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/docserver"
@ -72,7 +73,7 @@ type jsre struct {
prompter
}
func newJSRE(ethereum *eth.Ethereum, libPath, corsDomain string, interactive bool, f xeth.Frontend) *jsre {
func newJSRE(ethereum *eth.Ethereum, libPath, corsDomain string, bzzEnabled bool, bzzPort string, interactive bool, f xeth.Frontend) *jsre {
js := &jsre{ethereum: ethereum, ps1: "> "}
// set default cors domain used by startRpc from CLI flag
js.corsDomain = corsDomain
@ -85,6 +86,12 @@ func newJSRE(ethereum *eth.Ethereum, libPath, corsDomain string, interactive boo
js.re = re.New(libPath)
js.apiBindings(f)
js.adminBindings()
if bzzEnabled {
ds.RegisterProtocol("bzz", &bzz.RoundTripper{Port: bzzPort})
bzz.NewJSApi(js.re, js.ethereum.Swarm)
// nil for no natsped, memoryrrbmm d
js.ethereum.Swarm.Resolver = resolver.New(xeth.New(ethereum, nil))
}
if !liner.TerminalSupported() || !interactive {
js.prompter = dumbterm{bufio.NewReader(os.Stdin)}
@ -145,7 +152,7 @@ var net = web3.net;
js.re.Eval(resolver.GlobalRegistrar + "registrar = GlobalRegistrar.at(\"" + resolver.GlobalRegistrarAddr + "\");")
}
var ds, _ = docserver.New("/")
var ds = docserver.New("/")
func (self *jsre) ConfirmTransaction(tx string) bool {
if self.ethereum.NatSpec {

8
cmd/geth/js_test.go
View file

@ -98,12 +98,9 @@ func testJEthRE(t *testing.T) (string, *testjethre, *eth.Ethereum) {
}
assetPath := filepath.Join(os.Getenv("GOPATH"), "src", "github.com", "ethereum", "go-ethereum", "cmd", "mist", "assets", "ext")
ds, err := docserver.New("/")
if err != nil {
t.Errorf("Error creating DocServer: %v", err)
}
ds := docserver.New("/")
tf := &testjethre{ds: ds, stateDb: ethereum.ChainManager().State().Copy()}
repl := newJSRE(ethereum, assetPath, "", false, tf)
repl := newJSRE(ethereum, assetPath, "", false, "", false, tf)
tf.jsre = repl
return tmp, tf, ethereum
}
@ -116,6 +113,7 @@ func testJEthRE(t *testing.T) (string, *testjethre, *eth.Ethereum) {
// txc, self.xeth = self.xeth.ApplyTestTxs(self.xeth.repl.stateDb, coinbase, txc)
func TestNodeInfo(t *testing.T) {
t.Skip("broken after p2p update")
tmp, repl, ethereum := testJEthRE(t)
if err := ethereum.Start(); err != nil {
t.Fatalf("error starting ethereum: %v", err)

5
cmd/geth/main.go
View file

@ -267,6 +267,7 @@ JavaScript API. See https://github.com/ethereum/go-ethereum/wiki/Javascipt-Conso
utils.RPCPortFlag,
utils.WhisperEnabledFlag,
utils.SwarmEnabledFlag,
utils.SwarmProxyPortFlag,
utils.VMDebugFlag,
utils.ProtocolVersionFlag,
utils.NetworkIdFlag,
@ -337,6 +338,8 @@ func console(ctx *cli.Context) {
ethereum,
ctx.String(utils.JSpathFlag.Name),
ctx.GlobalString(utils.RPCCORSDomainFlag.Name),
ctx.GlobalBool(utils.SwarmEnabledFlag.Name),
ctx.GlobalString(utils.SwarmProxyPortFlag.Name),
true,
nil,
)
@ -358,6 +361,8 @@ func execJSFiles(ctx *cli.Context) {
ethereum,
ctx.String(utils.JSpathFlag.Name),
ctx.GlobalString(utils.RPCCORSDomainFlag.Name),
ctx.GlobalBool(utils.SwarmEnabledFlag.Name),
ctx.GlobalString(utils.SwarmProxyPortFlag.Name),
false,
nil,
)

5
cmd/utils/flags.go
View file

@ -243,6 +243,11 @@ var (
Name: "bzz",
Usage: "Enable swarm",
}
SwarmProxyPortFlag = cli.StringFlag{
Name: "bzzport",
Usage: "Swarm HTTP Proxy Port on localhost",
Value: "8500",
}
// ATM the url is left to the user and deployment to
JSpathFlag = cli.StringFlag{
Name: "jspath",

5
common/natspec/natspec_e2e_test.go
View file

@ -91,10 +91,7 @@ func (self *testFrontend) UnlockAccount(acc []byte) bool {
func (self *testFrontend) ConfirmTransaction(tx string) bool {
if self.wantNatSpec {
ds, err := docserver.New("/tmp/")
if err != nil {
self.t.Errorf("Error creating DocServer: %v", err)
}
ds := docserver.New("/tmp/")
self.lastConfirm = GetNotice(self.xeth, tx, ds)
}
return true

37
eth/backend.go
View file

@ -84,8 +84,9 @@ type Config struct {
NAT nat.Interface
Shh bool
Bzz bool
Dial bool
Bzz bool
BzzPort string
Etherbase string
GasPrice *big.Int
@ -194,8 +195,7 @@ type Ethereum struct {
pow *ethash.Ethash
protocolManager *ProtocolManager
downloader *downloader.Downloader
DPA *bzz.DPA
netStore *bzz.NetStore
Swarm *bzz.Api
SolcPath string
solc *compiler.Solidity
@ -310,20 +310,12 @@ func New(config *Config) (*Ethereum, error) {
protocols := []p2p.Protocol{eth.protocolManager.SubProtocol}
if config.Bzz {
eth.netStore, err = bzz.NewNetStore(filepath.Join(config.DataDir, "bzz"), filepath.Join(config.DataDir, "bzzpeers.json"))
if err != nil {
glog.V(logger.Warn).Infof("BZZ: error creating net store: %v. Protocol skipped", err)
} else {
chunker := &bzz.TreeChunker{}
chunker.Init()
eth.DPA = &bzz.DPA{
Chunker: chunker,
ChunkStore: eth.netStore,
}
bzzProto, err := bzz.BzzProtocol(eth.netStore)
eth.Swarm, err = bzz.NewApi(config.DataDir, config.BzzPort)
var proto p2p.Protocol
proto, err = eth.Swarm.Bzz()
if err == nil {
protocols = append(protocols, bzzProto)
}
glog.V(logger.Warn).Infof("BZZ: error creating swarm: %v. Protocol skipped", err)
protocols = append(protocols, proto)
}
}
@ -498,10 +490,8 @@ func (s *Ethereum) Start() error {
s.whisper.Start()
}
if s.DPA != nil {
s.DPA.Start()
s.netStore.Start(s.net.Self(), s.AddPeer)
go bzz.StartHttpServer(s.DPA)
if s.Swarm != nil {
s.Swarm.Start(s.net.Self(), s.AddPeer)
}
// broadcast transactions
@ -576,11 +566,8 @@ func (s *Ethereum) Stop() {
}
s.StopAutoDAG()
if s.DPA != nil {
s.DPA.Stop()
}
if s.netStore != nil {
s.netStore.Stop()
if s.Swarm != nil {
s.Swarm.Stop()
}
glog.V(logger.Info).Infoln("Server stopped")