close the internal state when Unsubscribe (#12)

* close the internal state when Unsubscribe

* Added String representation for Fields

hub.go

@@ -74,6 +74,7 @@ func (h *Hub) NonBlockingSubscribe(cap int, topics ...string) Subscription {
 // Unsubscribe remove and close the Subscription.
 func (h *Hub) Unsubscribe(sub Subscription) {
 	h.matcher.Unsubscribe(sub)
+	sub.subscriber.Close()
 }
 
 // Close will unsubscribe all the subscriptions and close them all.

hub_test.go

@@ -102,7 +102,10 @@ func TestHub(t *testing.T) {
 
 func TestNonBlockingSubscriberShouldAlertIfLoseMessages(t *testing.T) {
 	h := New()
-	h.NonBlockingSubscribe(10, "a.*.c")
+	sub := h.NonBlockingSubscribe(10, "a.*.c")
+
+	defer h.Unsubscribe(sub)
+
 	subsAlert := h.Subscribe(1, AlertTopic)
 	// send messages without a working subscriber
 	for i := 0; i < 11; i++ {
@@ -141,6 +144,10 @@ func TestWith(t *testing.T) {
 
 	msg = <-subs.Receiver
 	require.Equal(t, Fields{"msg": 4, "hub": "subH2", "something": 789}, msg.Fields)
+
+	h.Unsubscribe(subs)
+	v, ok := <-subs.Receiver
+	require.Falsef(t, ok, "Unsubscribe should close the internal channel, received fields %v", v)
 }
 
 func newMessageCounter(s Subscription) *messageCounter {

message.go

@@ -1,5 +1,11 @@
 package hub
 
+import (
+	"fmt"
+	"sort"
+	"strings"
+)
+
 type (
 	// Fields is a [key]value storage for Messages values.
 	Fields map[string]interface{}
@@ -17,3 +23,18 @@ type (
 func (m *Message) Topic() string {
 	return m.Name
 }
+
+func (f Fields) String() string {
+	if len(f) == 0 {
+		return "Fields(<empty>)"
+	}
+
+	fields := make([]string, 0, len(f))
+	for k, v := range f {
+		fields = append(fields, fmt.Sprintf("[%s]%v", k, v))
+	}
+
+	sort.Strings(fields)
+
+	return "Fields( " + strings.Join(fields, ", ") + " )"
+}

message_test.go

@@ -0,0 +1,23 @@
+package hub
+
+import "testing"
+
+func TestFields_String(t *testing.T) {
+	tests := []struct {
+		name string
+		f    Fields
+		want string
+	}{
+		{"empty fields", Fields{}, "Fields(<empty>)"},
+		{"one field", Fields{"foo": "bar"}, "Fields( [foo]bar )"},
+		{"two fields", Fields{"foo": "bar", "baz": true}, "Fields( [baz]true, [foo]bar )"},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			if got := tt.f.String(); got != tt.want {
+				t.Errorf("Fields.String() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}