transport: move key and namespace into redis_pubsub as it is redis specific.

transport/redis_pubsub/key.go

@@ -0,0 +1,19 @@
+package redis_pubsub
+
+import (
+	"fmt"
+
+	"eosio-ship-trace-reader/transport"
+)
+
+// Key consists of a namespace and a channel.
+// And is encoded to a string in this format: `<namespace>::<channel>`
+
+type Key struct {
+	NS      Namespace
+	Channel transport.ChannelInterface
+}
+
+func (k Key) String() string {
+	return fmt.Sprintf("%s::%s", k.NS, k.Channel)
+}

transport/redis_pubsub/key_test.go

@@ -0,0 +1,35 @@
+package redis_pubsub
+
+import (
+	"testing"
+
+	"eosio-ship-trace-reader/transport"
+)
+
+func TestKey_String(t *testing.T) {
+	type fields struct {
+		NS      Namespace
+		Channel transport.ChannelInterface
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		want   string
+	}{
+		{"Empty", fields{NS: Namespace{}, Channel: transport.Channel{}}, "ship::0000000000000000000000000000000000000000000000000000000000000000::"},
+		{"Transactions", fields{NS: Namespace{ChainID: "id"}, Channel: transport.Channel{"transactions"}}, "ship::id::transactions"},
+		{"Nested", fields{NS: Namespace{ChainID: "id"}, Channel: transport.Channel{"one.two"}}, "ship::id::one.two"},
+		{"Action", fields{NS: Namespace{ChainID: "id"}, Channel: transport.ActionChannel{Contract: "mycontract"}}, "ship::id::actions/contract/mycontract"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			k := Key{
+				NS:      tt.fields.NS,
+				Channel: tt.fields.Channel,
+			}
+			if got := k.String(); got != tt.want {
+				t.Errorf("Key.String() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

transport/redis_pubsub/namespace.go

@@ -0,0 +1,47 @@
+package redis_pubsub
+
+import (
+	"strings"
+
+	"eosio-ship-trace-reader/transport"
+)
+
+const (
+	// Default prefix to use when none is set.
+	defaultPrefix = "ship"
+
+	// We need to have some chain_id, so if no one is specified.
+	// we use a "null" id that is all zeros.
+	nullChain = "0000000000000000000000000000000000000000000000000000000000000000"
+)
+
+// Namespace type.
+//
+// Contains a prefix and chain_id to guard keys against collision.
+// Prefix should be sufficient to not collide with other application using the same redis database.
+// chain_id should be ok to not let multiple reader with different chains to write to the same channels.
+
+type Namespace struct {
+	Prefix  string
+	ChainID string
+}
+
+// Create a new key with this namespace.
+func (ns Namespace) NewKey(ch transport.ChannelInterface) Key {
+	return Key{NS: ns, Channel: ch}
+}
+
+func (ns Namespace) String() string {
+	// No Chain id, set to "nullChain"
+	if len(ns.ChainID) < 1 {
+		ns.ChainID = nullChain
+	}
+
+	// Set default prefix if empty.
+	if len(ns.Prefix) < 1 {
+		ns.Prefix = defaultPrefix
+	}
+
+	// Otherwise. return both.
+	return strings.Join([]string{ns.Prefix, ns.ChainID}, "::")
+}

transport/redis_pubsub/namespace_test.go

@@ -0,0 +1,23 @@
+package redis_pubsub
+
+import "testing"
+
+func TestNamespace_String(t *testing.T) {
+	tests := []struct {
+		name string
+		ns   Namespace
+		want string
+	}{
+		{"Empty", Namespace{}, "ship::0000000000000000000000000000000000000000000000000000000000000000"},
+		{"Prefix Only", Namespace{Prefix: "some.prefix"}, "some.prefix::0000000000000000000000000000000000000000000000000000000000000000"},
+		{"ChainID Only", Namespace{ChainID: "1234"}, "ship::1234"},
+		{"Both", Namespace{Prefix: "my.prefix", ChainID: "1234"}, "my.prefix::1234"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := tt.ns.String(); got != tt.want {
+				t.Errorf("Namespace.String() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

transport/redis_pubsub/redis.go

@@ -11,17 +11,19 @@ import (
 type RedisPubsub struct {
 	pipeline redis.Pipeliner
 	ctx      context.Context
+	ns       Namespace
 }
 
-func New(client *redis.Client) *RedisPubsub {
+func New(client *redis.Client, ns Namespace) *RedisPubsub {
 	return &RedisPubsub{
 		pipeline: client.Pipeline(),
 		ctx:      client.Context(),
+		ns:       ns,
 	}
 }
 
 func (r *RedisPubsub) Publish(channel transport.ChannelInterface, payload []byte) error {
-	return r.pipeline.Publish(r.ctx, channel.String(), payload).Err()
+	return r.pipeline.Publish(r.ctx, r.ns.NewKey(channel).String(), payload).Err()
 }
 
 func (r *RedisPubsub) Flush() error {

transport/redis_pubsub/redis_test.go

@@ -12,11 +12,11 @@ import (
 func TestRedisPubsub(t *testing.T) {
 	client, mock := redismock.NewClientMock()
 
-	pubsub := New(client)
+	pubsub := New(client, Namespace{ChainID: "id"})
 
 	mock.MatchExpectationsInOrder(true)
-	mock.ExpectPublish("test", []byte("some string")).SetVal(0)
-	mock.ExpectPublish("test2", []byte("some other string")).SetVal(0)
+	mock.ExpectPublish("ship::id::test", []byte("some string")).SetVal(0)
+	mock.ExpectPublish("ship::id::test2", []byte("some other string")).SetVal(0)
 
 	assert.NoError(t, pubsub.Publish(transport.Channel{"test"}, []byte("some string")))
 	assert.NoError(t, pubsub.Publish(transport.Channel{"test2"}, []byte("some other string")))