Channel reader (refactor) (#26)

* add chreader

* version bump

Author: wybiral

hookah.go

@@ -14,7 +14,7 @@ import (
 )
 
 // Version of hookah API
-const Version = "1.0.3"
+const Version = "1.0.4"
 
 // API is an instance of the Hookah API.
 type API struct {

pkg/chreader/chreader.go

@@ -0,0 +1,40 @@
+// Package chreader converts a []byte channel to an io.ReadCloser.
+package chreader
+
+import (
+	"io"
+	"sync"
+)
+
+type chReader struct {
+	ch  chan []byte
+	mu  sync.Mutex
+	top []byte
+}
+
+// New returns an io.ReadCloser that reads from and closes ch.
+func New(ch chan []byte) io.ReadCloser {
+	return &chReader{ch: ch}
+}
+
+func (c *chReader) Read(b []byte) (int, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if len(c.top) == 0 {
+		top := <-c.ch
+		c.top = make([]byte, len(top))
+		copy(c.top, top)
+		if len(c.top) == 0 {
+			// ch is closed
+			return 0, io.EOF
+		}
+	}
+	n := copy(b, c.top)
+	c.top = c.top[n:]
+	return n, nil
+}
+
+func (c *chReader) Close() error {
+	close(c.ch)
+	return nil
+}

pkg/input/httplisten.go

@@ -4,17 +4,16 @@ import (
 	"io"
 	"net/http"
 	"net/url"
-	"sync"
+
+	"github.com/wybiral/hookah/pkg/chreader"
 )
 
 type httpListenApp struct {
 	server *http.Server
 	// Channel of messages
 	ch chan []byte
-	// Lock for changing top
-	mu *sync.Mutex
-	// Current message being read
-	top []byte
+	// ch Reader
+	r io.Reader
 }
 
 // HTTPListen creates an HTTP listener and returns ReadCloser
@@ -25,28 +24,17 @@ func HTTPListen(addr string, opts url.Values) (io.ReadCloser, error) {
 		Handler: http.HandlerFunc(app.handle),
 	}
 	app.ch = make(chan []byte)
-	app.mu = &sync.Mutex{}
-	app.top = nil
+	app.r = chreader.New(app.ch)
 	go app.server.ListenAndServe()
 	return app, nil
 }
 
 func (app *httpListenApp) Read(b []byte) (int, error) {
-	app.mu.Lock()
-	defer app.mu.Unlock()
-	if len(app.top) == 0 {
-		app.top = <-app.ch
-		if len(app.top) == 0 {
-			// ch is closed
-			return 0, io.EOF
-		}
-	}
-	n := copy(b, app.top)
-	app.top = app.top[n:]
-	return n, nil
+	return app.r.Read(b)
 }
 
 func (app *httpListenApp) Close() error {
+	// Closing ch causes r.Read to return EOF
 	close(app.ch)
 	return app.server.Close()
 }

pkg/input/input.go

@@ -10,4 +10,4 @@ import (
 type Handler func(arg string, opts url.Values) (io.ReadCloser, error)
 
 // Buffer size used for incoming messages to servers
-const bufferSize = 4 * 1024
+const bufferSize = 8 * 1024

pkg/input/listen.go

@@ -3,17 +3,16 @@ package input
 import (
 	"io"
 	"net"
-	"sync"
+
+	"github.com/wybiral/hookah/pkg/chreader"
 )
 
 type listenApp struct {
 	ln net.Listener
 	// Channel of messages
 	ch chan []byte
-	// Lock for changing top
-	mu *sync.Mutex
-	// Current message being read
-	top []byte
+	// ch Reader
+	r io.Reader
 }
 
 // listen creates a generic listener and returns ReadCloser
@@ -25,28 +24,17 @@ func listen(network, addr string) (io.ReadCloser, error) {
 	}
 	app.ln = ln
 	app.ch = make(chan []byte)
-	app.mu = &sync.Mutex{}
-	app.top = nil
+	app.r = chreader.New(app.ch)
 	go app.serve()
 	return app, nil
 }
 
 func (app *listenApp) Read(b []byte) (int, error) {
-	app.mu.Lock()
-	defer app.mu.Unlock()
-	if len(app.top) == 0 {
-		app.top = <-app.ch
-		if len(app.top) == 0 {
-			// ch is closed
-			return 0, io.EOF
-		}
-	}
-	n := copy(b, app.top)
-	app.top = app.top[n:]
-	return n, nil
+	return app.r.Read(b)
 }
 
 func (app *listenApp) Close() error {
+	// Closing ch causes r.Read to return EOF
 	close(app.ch)
 	return app.ln.Close()
 }

pkg/input/wslisten.go

@@ -4,19 +4,17 @@ import (
 	"io"
 	"net/http"
 	"net/url"
-	"sync"
 
 	"github.com/gorilla/websocket"
+	"github.com/wybiral/hookah/pkg/chreader"
 )
 
 type wsListenApp struct {
 	server *http.Server
 	// Channel of messages
 	ch chan []byte
-	// Lock for top
-	mu *sync.Mutex
-	// Current message being read
-	top []byte
+	// ch Reader
+	r io.Reader
 }
 
 // WebSocket upgrader
@@ -36,29 +34,17 @@ func WSListen(addr string, opts url.Values) (io.ReadCloser, error) {
 		Handler: http.HandlerFunc(app.handle),
 	}
 	app.ch = make(chan []byte)
-	app.mu = &sync.Mutex{}
-	app.top = nil
+	app.r = chreader.New(app.ch)
 	go app.server.ListenAndServe()
 	return app, nil
 }
 
 func (app *wsListenApp) Read(b []byte) (int, error) {
-	app.mu.Lock()
-	defer app.mu.Unlock()
-	// top is empty, pull from ch
-	if len(app.top) == 0 {
-		app.top = <-app.ch
-		if len(app.top) == 0 {
-			// ch is closed
-			return 0, io.EOF
-		}
-	}
-	n := copy(b, app.top)
-	app.top = app.top[n:]
-	return n, nil
+	return app.r.Read(b)
 }
 
 func (app *wsListenApp) Close() error {
+	// Closing ch causes r.Read to return EOF
 	close(app.ch)
 	return app.server.Close()
 }