diff --git a/output_tcp.go b/output_tcp.go index f03e8ac..5be0d7c 100644 --- a/output_tcp.go +++ b/output_tcp.go @@ -95,16 +95,15 @@ func (o *TCPOutput) worker(bufferIndex int) { } } -func (o *TCPOutput) getBufferIndex(data []byte) int { +func (o *TCPOutput) getBufferIndex(msg *Message) int { if !o.config.Sticky { o.workerIndex++ return int(o.workerIndex) % o.config.Workers } hasher := fnv.New32a() - hasher.Write(payloadMeta(data)[1]) + hasher.Write(payloadID(msg.Meta)) return int(hasher.Sum32()) % o.config.Workers - } // PluginWrite writes message to this plugin @@ -113,7 +112,7 @@ func (o *TCPOutput) PluginWrite(msg *Message) (n int, err error) { return len(msg.Data), nil } - bufferIndex := o.getBufferIndex(msg.Data) + bufferIndex := o.getBufferIndex(msg) o.buf[bufferIndex] <- msg if Settings.OutputTCPStats { diff --git a/output_tcp_test.go b/output_tcp_test.go index 6026e8d..803601a 100644 --- a/output_tcp_test.go +++ b/output_tcp_test.go @@ -125,8 +125,9 @@ func TestBufferDistribution(t *testing.T) { } } -func getTestBytes() []byte { - reqh := payloadHeader(RequestPayload, uuid(), time.Now().UnixNano(), -1) - reqb := append(reqh, []byte("GET / HTTP/1.1\r\nHost: www.w3.org\r\nUser-Agent: Go 1.1 package http\r\nAccept-Encoding: gzip\r\n\r\n")...) - return reqb +func getTestBytes() *Message { + return &Message{ + Meta: payloadHeader(RequestPayload, uuid(), time.Now().UnixNano(), -1), + Data: []byte("GET / HTTP/1.1\r\nHost: www.w3.org\r\nUser-Agent: Go 1.1 package http\r\nAccept-Encoding: gzip\r\n\r\n"), + } }