Refactor networking code

networking/io.go

@@ -0,0 +1,38 @@
+package networking
+
+import (
+	"net"
+)
+
+const defaultBufSize = 1 * 1024
+
+
+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
+}
+
+func ConnectTCP(address string) (*net.TCPConn, error) {
+	addr, err := net.ResolveTCPAddr("tcp", address)
+	if err != nil {
+		return nil, err
+	}
+	conn, err := net.DialTCP("tcp", nil, addr)
+	if err != nil {
+		return nil, err
+	}
+	conn.SetNoDelay(true)
+	return conn, nil
+}

networking/reader.go

@@ -0,0 +1,57 @@
+package networking
+
+import (
+	"io"
+)
+
+
+type AsyncReader struct {
+	BufSize int
+	stream io.ReadCloser
+	dataChan chan []byte
+	stopFlag bool
+}
+
+// NewListen starts listening for a single connection on the given address.
+// This method blocks until this is done.
+func NewAsyncReader(reader io.ReadCloser) *AsyncReader {
+	return &AsyncReader{
+		BufSize: defaultBufSize,
+		stream: reader,
+		stopFlag: false,
+	}
+}
+
+// RecvChan 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.
+// The data read is owned by the reader, it is thread-safe to modify.
+func (ar *AsyncReader) RecvChan() <-chan []byte {
+	ar.dataChan = make(chan []byte)
+	go ar.recvLoop()
+	return ar.dataChan
+}
+
+func (ar *AsyncReader) recvLoop() {
+	buf := make([]byte, ar.BufSize)
+	for !ar.stopFlag {
+		n, err := ar.stream.Read(buf)
+		if err != nil {
+			ar.stopFlag = true
+		}
+		if n > 0 {
+			// the underlying memory of the buffer
+			// must be copied for thread safety
+			sendBuf := make([]byte, n)
+			copy(sendBuf, buf)
+			ar.dataChan <- sendBuf
+		}
+	}
+	close(ar.dataChan)
+}
+
+func (ar *AsyncReader) Close() {
+	ar.stopFlag = true
+	ar.stream.Close()
+}

networking/writer.go

@@ -0,0 +1,48 @@
+package networking
+
+import (
+	"fmt"
+	"io"
+)
+
+
+type AsyncWriter struct {
+	BufSize int
+	stream io.WriteCloser
+	dataChan chan []byte
+	stopFlag bool
+}
+
+func NewAsyncWriter(writer io.WriteCloser) *AsyncWriter {
+	return &AsyncWriter{
+		BufSize: defaultBufSize,
+		stream: writer,
+		stopFlag: false,
+	}
+}
+
+func (aw *AsyncWriter) SendChan() (chan<- []byte) {
+	aw.dataChan = make(chan []byte)
+	go aw.sendLoop()
+	return aw.dataChan
+}
+
+func (aw *AsyncWriter) sendLoop() {
+	for !aw.stopFlag {
+		data, open := <- aw.dataChan
+		if !open {
+			aw.stopFlag = true
+			return
+		}
+		_, err := aw.stream.Write(data)
+		if err != nil {
+			fmt.Println("ouchie")
+			aw.stopFlag = true
+		}
+	}
+}
+
+func (aw *AsyncWriter) Close() {
+	aw.stopFlag = true
+	close(aw.dataChan)
+}