Add quick & dirty prototype for streamer side

connection/connect.go

@@ -0,0 +1,51 @@
+package connection
+
+import (
+	"net"
+	"fmt"
+)
+
+
+func NewConnect(address string) (*Connection, error) {
+	conn, err := connectTCP(address)
+	if err != nil {
+		return nil, err
+	}
+	return &Connection{
+		conn: conn,
+		BufSize: defaultBufSize,
+		stopFlag: false,
+	}, 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
+	}
+	return conn, nil
+}
+
+func (c *Connection) SendChan() (chan<- []byte) {
+	c.dataChan = make(chan []byte)
+	go c.sendLoop()
+	return c.dataChan
+}
+
+func (c *Connection) sendLoop() {
+	for !c.stopFlag {
+		data, open := <- c.dataChan
+		if !open {
+			c.stopFlag = true
+		}
+		_, err := c.conn.Write(data)
+		if err != nil {
+			fmt.Println(err)
+			c.stopFlag = true
+		}
+	}
+}

connection/connection.go

@@ -0,0 +1,22 @@
+package connection
+
+import (
+	"net"
+)
+
+const defaultBufSize = 1 * 1024
+
+
+// Connection represents a recv-only network connection.
+type Connection struct {
+	BufSize int
+	conn *net.TCPConn
+	dataChan chan []byte
+	stopFlag bool
+}
+
+// Close closes the connection
+func (c *Connection) Close() {
+	c.stopFlag = true
+	c.conn.Close()
+}

connection/listen.go

@@ -2,19 +2,8 @@ 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.
@@ -53,15 +42,15 @@ func listenAndAcceptTCP(address string) (*net.TCPConn, error) {
 // Use this to read bytes from the network connection.
 // The data read is owned by the reader, it is thread-safe to modify.
 func (c *Connection) RecvChan() <-chan []byte {
-	c.recvChan = make(chan []byte)
+	c.dataChan = make(chan []byte)
 	go c.recvLoop()
-	return c.recvChan
+	return c.dataChan
 }
 
 func (c *Connection) recvLoop() {
 	buf := make([]byte, c.BufSize)
 	for !c.stopFlag {
-		n, err := io.ReadFull(c.conn, buf)
+		n, err := c.conn.Read(buf)
 		if err != nil {
 			c.stopFlag = true
 		}
@@ -70,14 +59,8 @@ func (c *Connection) recvLoop() {
 			// must be copied for thread safety
 			sendBuf := make([]byte, n)
 			copy(sendBuf, buf)
-			c.recvChan <- sendBuf
+			c.dataChan <- sendBuf
 		}
 	}
-	close(c.recvChan)
-}
-
-// Close closes the connection
-func (c *Connection) Close() {
-	c.stopFlag = true
-	c.conn.Close()
+	close(c.dataChan)
 }