Many changes

Author: sqs

client.go

@@ -0,0 +1,124 @@
+package httpfstream
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"github.com/garyburd/go-websocket/websocket"
+	"io"
+	"net"
+	"net/http"
+	"net/url"
+	"os"
+	"time"
+)
+
+func Get(u *url.URL) (io.ReadCloser, error) {
+	req, err := http.NewRequest("HEAD", u.String(), nil)
+	if err != nil {
+		return nil, err
+	}
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	if resp.StatusCode != http.StatusOK {
+		switch resp.StatusCode {
+		case http.StatusNotFound:
+			return nil, os.ErrNotExist
+		default:
+			return nil, fmt.Errorf("head failed with status %d", resp.StatusCode)
+		}
+	}
+
+	ws, err := newClient(u, "GET")
+	if err != nil {
+		return nil, err
+	}
+
+	op, r, err := ws.NextReader()
+	if err != nil {
+		return nil, err
+	}
+	if op != websocket.OpText {
+		return nil, errors.New("websocket op is not text")
+	}
+
+	return &webSocketReadCloser{r, ws}, nil
+}
+
+type webSocketReadCloser struct {
+	io.Reader
+	ws *websocket.Conn
+}
+
+func (r *webSocketReadCloser) Read(p []byte) (n int, err error) {
+	return r.Reader.Read(p)
+}
+
+func (r *webSocketReadCloser) Close() error {
+	return r.ws.Close()
+}
+
+func Put(u *url.URL, r io.Reader) error {
+	w, err := OpenPut(u)
+	if err != nil {
+		return err
+	}
+	defer w.Close()
+
+	_, err = io.Copy(w, r)
+	return err
+}
+
+func OpenPut(u *url.URL) (io.WriteCloser, error) {
+	req, err := http.NewRequest("HEAD", u.String(), nil)
+	if err != nil {
+		return nil, err
+	}
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusNotFound {
+		switch resp.StatusCode {
+		default:
+			return nil, fmt.Errorf("head failed with status %d", resp.StatusCode)
+		}
+	}
+
+	ws, err := newClient(u, "PUT")
+	if err != nil {
+		return nil, err
+	}
+
+	return &putWriteCloser{new(bytes.Buffer), ws}, nil
+}
+
+type putWriteCloser struct {
+	io.Writer
+	ws *websocket.Conn
+}
+
+func (pw *putWriteCloser) Write(p []byte) (n int, err error) {
+	pw.ws.SetWriteDeadline(time.Now().Add(writeWait))
+	w, err := pw.ws.NextWriter(websocket.OpText)
+	if err != nil {
+		return 0, err
+	}
+	defer w.Close()
+	return w.Write(p)
+}
+
+func (pw *putWriteCloser) Close() error {
+	return pw.ws.Close()
+}
+
+func newClient(u *url.URL, method string) (*websocket.Conn, error) {
+	c, err := net.Dial("tcp", u.Host)
+	if err != nil {
+		return nil, err
+	}
+	ws, _, err := websocket.NewClient(c, u, http.Header{xMethod: []string{method}}, readBufSize, writeBufSize)
+	return ws, err
+}

client_test.go

@@ -0,0 +1,146 @@
+package httpfstream
+
+import (
+	"bytes"
+	"io"
+	"net/url"
+	"os"
+	"testing"
+	"time"
+)
+
+type getTest struct {
+	path       string
+	body       string
+	writeFiles map[string]string
+	err        error
+}
+
+func TestGet(t *testing.T) {
+	tests := []getTest{
+		{path: "/foo1", body: "bar", writeFiles: map[string]string{"/foo1": "bar"}},
+
+		{path: "/doesntexist", err: os.ErrNotExist},
+	}
+	for _, test := range tests {
+		testGet(t, test)
+	}
+}
+
+func testGet(t *testing.T, test getTest) {
+	label := test.path
+
+	server := newTestServer()
+	defer server.close()
+
+	for path, data := range test.writeFiles {
+		w, err := server.fs.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC|os.O_EXCL)
+		if err != nil {
+			t.Fatalf("%s: fs.WriterOpen: %s", label, err)
+		}
+		_, err = w.Write([]byte(data))
+		if err != nil {
+			t.Fatalf("%s: Write: %s", label, err)
+		}
+		err = w.Close()
+		if err != nil {
+			t.Fatalf("%s: Close: %s", label, err)
+		}
+	}
+
+	u, _ := url.Parse(server.URL + test.path)
+	r, err := Get(u)
+	if err == nil {
+		defer r.Close()
+	}
+	if test.err != err {
+		t.Errorf("%s: Get: want error %v, got %v", label, test.err, err)
+		return
+	}
+	if test.err != nil {
+		return
+	}
+
+	body := string(readAll(t, r))
+	if test.body != body {
+		t.Errorf("%s: want body == %q, got %q", label, test.body, body)
+	}
+}
+
+type putTest struct {
+	path     string
+	data     io.Reader
+	fileData string
+	err      error
+}
+
+func TestPut(t *testing.T) {
+	tests := []putTest{
+		{path: "/foo1", data: bytes.NewReader([]byte("bar")), fileData: "bar"},
+		{path: "/foo2", data: bytes.NewBuffer([]byte("bar")), fileData: "bar"},
+		{path: "/foo3", data: &slowReader{R: &fixedReader{R: bytes.NewReader([]byte("quxx")), N: 2}, Wait: time.Millisecond * 25}, fileData: "quxx"},
+	}
+	for _, test := range tests {
+		testPut(t, test)
+	}
+}
+
+func testPut(t *testing.T, test putTest) {
+	label := test.path
+
+	server := newTestServer()
+	defer server.close()
+
+	u, _ := url.Parse(server.URL + test.path)
+	err := Put(u, test.data)
+	if test.err != err {
+		t.Errorf("%s: Put: want error %v, got %v", label, test.err, err)
+		return
+	}
+	if test.err != nil {
+		return
+	}
+
+	time.Sleep(50 * time.Millisecond)
+
+	_, err = server.fs.Stat(test.path)
+	if err != nil {
+		t.Errorf("%s: Stat: %s", label, err)
+		return
+	}
+
+	f, err := server.fs.Open(test.path)
+	if err != nil {
+		t.Errorf("%s: Open: %s", label, err)
+		return
+	}
+	fileData := string(readAll(t, f))
+	if test.fileData != fileData {
+		t.Errorf("%s: want fileData == %q, got %q", label, test.fileData, fileData)
+	}
+}
+
+type slowReader struct {
+	R         io.Reader
+	Wait      time.Duration
+	afterRead func(read []byte)
+}
+
+func (r *slowReader) Read(p []byte) (n int, err error) {
+	if r.afterRead != nil {
+		defer func() {
+			r.afterRead(p)
+		}()
+	}
+	time.Sleep(r.Wait)
+	return r.R.Read(p)
+}
+
+type fixedReader struct {
+	R io.Reader
+	N int
+}
+
+func (r *fixedReader) Read(p []byte) (n int, err error) {
+	return io.LimitReader(r.R, int64(r.N)).Read(p)
+}

cmd/httpfstream-server/server.go

@@ -0,0 +1,48 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+	"github.com/gorilla/handlers"
+	"github.com/sourcegraph/httpfstream"
+	"github.com/sourcegraph/rwvfs"
+	"log"
+	"net/http"
+	"os"
+)
+
+var bindAddr = flag.String("http", ":8080", "HTTP bind address for server")
+var root = flag.String("root", "/tmp/httpfstream", "storage root directory")
+
+func main() {
+	flag.Usage = func() {
+		fmt.Fprintf(os.Stderr, "httpfstream-server supports simultaneous, streaming file uploading and downloading over HTTP.\n\n")
+		fmt.Fprintf(os.Stderr, "Usage:\n\n")
+		fmt.Fprintf(os.Stderr, "\thttpfstream-server [options]\n\n")
+		fmt.Fprintf(os.Stderr, "The options are:\n\n")
+		flag.PrintDefaults()
+		fmt.Fprintln(os.Stderr)
+		fmt.Fprintln(os.Stderr)
+		fmt.Fprintf(os.Stderr, "Example usage:\n\n")
+		fmt.Fprintf(os.Stderr, "\tTo run on http://localhost:8080:\n")
+		fmt.Fprintf(os.Stderr, "\t    $ httpfstream-server -http=:8080\n\n")
+		fmt.Fprintln(os.Stderr)
+		os.Exit(1)
+	}
+	flag.Parse()
+	if flag.NArg() != 0 {
+		flag.Usage()
+	}
+
+	os.MkdirAll(*root, 0700)
+
+	h := httpfstream.New(rwvfs.OS(*root))
+	h.Log = log.New(os.Stderr, "", 0)
+	http.Handle("/", handlers.CombinedLoggingHandler(os.Stdout, h))
+
+	log.Printf("Starting server on %s\n", *bindAddr)
+	err := http.ListenAndServe(*bindAddr, nil)
+	if err != nil {
+		log.Fatalf("ListenAndServe: %s", err)
+	}
+}