1 files changed, 34 insertions, 4 deletions
diff --git a/src/hub/src/spreadspace.org/sfive/s5srvPipe.go b/src/hub/src/spreadspace.org/sfive/s5srvPipe.go
index 0bce7a6..8084461 100644
--- a/src/hub/src/spreadspace.org/sfive/s5srvPipe.go
+++ b/src/hub/src/spreadspace.org/sfive/s5srvPipe.go
@@ -9,19 +9,19 @@ func (self StatsSinkServer) handleConnection(conn net.Conn) {
 	reader := bufio.NewReader(conn)
 	buffer, err := reader.ReadBytes('\n')
 	if err != nil {
-		s5l.Printf("failed to read from connection: %v\n", err)
+		s5l.Printf("pipe: failed to read from connection: %v\n", err)
 		return
 	}
 	marshaller, err := NewStatefulDecoder(buffer)
 	if err != nil {
-		s5l.Printf("failed initializing decoder with init message: %v\n", err)
+		s5l.Printf("pipe: failed initializing decoder with init message: %v\n", err)
 		return
 	}
 
 	for {
 		buffer, err := reader.ReadBytes('\n')
 		if err != nil {
-			s5l.Printf("failed to read from connection: %v\n", err)
+			s5l.Printf("pipe: failed to read from connection: %v\n", err)
 			return
 		}
 
@@ -29,7 +29,7 @@ func (self StatsSinkServer) handleConnection(conn net.Conn) {
 
 		value, err := marshaller.Decode(buffer)
 		if err != nil {
-			s5l.Printf("failed to decode message: %v\n", err)
+			s5l.Printf("pipe: failed to decode message: %v\n", err)
 			continue
 		}
 
@@ -37,6 +37,25 @@ func (self StatsSinkServer) handleConnection(conn net.Conn) {
 	}
 }
 
+func (self StatsSinkServer) handlePacketConn(pconn net.PacketConn) {
+	decoder := NewPlainDecoder()
+	buffer := make([]byte, 64*1024)
+	for {
+		n, _, err := pconn.ReadFrom(buffer)
+		if err != nil {
+			s5l.Printf("p-pipe: failed read: %v", err)
+			continue
+		}
+		data := buffer[0:n]
+		value, err := decoder.Decode(data)
+		if err == nil {
+			self.appendData <- value
+		} else {
+			s5l.Printf("p-pipe: failed to decode message: %v\n", err)
+		}
+	}
+}
+
 func (self StatsSinkServer) ServePipe(pipePath string) {
 	ln, err := net.Listen("unix", pipePath)
 	if err != nil {
@@ -55,3 +74,14 @@ func (self StatsSinkServer) ServePipe(pipePath string) {
 		go self.handleConnection(conn)
 	}
 }
+
+func (self StatsSinkServer) ServeGramPipe(pipePath string) {
+	pconn, err := net.ListenPacket("unixgram", pipePath)
+	if err != nil {
+		s5l.Printf("p-pipe: failed to listen: %v", err)
+		return
+	}
+	defer pconn.Close()
+
+	self.handlePacketConn(pconn)
+}