tests: add more tests

hub/hub.go

@@ -1,6 +1,7 @@
 package hub
 
 import (
+	"fmt"
 	"log"
 	"net/http"
 
@@ -41,7 +42,7 @@ func NewHub(v *viper.Viper) (*Hub, error) {
 		logger, err = zap.NewProduction()
 	}
 	if err != nil {
-		panic(err)
+		return nil, fmt.Errorf("can't initialize zap logger: %w", err)
 	}
 
 	t, err := NewTransport(v, logger)

hub/log.go

@@ -12,13 +12,14 @@ func (ss stringArray) MarshalLogArray(arr zapcore.ArrayEncoder) error {
 	return nil
 }
 
-type Field = zapcore.Field
+// LogField is an alias of zapcore.Field, it could be replaced by a custom contract when Go will support generics.
+type LogField = zapcore.Field
 
 type Logger interface {
-	Debug(msg string, fields ...Field)
-	Info(msg string, fields ...Field)
-	Warn(msg string, fields ...Field)
-	Error(msg string, fields ...Field)
-	Fatal(msg string, fields ...Field)
+	Debug(msg string, fields ...LogField)
+	Info(msg string, fields ...LogField)
+	Warn(msg string, fields ...LogField)
+	Error(msg string, fields ...LogField)
+	Fatal(msg string, fields ...LogField)
 	Sync() error
 }

hub/log_test.go

@@ -0,0 +1,43 @@
+package hub
+
+import (
+	"bytes"
+	"net/url"
+	"testing"
+
+	"go.uber.org/zap"
+)
+
+// MemorySink implements zap.Sink by writing all messages to a buffer.
+type MemorySink struct {
+	*bytes.Buffer
+}
+
+// Implement Close and Sync as no-ops to satisfy the interface. The Write
+// method is provided by the embedded buffer.
+
+func (s *MemorySink) Close() error { return nil }
+func (s *MemorySink) Sync() error  { return nil }
+
+var sink *MemorySink
+
+func newTestLogger(t *testing.T) (*MemorySink, Logger) {
+	if sink == nil {
+		sink = &MemorySink{new(bytes.Buffer)}
+		if err := zap.RegisterSink("memory", func(*url.URL) (zap.Sink, error) {
+			return sink, nil
+		}); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	conf := zap.NewProductionConfig()
+	conf.OutputPaths = []string{"memory://"}
+
+	logger, err := conf.Build()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	return sink, logger
+}

hub/publish.go

@@ -52,6 +52,7 @@ func (h *Hub) PublishHandler(w http.ResponseWriter, r *http.Request) {
 	u := &Update{
 		Topics:  topics,
 		Private: private,
+		Debug:   h.config.GetBool("debug"),
 		Event: Event{
 			r.PostForm.Get("data"),
 			r.PostForm.Get("id"),

hub/server_test.go

@@ -56,7 +56,7 @@ func TestForwardedHeaders(t *testing.T) {
 	require.Nil(t, err)
 	defer resp2.Body.Close()
 
-	assert.Equal(t, 1, logs.FilterField(zap.String("remote_addr", "192.0.2.1")).Len())
+	assert.True(t, logs.FilterField(zap.String("remote_addr", "192.0.2.1")).Len() == 1)
 
 	h.server.Shutdown(context.Background())
 }

hub/subscribe.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
 )
 
 // SubscribeHandler creates a keep alive connection and sends the events to the subscribers.
@@ -155,7 +156,7 @@ func retrieveLastEventID(r *http.Request) string {
 // Write sends the given string to the client.
 // It returns false if the dispatch timed out.
 // The current write cannot be cancelled because of https://github.com/golang/go/issues/16100
-func (h *Hub) write(w io.Writer, s *Subscriber, data string, d time.Duration) bool {
+func (h *Hub) write(w io.Writer, s zapcore.ObjectMarshaler, data string, d time.Duration) bool {
 	if d == time.Duration(0) {
 		fmt.Fprint(w, data)
 		w.(http.Flusher).Flush()
@@ -204,6 +205,7 @@ func (h *Hub) dispatchSubscriptionUpdate(s *Subscriber, active bool) {
 		u := &Update{
 			Topics:  []string{subscription.ID},
 			Private: true,
+			Debug:   h.config.GetBool("debug"),
 			Event:   Event{Data: string(json)},
 		}
 		h.transport.Dispatch(u)

hub/subscriber_test.go

@@ -36,3 +36,22 @@ func TestDisconnect(t *testing.T) {
 
 	assert.False(t, s.Dispatch(&Update{}, false))
 }
+
+func TestLogSubscriber(t *testing.T) {
+	sink, logger := newTestLogger(t)
+	defer sink.Reset()
+
+	s := NewSubscriber("123", logger, NewTopicSelectorStore())
+	s.RemoteAddr = "127.0.0.1"
+	s.TopicSelectors = []string{"https://example.com/foo"}
+	s.Topics = []string{"https://example.com/bar"}
+
+	f := zap.Object("subscriber", s)
+	logger.Info("test", f)
+
+	log := sink.String()
+	assert.Contains(t, log, `"last_event_id":"123"`)
+	assert.Contains(t, log, `"remote_addr":"127.0.0.1"`)
+	assert.Contains(t, log, `"topic_selectors":["https://example.com/foo"]`)
+	assert.Contains(t, log, `"topics":["https://example.com/bar"]`)
+}

hub/update.go

@@ -14,6 +14,9 @@ type Update struct {
 	// Private updates can only be dispatched to subscribers authorized to receive them.
 	Private bool
 
+	// To print debug informations
+	Debug bool
+
 	// The Server-Sent Event to send.
 	Event
 }
@@ -25,8 +28,9 @@ func (u *Update) MarshalLogObject(enc zapcore.ObjectEncoder) error {
 	enc.AddArray("topics", stringArray(u.Topics))
 	enc.AddBool("private", u.Private)
 
-	// TODO: only in debug mode
-	enc.AddString("data", u.Data)
+	if u.Debug {
+		enc.AddString("data", u.Data)
+	}
 
 	return nil
 }

hub/update_test.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/gofrs/uuid"
 	"github.com/stretchr/testify/assert"
+	"go.uber.org/zap"
 )
 
 func TestAssingUUID(t *testing.T) {
@@ -24,3 +25,26 @@ func TestAssingUUID(t *testing.T) {
 	_, err := uuid.FromString(strings.TrimPrefix(u.ID, "urn:uuid:"))
 	assert.Nil(t, err)
 }
+
+func TestLogUpdate(t *testing.T) {
+	sink, logger := newTestLogger(t)
+	defer sink.Reset()
+
+	u := &Update{
+		Topics:  []string{"https://example.com/foo"},
+		Private: true,
+		Debug:   true,
+		Event:   Event{ID: "a", Retry: 3, Data: "bar", Type: "baz"},
+	}
+
+	f := zap.Object("update", u)
+	logger.Info("test", f)
+
+	log := sink.String()
+	assert.Contains(t, log, `"id":"a"`)
+	assert.Contains(t, log, `"type":"baz"`)
+	assert.Contains(t, log, `"retry":3`)
+	assert.Contains(t, log, `"topics":["https://example.com/foo"]`)
+	assert.Contains(t, log, `"private":true`)
+	assert.Contains(t, log, `"data":"bar"`)
+}