simple socket encoding, auto reconnect, integration with pubsub

Author: nidorx

examples/pubsub/main.go

@@ -2,15 +2,16 @@ package main
 
 import (
 	"fmt"
+	"time"
+
 	"github.com/nidorx/chain"
 	"github.com/nidorx/chain/pubsub"
-	"time"
 )
 
 type MyDispatcher struct {
 }
 
-func (d *MyDispatcher) Dispatch(topic string, message any, from string) {
+func (d *MyDispatcher) Dispatch(topic string, message []byte, from string) {
 	println(fmt.Sprintf("New Message. Topic: %s, Content: %s", topic, message))
 }
 

pubsub/compression.go

@@ -23,7 +23,7 @@ func compressPayload(payload []byte) ([]byte, error) {
 
 	// Create a compressed message
 	buf := bytes.NewBuffer(nil)
-	buf.WriteByte(byte(messageTypeCompress))
+	buf.WriteByte(byte(MessageTypeCompress))
 	buf.Write(buffer.Bytes())
 	return buf.Bytes(), nil
 }

pubsub/crypto.go

@@ -2,13 +2,14 @@ package pubsub
 
 import (
 	"bytes"
+
 	"github.com/nidorx/chain"
 	"github.com/nidorx/chain/crypto"
 )
 
 var globalKeyring = chain.NewKeyring("chain.pubsub.keyring.salt", 1000, 32, "sha256")
 
-var aad = append([]byte{byte(messageTypeEncrypt)}, []byte("chain.pubsub.aad")...)
+var aad = append([]byte{byte(MessageTypeEncrypt)}, []byte("chain.pubsub.aad")...)
 
 // encryptPayload is used to encrypt a message before sending
 func encryptPayload(keyring *crypto.Keyring, payload []byte) ([]byte, error) {
@@ -19,7 +20,7 @@ func encryptPayload(keyring *crypto.Keyring, payload []byte) ([]byte, error) {
 
 	// return encrypted cipher text
 	buf := bytes.NewBuffer(nil)
-	buf.WriteByte(byte(messageTypeEncrypt))
+	buf.WriteByte(byte(MessageTypeEncrypt))
 	buf.Write(encrypted)
 	return buf.Bytes(), nil
 }

pubsub/message.go

@@ -1,22 +1,22 @@
 package pubsub
 
-// messageType is an integer ID of a type of message that can be received on network channels from other members.
-type messageType uint8
+// MessageType is an integer ID of a type of message that can be received on network channels from other members.
+type MessageType uint8
 
 // The list of available message types.
 const (
-	messageTypeCompress messageType = iota
-	messageTypeEncrypt
-	messageTypeBroadcast
-	messageTypeDirectBroadcast
-	indirectPingMsg
-	ackRespMsg
-	suspectMsg
-	aliveMsg
-	deadMsg
-	pushPullMsg
-	compoundMsg
-	userMsg
-	nackRespMsg
-	errMsg
+	MessageTypeCompress MessageType = iota
+	MessageTypeEncrypt
+	MessageTypeBroadcast
+	MessageTypeDirectBroadcast
+	IndirectPingMsg
+	AckRespMsg
+	SuspectMsg
+	AliveMsg
+	DeadMsg
+	PushPullMsg
+	CompoundMsg
+	UserMsg
+	NackRespMsg
+	ErrMsg
 )

pubsub/pubsub.go

@@ -22,21 +22,21 @@ var (
 )
 
 type Dispatcher interface {
-	Dispatch(topic string, message any, from string)
+	Dispatch(topic string, message []byte, from string)
 }
 
 type DispatcherFuncImpl struct {
-	Dispatcher func(topic string, message any, from string)
+	Dispatcher func(topic string, message []byte, from string)
 }
 
-func (d *DispatcherFuncImpl) Dispatch(topic string, message any, from string) {
+func (d *DispatcherFuncImpl) Dispatch(topic string, message []byte, from string) {
 	if d.Dispatcher == nil {
 		return
 	}
 	d.Dispatcher(topic, message, from)
 }
 
-func DispatcherFunc(d func(topic string, message any, from string)) Dispatcher {
+func DispatcherFunc(d func(topic string, message []byte, from string)) Dispatcher {
 	return &DispatcherFuncImpl{Dispatcher: d}
 }
 
@@ -122,7 +122,7 @@ func Broadcast(topic string, message []byte, options ...*Option) (err error) {
 	msgToSend := message
 
 	// [messageType: byte] [from: 20 bytes] [msgToSend: ...]
-	msgToSend = append(append([]byte{byte(messageTypeBroadcast)}, selfIdBytes...), msgToSend...)
+	msgToSend = append(append([]byte{byte(MessageTypeBroadcast)}, selfIdBytes...), msgToSend...)
 
 	// Check if we have compression enabled
 	if config.DisableCompression == false {
@@ -178,11 +178,11 @@ func DirectBroadcast(nodeId string, topic string, message []byte, options ...*Op
 	buf.WriteString(topic)
 	buf.Write(message)
 
-	return broadcastMessage(messageTypeDirectBroadcast, "direct:"+nodeId, buf.Bytes(), options...)
+	return broadcastMessage(MessageTypeDirectBroadcast, "direct:"+nodeId, buf.Bytes(), options...)
 }
 
 // Broadcast broadcasts message on given topic across the whole cluster.
-func broadcastMessage(msgType messageType, topic string, message []byte, options ...*Option) (err error) {
+func broadcastMessage(msgType MessageType, topic string, message []byte, options ...*Option) (err error) {
 	var config *AdapterConfig
 	if config = GetAdapter(topic); config == nil {
 		return ErrNoAdapter
@@ -210,7 +210,7 @@ func broadcastMessage(msgType messageType, topic string, message []byte, options
 	msgToSend := buf.Bytes()
 
 	// Check if we have compression enabled
-	if config.DisableCompression == false {
+	if !config.DisableCompression {
 		var compressed []byte
 		if compressed, err = compressPayload(msgToSend); err != nil {
 			slog.Warn(
@@ -224,7 +224,7 @@ func broadcastMessage(msgType messageType, topic string, message []byte, options
 	}
 
 	// Check if we have encryption enabled
-	if config.DisableEncryption == false {
+	if !config.DisableEncryption {
 		keyring := config.Keyring
 		if keyring == nil {
 			keyring = globalKeyring
@@ -245,10 +245,10 @@ func broadcastMessage(msgType messageType, topic string, message []byte, options
 func Dispatch(topic string, message []byte) {
 	if config := GetAdapter(topic); config != nil {
 		// Read the message type
-		msgType := messageType(message[0])
+		msgType := MessageType(message[0])
 
 		// Check if the message is encrypted
-		if msgType == messageTypeEncrypt {
+		if msgType == MessageTypeEncrypt {
 			if config.DisableEncryption {
 				slog.Error(
 					"[chain.pubsub] remote message is encrypted and encryption is not configured",
@@ -274,9 +274,9 @@ func Dispatch(topic string, message []byte) {
 			}
 
 			// Reset message type and buf
-			msgType = messageType(plain[0])
+			msgType = MessageType(plain[0])
 			message = plain
-		} else if config.DisableEncryption == false {
+		} else if !config.DisableEncryption {
 			slog.Error(
 				"[chain.pubsub] encryption is configured but remote message is not encrypted",
 				slog.String("Topic", topic),
@@ -286,7 +286,7 @@ func Dispatch(topic string, message []byte) {
 		}
 
 		// Check if we have a compressed message
-		if msgType == messageTypeCompress {
+		if msgType == MessageTypeCompress {
 			decompressed, err := decompressPayload(message)
 			if err != nil {
 				slog.Error(
@@ -299,7 +299,7 @@ func Dispatch(topic string, message []byte) {
 			}
 
 			// Reset message type and buf
-			msgType = messageType(decompressed[0])
+			msgType = MessageType(decompressed[0])
 			message = decompressed
 		}
 
@@ -334,7 +334,7 @@ func Dispatch(topic string, message []byte) {
 		message = message[20:]
 
 		// Check if is a direct broadcast
-		if msgType == messageTypeDirectBroadcast {
+		if msgType == MessageTypeDirectBroadcast {
 			if topic != directTopic {
 				slog.Error(
 					"[chain.pubsub] invalid topic for remote direct broadcast message",
@@ -379,7 +379,7 @@ func Dispatch(topic string, message []byte) {
 			}
 			topic = string(message[:topicNameLen])
 			message = message[topicNameLen:]
-		} else if msgType != messageTypeBroadcast {
+		} else if msgType != MessageTypeBroadcast {
 			slog.Error(
 				"[chain.pubsub] invalid remote message type",
 				slog.String("Topic", topic),
@@ -396,7 +396,7 @@ func Dispatch(topic string, message []byte) {
 //
 // `topic` - The topic to broadcast to, ie: `"users:123"`
 // `message` - The payload of the broadcast
-func LocalBroadcast(topic string, message any) {
+func LocalBroadcast(topic string, message []byte) {
 	dispatchMessage(topic, message, selfIdString)
 }
 
@@ -480,7 +480,7 @@ func scheduleUnsubscribe(topic string) {
 }
 
 // dispatchMessage deliver the message locally
-func dispatchMessage(topic string, message any, from string) {
+func dispatchMessage(topic string, message []byte, from string) {
 	go func() {
 		if from == "" {
 			from = selfIdString

pubsub/pubsub_test.go

@@ -1,12 +1,13 @@
 package pubsub
 
 import (
-	"github.com/nidorx/chain"
-	"github.com/segmentio/ksuid"
 	"reflect"
 	"sync"
 	"testing"
 	"time"
+
+	"github.com/nidorx/chain"
+	"github.com/segmentio/ksuid"
 )
 
 var (
@@ -197,7 +198,7 @@ type testDispatcherStruct struct {
 	mutex    sync.Mutex
 }
 
-func (d *testDispatcherStruct) Dispatch(topic string, message any, from string) {
+func (d *testDispatcherStruct) Dispatch(topic string, message []byte, from string) {
 	d.mutex.Lock()
 	defer d.mutex.Unlock()
 	d.messages = append(d.messages, &testDispatcherMessage{topic, message, from})

socket/channel.go

@@ -1,6 +1,7 @@
 package socket
 
 import (
+	"encoding/json"
 	"fmt"
 	"log/slog"
 	"sync"
@@ -144,11 +145,11 @@ func (c *Channel) Leave(topic string, handler LeaveHandler) {
 
 // Broadcast on the pubsub server with the given topic, event and payload.
 func (c *Channel) Broadcast(topic string, event string, payload any) (err error) {
-	broadcast := newMessage(MessageTypeBroadcast, topic, event, payload)
-	defer deleteMessage(broadcast)
+	message := newMessage(MessageTypeBroadcast, topic, event, payload)
+	defer deleteMessage(message)
 
 	var bytes []byte
-	if bytes, err = c.serializer.Encode(broadcast); err != nil {
+	if bytes, err = c.serializer.Encode(message); err != nil {
 		return
 	}
 	err = pubsub.Broadcast(topic, bytes)
@@ -157,36 +158,59 @@ func (c *Channel) Broadcast(topic string, event string, payload any) (err error)
 
 // LocalBroadcast on the pubsub server with the given topic, event and payload.
 func (c *Channel) LocalBroadcast(topic string, event string, payload any) (err error) {
-	broadcast := newMessage(MessageTypeBroadcast, topic, event, payload)
-	pubsub.LocalBroadcast(topic, broadcast)
+	message := newMessage(MessageTypeBroadcast, topic, event, payload)
+	var bytes []byte
+	if bytes, err = c.serializer.Encode(message); err != nil {
+		return
+	}
+	pubsub.LocalBroadcast(topic, bytes)
 	return
 }
 
-// Dispatch Hook invoked by pubsub dispatch.
-func (c *Channel) Dispatch(topic string, msg any, from string) {
-	var message *Message
-	var valid bool
-	var payload []byte
-	isByteArray := false
-
-	if payload, valid = msg.([]byte); valid {
-		isByteArray = true
-		message = newMessageAny()
-		if _, err := c.serializer.Decode(payload, message); err != nil {
-			slog.Debug(
-				"[chain.socket] could not decode serialized data",
-				slog.Any("Error", err),
-				slog.Any("Payload", payload),
-				slog.String("Topic", topic),
+// Subscribe ma
+func (c *Channel) Subscribe(pubsubTopic, channelTopic, channelEvent string) {
+	pubsub.Subscribe(pubsubTopic, pubsub.DispatcherFunc(func(topic string, pubsubPayload []byte, from string) {
+		payload := map[string]any{}
+		if err := json.Unmarshal(pubsubPayload, &payload); err != nil {
+			slog.Warn(
+				"[chain.socket] failed to decode pubsub message",
+				slog.Any("error", err),
+				slog.String("from", from),
+				slog.String("event", channelEvent),
+				slog.String("pubsubTopic", pubsubTopic),
+				slog.String("topic", topic),
+				slog.String("message", string(pubsubPayload)),
 			)
-
-			deleteMessage(message)
 			return
 		}
-	} else if message, valid = msg.(*Message); !valid {
+
+		message := newMessage(MessageTypeBroadcast, channelTopic, channelEvent, payload)
+		c.dispatch(channelTopic, message, from)
+	}))
+}
+
+// Dispatch Hook invoked by pubsub dispatch.
+func (c *Channel) Dispatch(topic string, channelMessageEncoded []byte, from string) {
+	var message = newMessageAny()
+
+	if _, err := c.serializer.Decode(channelMessageEncoded, message); err != nil {
+		slog.Debug(
+			"[chain.socket] could not decode serialized data",
+			slog.Any("Error", err),
+			slog.String("topic", topic),
+			slog.String("from", from),
+		)
+		deleteMessage(message)
 		return
 	}
 
+	c.dispatch(topic, message, from)
+}
+
+func (c *Channel) dispatch(topic string, message *Message, from string) {
+
+	defer deleteMessage(message)
+
 	// get sockets
 	c.socketsMutex.RLock()
 	var sockets []*Socket
@@ -216,15 +240,13 @@ func (c *Channel) Dispatch(topic string, msg any, from string) {
 
 	// fastlane (not intercepted, single encode for all sockets)
 
-	if !isByteArray {
-		var err error
-		if payload, err = c.serializer.Encode(message); err != nil {
-			return
-		}
+	encoded, err := c.serializer.Encode(message)
+	if err != nil {
+		return
 	}
 
 	for _, socket := range sockets {
-		socket.Send(payload)
+		socket.Send(encoded)
 	}
 }