Refactor existing networking bits

connection/listen.go

@@ -0,0 +1,80 @@
+package connection
+
+import (
+	"net"
+	"io"
+)
+
+const defaultBufSize = 1 * 1024
+
+
+// Connection represents a recv-only network connection.
+type Connection struct {
+	BufSize int
+	conn *net.TCPConn
+	recvChan chan []byte
+	stopFlag bool
+}
+
+// NewListen starts listening for a single connection on the given address.
+// This method blocks until this is done.
+func NewListen(address string) (*Connection, error) {
+	conn, err := listenAndAcceptTCP(address)
+	if err != nil {
+		return nil, err
+	}
+	return &Connection{
+		conn: conn,
+		BufSize: defaultBufSize,
+		stopFlag: false,
+	}, nil
+}
+
+func listenAndAcceptTCP(address string) (*net.TCPConn, error) {
+	addr, err := net.ResolveTCPAddr("tcp", address)
+	if err != nil {
+		return nil, err
+	}
+	sock, err := net.ListenTCP("tcp", addr)
+	if err != nil {
+		return nil, err
+	}
+	conn, err := sock.AcceptTCP()
+	if err != nil {
+		return nil, err
+	}
+	conn.SetNoDelay(true)
+	return conn, nil
+}
+
+// Recv returns a read-only channel of byte slices and starts
+// receiving and pushing data into it asynchronously.
+// The channel is closed when the sender closes the connection.
+// Use this to read bytes from the network connection.
+func (c *Connection) Recv() <-chan []byte {
+	c.recvChan = make(chan []byte)
+	go c.recvLoop()
+	return c.recvChan
+}
+
+func (c *Connection) recvLoop() {
+	buf := make([]byte, c.BufSize)
+	for !c.stopFlag {
+		n, err := io.ReadFull(c.conn, buf)
+		if err != nil {
+			c.stopFlag = true
+		}
+		// the underlying memory of the buffer
+		// must be copied for thread safety
+		sendBuf := make([]byte, n)
+		copy(sendBuf, buf)
+		c.recvChan <- sendBuf
+	}
+	close(c.recvChan)
+}
+
+// Close closes the connection
+func (c *Connection) Close() {
+	c.stopFlag = true
+	c.conn.Close()
+}