Fix bug in datagram message due to shared buffer

server.go

@@ -39,11 +39,16 @@ type Server struct {
 	lastError               error
 	readTimeoutMilliseconds int64
 	tlsPeerNameFunc         TlsPeerNameFunc
+	datagramPool            sync.Pool
 }
 
 //NewServer returns a new Server
 func NewServer() *Server {
-	return &Server{tlsPeerNameFunc: defaultTlsPeerName}
+	return &Server{tlsPeerNameFunc: defaultTlsPeerName, datagramPool: sync.Pool{
+		New: func() interface{} {
+			return make([]byte, 65536)
+		},
+	}}
 }
 
 //Sets the syslog format (RFC3164 or RFC5424 or RFC6587)
@@ -319,8 +324,8 @@ func (s *Server) goReceiveDatagrams(packetconn net.PacketConn) {
 	s.wait.Add(1)
 	go func() {
 		defer s.wait.Done()
-		buf := make([]byte, 65536)
 		for {
+			buf := s.datagramPool.Get().([]byte)
 			n, addr, err := packetconn.ReadFrom(buf)
 			if err == nil {
 				// Ignore trailing control characters and NULs
@@ -366,6 +371,7 @@ func (s *Server) goParseDatagrams() {
 				} else {
 					s.parser(msg.message, msg.client, "")
 				}
+				s.datagramPool.Put(msg.message[:cap(msg.message)])
 			}
 		}
 	}()

server_test.go

@@ -271,3 +271,58 @@ func (s *ServerSuite) TestUDPAutomatic5424Plus6587OctetCount(c *C) {
 	c.Check(handler.LastMessageLength, Equals, int64(len(exampleRFC5424Syslog)))
 	c.Check(handler.LastError, IsNil)
 }
+
+type handlerSlow struct {
+	*handlerCounter
+	contents []string
+}
+
+func (s *handlerSlow) Handle(logParts format.LogParts, msgLen int64, err error) {
+	if len(s.contents) == 0 {
+		time.Sleep(time.Second)
+	}
+	s.contents = append(s.contents, logParts["content"].(string))
+	s.handlerCounter.Handle(logParts, msgLen, err)
+}
+
+func (s *ServerSuite) TestUDPRace(c *C) {
+	handler := &handlerSlow{handlerCounter: &handlerCounter{expected: 3, done: make(chan struct{})}}
+	server := NewServer()
+	server.SetFormat(Automatic)
+	server.SetHandler(handler)
+	server.SetTimeout(10)
+	server.ListenUDP("127.0.0.1:0")
+	server.Boot()
+	conn, err := net.Dial("udp", server.connections[0].LocalAddr().String())
+	c.Assert(err, IsNil)
+	_, err = conn.Write([]byte(exampleSyslog + "1"))
+	c.Assert(err, IsNil)
+	_, err = conn.Write([]byte(exampleSyslog + "2"))
+	c.Assert(err, IsNil)
+	_, err = conn.Write([]byte(exampleSyslog + "3"))
+	c.Assert(err, IsNil)
+	conn.Close()
+	<-handler.done
+	c.Check(handler.contents, DeepEquals, []string{"content1", "content2", "content3"})
+}
+
+func (s *ServerSuite) TestTCPRace(c *C) {
+	handler := &handlerSlow{handlerCounter: &handlerCounter{expected: 3, done: make(chan struct{})}}
+	server := NewServer()
+	server.SetFormat(Automatic)
+	server.SetHandler(handler)
+	server.SetTimeout(10)
+	server.ListenTCP("127.0.0.1:0")
+	server.Boot()
+	conn, err := net.Dial("tcp", server.listeners[0].Addr().String())
+	c.Assert(err, IsNil)
+	_, err = conn.Write([]byte(exampleSyslog + "1\n"))
+	c.Assert(err, IsNil)
+	_, err = conn.Write([]byte(exampleSyslog + "2\n"))
+	c.Assert(err, IsNil)
+	_, err = conn.Write([]byte(exampleSyslog + "3\n"))
+	c.Assert(err, IsNil)
+	conn.Close()
+	<-handler.done
+	c.Check(handler.contents, DeepEquals, []string{"content1", "content2", "content3"})
+}