swarm: plan bee for content storage and distribution on web3

This change imports the Swarm protocol codebase. Compared to the 'swarm'
branch, a few mostly cosmetic changes had to be made:

* The various redundant log message prefixes are gone.
* All files now have LGPLv3 license headers.
* Minor code changes were needed to please go vet and make the tests
  pass on Windows.
* Further changes were required to adapt to the go-ethereum develop
  branch and its new Go APIs.

Some code has not (yet) been brought over:

* swarm/cmd/bzzhash: will reappear as cmd/bzzhash later
* swarm/cmd/bzzup.sh: will be reimplemented in cmd/bzzup
* swarm/cmd/makegenesis: will reappear somehow
* swarm/examples/album: will move to a separate repository
* swarm/examples/filemanager: ditto
* swarm/examples/files: will not be merged
* swarm/test/*: will not be merged
* swarm/services/swear: will reappear as contracts/swear when needed

1 files changed, 283 insertions, 0 deletions

diff --git a/swarm/api/filesystem.go b/swarm/api/filesystem.go
new file mode 100644
index 000000000..428f3e3ac
--- /dev/null
+++ b/swarm/api/filesystem.go
@@ -0,0 +1,283 @@
+// Copyright 2016 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 <http://www.gnu.org/licenses/>.
+
+package api
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"path/filepath"
+	"sync"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/logger"
+	"github.com/ethereum/go-ethereum/logger/glog"
+	"github.com/ethereum/go-ethereum/swarm/storage"
+)
+
+const maxParallelFiles = 5
+
+type FileSystem struct {
+	api *Api
+}
+
+func NewFileSystem(api *Api) *FileSystem {
+	return &FileSystem{api}
+}
+
+// Upload replicates a local directory as a manifest file and uploads it
+// using dpa store
+// TODO: localpath should point to a manifest
+func (self *FileSystem) Upload(lpath, index string) (string, error) {
+	var list []*manifestTrieEntry
+	localpath, err := filepath.Abs(filepath.Clean(lpath))
+	if err != nil {
+		return "", err
+	}
+
+	f, err := os.Open(localpath)
+	if err != nil {
+		return "", err
+	}
+	stat, err := f.Stat()
+	if err != nil {
+		return "", err
+	}
+
+	var start int
+	if stat.IsDir() {
+		start = len(localpath)
+		glog.V(logger.Debug).Infof("uploading '%s'", localpath)
+		err = filepath.Walk(localpath, func(path string, info os.FileInfo, err error) error {
+			if (err == nil) && !info.IsDir() {
+				//fmt.Printf("lp %s  path %s\n", localpath, path)
+				if len(path) <= start {
+					return fmt.Errorf("Path is too short")
+				}
+				if path[:start] != localpath {
+					return fmt.Errorf("Path prefix of '%s' does not match localpath '%s'", path, localpath)
+				}
+				entry := &manifestTrieEntry{
+					Path: filepath.ToSlash(path),
+				}
+				list = append(list, entry)
+			}
+			return err
+		})
+		if err != nil {
+			return "", err
+		}
+	} else {
+		dir := filepath.Dir(localpath)
+		start = len(dir)
+		if len(localpath) <= start {
+			return "", fmt.Errorf("Path is too short")
+		}
+		if localpath[:start] != dir {
+			return "", fmt.Errorf("Path prefix of '%s' does not match dir '%s'", localpath, dir)
+		}
+		entry := &manifestTrieEntry{
+			Path: filepath.ToSlash(localpath),
+		}
+		list = append(list, entry)
+	}
+
+	cnt := len(list)
+	errors := make([]error, cnt)
+	done := make(chan bool, maxParallelFiles)
+	dcnt := 0
+	awg := &sync.WaitGroup{}
+
+	for i, entry := range list {
+		if i >= dcnt+maxParallelFiles {
+			<-done
+			dcnt++
+		}
+		awg.Add(1)
+		go func(i int, entry *manifestTrieEntry, done chan bool) {
+			f, err := os.Open(entry.Path)
+			if err == nil {
+				stat, _ := f.Stat()
+				var hash storage.Key
+				wg := &sync.WaitGroup{}
+				hash, err = self.api.dpa.Store(f, stat.Size(), wg, nil)
+				if hash != nil {
+					list[i].Hash = hash.String()
+				}
+				wg.Wait()
+				awg.Done()
+				if err == nil {
+					first512 := make([]byte, 512)
+					fread, _ := f.ReadAt(first512, 0)
+					if fread > 0 {
+						mimeType := http.DetectContentType(first512[:fread])
+						if filepath.Ext(entry.Path) == ".css" {
+							mimeType = "text/css"
+						}
+						list[i].ContentType = mimeType
+					}
+				}
+				f.Close()
+			}
+			errors[i] = err
+			done <- true
+		}(i, entry, done)
+	}
+	for dcnt < cnt {
+		<-done
+		dcnt++
+	}
+
+	trie := &manifestTrie{
+		dpa: self.api.dpa,
+	}
+	quitC := make(chan bool)
+	for i, entry := range list {
+		if errors[i] != nil {
+			return "", errors[i]
+		}
+		entry.Path = RegularSlashes(entry.Path[start:])
+		if entry.Path == index {
+			ientry := &manifestTrieEntry{
+				Path:        "",
+				Hash:        entry.Hash,
+				ContentType: entry.ContentType,
+			}
+			trie.addEntry(ientry, quitC)
+		}
+		trie.addEntry(entry, quitC)
+	}
+
+	err2 := trie.recalcAndStore()
+	var hs string
+	if err2 == nil {
+		hs = trie.hash.String()
+	}
+	awg.Wait()
+	return hs, err2
+}
+
+// Download replicates the manifest path structure on the local filesystem
+// under localpath
+func (self *FileSystem) Download(bzzpath, localpath string) error {
+	lpath, err := filepath.Abs(filepath.Clean(localpath))
+	if err != nil {
+		return err
+	}
+	err = os.MkdirAll(lpath, os.ModePerm)
+	if err != nil {
+		return err
+	}
+
+	//resolving host and port
+	key, _, path, err := self.api.parseAndResolve(bzzpath, true)
+	if err != nil {
+		return err
+	}
+
+	if len(path) > 0 {
+		path += "/"
+	}
+
+	quitC := make(chan bool)
+	trie, err := loadManifest(self.api.dpa, key, quitC)
+	if err != nil {
+		glog.V(logger.Warn).Infof("fs.Download: loadManifestTrie error: %v", err)
+		return err
+	}
+
+	type downloadListEntry struct {
+		key  storage.Key
+		path string
+	}
+
+	var list []*downloadListEntry
+	var mde error
+
+	prevPath := lpath
+	err = trie.listWithPrefix(path, quitC, func(entry *manifestTrieEntry, suffix string) {
+		glog.V(logger.Detail).Infof("fs.Download: %#v", entry)
+
+		key = common.Hex2Bytes(entry.Hash)
+		path := lpath + "/" + suffix
+		dir := filepath.Dir(path)
+		if dir != prevPath {
+			mde = os.MkdirAll(dir, os.ModePerm)
+			prevPath = dir
+		}
+		if (mde == nil) && (path != dir+"/") {
+			list = append(list, &downloadListEntry{key: key, path: path})
+		}
+	})
+	if err != nil {
+		return err
+	}
+
+	wg := sync.WaitGroup{}
+	errC := make(chan error)
+	done := make(chan bool, maxParallelFiles)
+	for i, entry := range list {
+		select {
+		case done <- true:
+			wg.Add(1)
+		case <-quitC:
+			return fmt.Errorf("aborted")
+		}
+		go func(i int, entry *downloadListEntry) {
+			defer wg.Done()
+			f, err := os.Create(entry.path) // TODO: path separators
+			if err == nil {
+
+				reader := self.api.dpa.Retrieve(entry.key)
+				writer := bufio.NewWriter(f)
+				size, err := reader.Size(quitC)
+				if err == nil {
+					_, err = io.CopyN(writer, reader, size) // TODO: handle errors
+					err2 := writer.Flush()
+					if err == nil {
+						err = err2
+					}
+					err2 = f.Close()
+					if err == nil {
+						err = err2
+					}
+				}
+			}
+			if err != nil {
+				select {
+				case errC <- err:
+				case <-quitC:
+				}
+				return
+			}
+			<-done
+		}(i, entry)
+	}
+	go func() {
+		wg.Wait()
+		close(errC)
+	}()
+	select {
+	case err = <-errC:
+		return err
+	case <-quitC:
+		return fmt.Errorf("aborted")
+	}
+
+}