diff options
Diffstat (limited to 'src/hub/src/spreadspace.org/sfive/s5srvPipe.go')
-rw-r--r-- | src/hub/src/spreadspace.org/sfive/s5srvPipe.go | 38 |
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) +} |