input/nats: fix testing

Signed-off-by: tsaikd <[email protected]>

Author: tsaikd

input/nats/README.md

@@ -1,4 +1,4 @@
-gogstash input nats 
+gogstash input nats
 ====================
 
 ## Synopsis
@@ -8,8 +8,8 @@ input:
   # type Must be "nats"
   - type: "nats"
 
-    # nats server host:port, default: "localhost:4222"
-    host: "localhost:4222"
+    # nats server host:port, default: "nats://localhost:4222"
+    host: "nats://localhost:4222"
 
     # creditials for nats, default: ""
     creds: ""

input/nats/inputnats.go

@@ -35,7 +35,7 @@ func DefaultInputConfig() InputConfig {
 				Type: ModuleName,
 			},
 		},
-		Host:  "localhost:4222",
+		Host:  "nats://localhost:4222",
 		Creds: "",
 	}
 }

input/nats/inputnats_test.go

@@ -2,7 +2,6 @@ package inputnats
 
 import (
 	"context"
-	"os"
 	"strings"
 	"testing"
 	"time"
@@ -23,66 +22,52 @@ func init() {
 	config.RegistCodecHandler(codecjson.ModuleName, codecjson.InitHandler)
 }
 
-func TestMain(m *testing.M) {
-	opts := server.Options{
-		Host:     "127.0.0.1",
-		Port:     4222,
-		HTTPPort: -1,
-		Cluster:  server.ClusterOpts{Port: -1},
-		NoLog:    true,
-		NoSigs:   true,
-		Debug:    true,
-		Trace:    true,
-	}
-
-	s, err := server.NewServer(&opts)
-	if err != nil {
-		panic(err)
-	}
-
-	go s.Start()
-	defer s.Shutdown()
-
-	ret := m.Run()
-
-	os.Exit(ret)
-}
-
 func TestInputNats(t *testing.T) {
 	assert := assert.New(t)
 	assert.NotNil(assert)
 	require := require.New(t)
 	require.NotNil(require)
 
+	s, err := server.NewServer(&server.Options{
+		Host:                  "127.0.0.1",
+		Trace:                 true,
+		Debug:                 true,
+		DisableShortFirstPing: true,
+		NoLog:                 true,
+		NoSigs:                true,
+	})
+	require.NoError(err)
+	go s.Start()
+	defer s.Shutdown()
+	time.Sleep(500 * time.Millisecond)
+
 	ctx := context.Background()
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	conf, err := config.LoadFromYAML([]byte(strings.TrimSpace(`
 debugch: true
 input:
   - type:  "nats"
-    host:  "127.0.0.1:4222"
+    host:  "` + s.ClientURL() + `"
     topic: "test.*"
 	`)))
 	require.NoError(err)
 	require.NoError(conf.Start(ctx))
 
-	time.Sleep(500 * time.Millisecond)
-
 	// start a publisher client
 	opts := []nats.Option{nats.Name("test")}
 
 	nc, err := nats.Connect(nats.DefaultURL, opts...)
 	require.NoError(err)
 
-	err = nc.Publish("test.1", []byte("{\"foo\":\"bar\"}"))
+	err = nc.Publish("test.1", []byte(`{"foo":"bar"}`))
 	require.NoError(err)
 
 	err = nc.Flush()
 	require.NoError(err)
 
 	// check event
 	if event, err := conf.TestGetOutputEvent(100 * time.Millisecond); assert.NoError(err) {
-		assert.Equal(map[string]interface{}{"foo": "bar"}, event.Extra)
+		require.EqualValues(map[string]interface{}{"foo": "bar"}, event.Extra)
 	}
 }