add more tests

.travis.yml

@@ -1,10 +1,11 @@
 language: go
 
-sudo: false
 go:
   - 1.x
+
 before_install:
   - go get github.com/mattn/goveralls
+
 script:
   - go test ./... -coverprofile=psql-streamer.coverprofile
   - $GOPATH/bin/goveralls -service=travis-ci -coverprofile psql-streamer.coverprofile

sink/kafka/kafka.go

@@ -71,7 +71,6 @@ func New(name string, v *viper.Viper) (s *Kfk, err error) {
 		Balancer:          &kafka.Hash{},
 		Dialer:            &kafka.Dialer{Timeout: v.GetDuration("timeout")},
 		Brokers:           v.GetStringSlice("hosts"),
-		Topic:             v.GetString("topic"),
 		BatchTimeout:      v.GetDuration("batchTimeout"),
 		BatchSize:         v.GetInt("batchSize"),
 		RequiredAcks:      v.GetInt("requiredAcks"),

sink/kafka/kafka_test.go

@@ -1 +1,37 @@
 package kafkasink
+
+import (
+	"github.com/blind-oracle/psql-streamer/event"
+	"github.com/spf13/viper"
+	"github.com/stretchr/testify/assert"
+	"testing"
+	"time"
+)
+
+func Test_Kafka(t *testing.T) {
+	ev := event.Event{
+		Host:     "foo",
+		Database: "bar",
+		Table:    "baz",
+		Action:   "insert",
+		Columns: map[string]interface{}{
+			"a": "B",
+		},
+		Timestamp: time.Unix(123456789, 0),
+	}
+
+	v := viper.New()
+	v.Set("hosts", []string{"127.0.0.1:1234"})
+	v.Set("handlers", []string{"passthrough"})
+	v.Set("topicFallback", "tfb")
+	v.Set("async", true)
+
+	kfk, err := New("test", v)
+	assert.Nil(t, err)
+
+	err = kfk.ProcessEventsBatch([]event.Event{ev})
+	assert.Nil(t, err)
+
+	err = kfk.Close()
+	assert.Nil(t, err)
+}

source/postgres/postgres_test.go

@@ -7,13 +7,13 @@ import (
 	"time"
 
 	"github.com/blind-oracle/psql-streamer/mux"
+	"github.com/stretchr/testify/assert"
 
 	"github.com/blind-oracle/pgoutput"
 	"github.com/blind-oracle/psql-streamer/common"
 	"github.com/blind-oracle/psql-streamer/event"
 	"github.com/blind-oracle/psql-streamer/sink"
 	"github.com/blind-oracle/psql-streamer/sink/stub"
-	"github.com/stretchr/testify/assert"
 )
 
 var (