Fix tests

.travis.yml

@@ -1,3 +1,3 @@
 language: go 
 go: 1.1
-script: sudo -E bash -c "source /etc/profile && gvm use go1.1 && export GOPATH=$HOME/gopath:$GOPATH && go test -v"
+script: sudo -E bash -c "source /etc/profile && gvm use go1.1 && export GOPATH=$HOME/gopath:$GOPATH && go get && go test -v"

Makefile

@@ -1,16 +1,11 @@
-all: build-macosx build-x86 build-x64
+all: build-x86 build-x64
 
 build-x64:
-    GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build
-    tar -czf gor_x64.tar.gz gor
-    rm gor
+	GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build
+	tar -czf gor_x64.tar.gz gor
+	rm gor
 
 build-x86:
-    GOOS=linux GOARCH=386 CGO_ENABLED=0 go build
-    tar -czf gor_x86.tar.gz gor
-    rm gor
-
-build-macosx:
-    GOOS=darwin GOARCH=amd64 CGO_ENABLED=0 go build
-    tar -czf gor_macosx.tar.gz gor
-    rm gor
+	GOOS=linux GOARCH=386 CGO_ENABLED=0 go build
+	tar -czf gor_x86.tar.gz gor
+	rm gor

elasticsearch/elasticsearch.go

@@ -1,4 +1,4 @@
-package gor
+package elasticsearch
 
 import (
 	"encoding/json"

emitter.go

@@ -1,4 +1,4 @@
-package gor
+package main
 
 import (
 	"io"
@@ -22,7 +22,7 @@ func CopyMulty(src io.Reader, writers ...io.Writer) (err error) {
 
 	for {
 		nr, er := src.Read(buf)
-		if nr > 0 {
+		if nr > 0 && len(buf) < nr {
 			Debug("Sending", src, ": ", string(buf[0:nr]))
 
 			if Settings.splitOutput {

emitter_test.go

@@ -1,4 +1,4 @@
-package gor
+package main
 
 import (
 	"io"

gor.go

@@ -10,8 +10,6 @@ import (
 	"runtime/debug"
 	"runtime/pprof"
 	"time"
-
-	"github.com/buger/gor"
 )
 
 var (
@@ -30,12 +28,12 @@ func main() {
 		}
 	}()
 
-	fmt.Println("Version:", gor.VERSION)
+	fmt.Println("Version:", VERSION)
 
 	flag.Parse()
-	gor.InitPlugins()
+	InitPlugins()
 
-	if len(gor.Plugins.Inputs) == 0 || len(gor.Plugins.Outputs) == 0 {
+	if len(Plugins.Inputs) == 0 || len(Plugins.Outputs) == 0 {
 		log.Fatal("Required at least 1 input and 1 output")
 	}
 
@@ -47,7 +45,7 @@ func main() {
 		profileCPU(*cpuprofile)
 	}
 
-	gor.Start(nil)
+	Start(nil)
 }
 
 func profileCPU(cpuprofile string) {

input_dummy.go

@@ -1,4 +1,4 @@
-package gor
+package main
 
 import (
 	"time"

input_file.go

@@ -1,4 +1,4 @@
-package gor
+package main
 
 import (
 	"encoding/gob"

input_raw.go

@@ -1,4 +1,4 @@
-package gor
+package main
 
 import (
 	raw "github.com/buger/gor/raw_socket_listener"

input_raw_test.go

@@ -1,4 +1,4 @@
-package gor
+package main
 
 import (
 	"io"

input_tcp.go

@@ -1,4 +1,4 @@
-package gor
+package main
 
 import (
 	"bufio"

input_tcp_test.go

@@ -1,4 +1,4 @@
-package gor
+package main
 
 import (
 	"io"

limiter.go

@@ -1,4 +1,4 @@
-package gor
+package main
 
 import (
 	"fmt"

limiter_test.go

@@ -1,4 +1,4 @@
-package gor
+package main
 
 import (
 	"io"

output_dummy.go

@@ -1,4 +1,4 @@
-package gor
+package main
 
 import (
 	"fmt"

output_file.go

@@ -1,4 +1,4 @@
-package gor
+package main
 
 import (
 	"encoding/gob"

output_file_test.go

@@ -1,4 +1,4 @@
-package gor
+package main
 
 import (
 	"io"

output_http.go

@@ -1,4 +1,4 @@
-package gor
+package main
 
 import (
 	"bufio"
@@ -115,7 +115,7 @@ func (o *HTTPOutput) sendRequest(client *http.Client, data []byte) {
 	if len(o.methods) > 0 && !o.methods.Contains(request.Method) {
 		return
 	}
-	
+
 	// Change HOST of original request
 	URL := o.address + request.URL.Path + "?" + request.URL.RawQuery
 

output_http_test.go

@@ -1,4 +1,4 @@
-package gor
+package main
 
 import (
 	"io"

output_tcp.go

@@ -1,4 +1,4 @@
-package gor
+package main
 
 import (
 	"fmt"

output_tcp_test.go

@@ -1,4 +1,4 @@
-package gor
+package main
 
 import (
 	"bufio"

plugins.go

@@ -1,4 +1,4 @@
-package gor
+package main
 
 import (
 	"io"

raw_socket_listener/listener.go

@@ -15,7 +15,7 @@ import (
 // Since we can't use default TCP libraries RAWTCPLitener implements own TCP layer
 // TCP packets is parsed using tcp_packet.go, and flow control is managed by tcp_message.go
 type Listener struct {
-	messages map[uint32]*TCPMessage // buffer of TCPMessages waiting to be send
+	messages map[string]*TCPMessage // buffer of TCPMessages waiting to be send
 
 	c_packets  chan *TCPPacket
 	c_messages chan *TCPMessage // Messages ready to be send to client
@@ -33,7 +33,7 @@ func NewListener(addr string, port string) (rawListener *Listener) {
 	rawListener.c_packets = make(chan *TCPPacket, 100)
 	rawListener.c_messages = make(chan *TCPMessage, 100)
 	rawListener.c_del_message = make(chan *TCPMessage, 100)
-	rawListener.messages = make(map[uint32]*TCPMessage)
+	rawListener.messages = make(map[string]*TCPMessage)
 
 	rawListener.addr = addr
 	rawListener.port, _ = strconv.Atoi(port)
@@ -50,7 +50,7 @@ func (t *Listener) listen() {
 		// If message ready for deletion it means that its also complete or expired by timeout
 		case message := <-t.c_del_message:
 			t.c_messages <- message
-			delete(t.messages, message.Ack)
+			delete(t.messages, message.ID)
 
 		// We need to use channels to process each packet to avoid data races
 		case packet := <-t.c_packets:
@@ -71,7 +71,7 @@ func (t *Listener) readRAWSocket() {
 
 	for {
 		// Note: ReadFrom receive messages without IP header
-		n, _, err := conn.ReadFrom(buf)
+		n, addr, err := conn.ReadFrom(buf)
 
 		if err != nil {
 			log.Println("Error:", err)
@@ -79,17 +79,17 @@ func (t *Listener) readRAWSocket() {
 		}
 
 		if n > 0 {
-			t.parsePacket(buf[:n])
+			t.parsePacket(addr, buf[:n])
 		}
 	}
 }
 
-func (t *Listener) parsePacket(buf []byte) {
+func (t *Listener) parsePacket(addr net.Addr, buf []byte) {
 	if t.isIncomingDataPacket(buf) {
 		new_buf := make([]byte, len(buf))
 		copy(new_buf, buf)
 
-		t.c_packets <- ParseTCPPacket(new_buf)
+		t.c_packets <- ParseTCPPacket(addr, new_buf)
 	}
 }
 
@@ -119,15 +119,16 @@ func (t *Listener) isIncomingDataPacket(buf []byte) bool {
 // For TCP message unique id is Acknowledgment number (see tcp_packet.go)
 func (t *Listener) processTCPPacket(packet *TCPPacket) {
 	defer func() { recover() }()
-	
-	var message *TCPMessage
 
-	message, ok := t.messages[packet.Ack]
+	var message *TCPMessage
+	m_id := packet.Addr.String() + strconv.Itoa(int(packet.Ack))
+
+	message, ok := t.messages[m_id]
 
 	if !ok {
 		// We sending c_del_message channel, so message object can communicate with Listener and notify it if message completed
-		message = NewTCPMessage(packet.Ack, t.c_del_message)
-		t.messages[packet.Ack] = message
+		message = NewTCPMessage(m_id, t.c_del_message)
+		t.messages[m_id] = message
 	}
 
 	// Adding packet to message

raw_socket_listener/tcp_message.go

@@ -15,7 +15,7 @@ const MSG_EXPIRE = 200 * time.Millisecond
 // Message can be compiled from unique packets with same message_id which sorted by sequence
 // Message is received if we didn't receive any packets for 200ms
 type TCPMessage struct {
-	Ack     uint32 // Message ID
+	ID      string // Message ID
 	packets []*TCPPacket
 
 	timer *time.Timer // Used for expire check
@@ -26,8 +26,8 @@ type TCPMessage struct {
 }
 
 // NewTCPMessage pointer created from a Acknowledgment number and a channel of messages readuy to be deleted
-func NewTCPMessage(Ack uint32, c_del chan *TCPMessage) (msg *TCPMessage) {
-	msg = &TCPMessage{Ack: Ack}
+func NewTCPMessage(ID string, c_del chan *TCPMessage) (msg *TCPMessage) {
+	msg = &TCPMessage{ID: ID}
 
 	msg.c_packets = make(chan *TCPPacket)
 	msg.c_del_message = c_del // used for notifying that message completed or expired

raw_socket_listener/tcp_packet.go

@@ -2,6 +2,7 @@ package raw_socket
 
 import (
 	"encoding/binary"
+	"net"
 	"strconv"
 	"strings"
 )
@@ -34,11 +35,14 @@ type TCPPacket struct {
 	Urgent     uint16
 
 	Data []byte
+
+	Addr net.Addr
 }
 
-func ParseTCPPacket(b []byte) (p *TCPPacket) {
+func ParseTCPPacket(addr net.Addr, b []byte) (p *TCPPacket) {
 	p = &TCPPacket{Data: b}
 	p.ParseBasic()
+	p.Addr = addr
 
 	return p
 }

settings.go

@@ -1,4 +1,4 @@
-package gor
+package main
 
 import (
 	"flag"

settings_headers.go

@@ -1,4 +1,4 @@
-package gor
+package main
 
 import (
 	"errors"

settings_methods.go

@@ -1,4 +1,4 @@
-package gor
+package main
 
 import (
 	"fmt"

settings_methods_test.go

@@ -1,4 +1,4 @@
-package gor
+package main
 
 import (
 	"testing"

settings_option.go

@@ -1,4 +1,4 @@
-package gor
+package main
 
 import (
 	"fmt"

test_input.go

@@ -1,4 +1,4 @@
-package gor
+package main
 
 type TestInput struct {
 	data chan []byte

test_output.go

@@ -1,4 +1,4 @@
-package gor
+package main
 
 type writeCallback func(data []byte)