Added coloured logs and fixed a concurrency bug in the localpubsub adapter

lib/localpubsub.go

@@ -2,53 +2,54 @@ package lib
 
 import (
 	"context"
-	"log"
 	"sync"
 	"time"
+
+	"goth.stack/lib/pubsub"
 )
 
 type LocalPubSub struct {
-	subscribers map[string][]chan Message
+	subscribers map[string][]chan pubsub.Message
 	lock        sync.RWMutex
 }
 
 type LocalPubSubMessage struct {
-	messages <-chan Message
+	messages <-chan pubsub.Message
 }
 
-func (ps *LocalPubSub) SubscribeToChannel(channel string) (PubSubMessage, error) {
+func (ps *LocalPubSub) SubscribeToChannel(channel string) (pubsub.PubSubMessage, error) {
 	ps.lock.Lock()
 	defer ps.lock.Unlock()
 
 	if ps.subscribers == nil {
-		ps.subscribers = make(map[string][]chan Message)
+		ps.subscribers = make(map[string][]chan pubsub.Message)
 	}
 
-	ch := make(chan Message, 100)
+	ch := make(chan pubsub.Message, 100)
 	ps.subscribers[channel] = append(ps.subscribers[channel], ch)
 
-	log.Printf("Subscribed to channel %s", channel)
+	LogInfo.Printf("[PUBSUB/LOCAL] Subscribed to channel %s", channel)
 
 	return &LocalPubSubMessage{messages: ch}, nil
 }
 
 func (ps *LocalPubSub) PublishToChannel(channel string, message string) error {
-	ps.lock.RLock()
-	defer ps.lock.RUnlock()
+	ps.lock.Lock()         // Changed from RLock to Lock
+	defer ps.lock.Unlock() // Changed from RUnlock to Unlock
 
 	if subscribers, ok := ps.subscribers[channel]; ok {
-		log.Printf("Publishing message to channel %s: %s", channel, message)
+		LogInfo.Printf("[PUBSUB/LOCAL] Publishing message to channel %s: %s", channel, message)
 		for _, ch := range subscribers {
-			ch <- Message{Payload: message}
+			ch <- pubsub.Message{Payload: message}
 		}
 	} else {
-		log.Printf("No subscribers for channel %s", channel)
+		LogWarning.Printf("[PUBSUB/LOCAL] No subscribers for channel %s", channel)
 	}
 
 	return nil
 }
 
-func (m *LocalPubSubMessage) ReceiveMessage(ctx context.Context) (*Message, error) {
+func (m *LocalPubSubMessage) ReceiveMessage(ctx context.Context) (*pubsub.Message, error) {
 	for {
 		select {
 		case <-ctx.Done():
@@ -56,16 +57,16 @@ func (m *LocalPubSubMessage) ReceiveMessage(ctx context.Context) (*Message, erro
 			return nil, ctx.Err()
 		case msg := <-m.messages:
 			// A message has been received. Send it to the client.
-			log.Printf("Received message: %s", msg.Payload)
+			LogInfo.Printf("[PUBSUB/LOCAL] Received message: %s", msg.Payload)
 			return &msg, nil
 		case <-time.After(30 * time.Second):
 			// No message has been received for 30 seconds. Send a keep-alive message.
-			return &Message{Payload: "keep-alive"}, nil
+			return &pubsub.Message{Payload: "keep-alive"}, nil
 		}
 	}
 }
 
-func (ps *LocalPubSub) UnsubscribeFromChannel(channel string, ch <-chan Message) {
+func (ps *LocalPubSub) UnsubscribeFromChannel(channel string, ch <-chan pubsub.Message) {
 	ps.lock.Lock()
 	defer ps.lock.Unlock()
 

lib/logging.go

@@ -0,0 +1,27 @@
+package lib
+
+import "github.com/fatih/color"
+
+// Error logging
+var red = color.New(color.FgRed)
+var LogError = red.Add(color.Bold)
+
+// Info logging
+var cyan = color.New(color.FgCyan)
+var LogInfo = cyan.Add(color.Bold)
+
+// Success logging
+var green = color.New(color.FgGreen)
+var LogSuccess = green.Add(color.Bold)
+
+// Warning logging
+var yellow = color.New(color.FgYellow)
+var LogWarning = yellow.Add(color.Bold)
+
+// Debug logging
+var magenta = color.New(color.FgMagenta)
+var LogDebug = magenta.Add(color.Bold)
+
+// Custom logging
+var white = color.New(color.FgWhite)
+var LogCustom = white.Add(color.Bold)

lib/pubsub/interface.go

@@ -1,4 +1,4 @@
-package lib
+package pubsub
 
 import "context"
 

lib/redis.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/joho/godotenv"
 	"github.com/redis/go-redis/v9"
+	"goth.stack/lib/pubsub"
 )
 
 var RedisClient *redis.Client
@@ -39,16 +40,16 @@ func NewRedisClient() *redis.Client {
 	return RedisClient
 }
 
-func (m *RedisPubSubMessage) ReceiveMessage(ctx context.Context) (*Message, error) {
+func (m *RedisPubSubMessage) ReceiveMessage(ctx context.Context) (*pubsub.Message, error) {
 	msg, err := m.pubsub.ReceiveMessage(ctx)
 	if err != nil {
 		return nil, err
 	}
 
-	return &Message{Payload: msg.Payload}, nil
+	return &pubsub.Message{Payload: msg.Payload}, nil
 }
 
-func (ps *RedisPubSub) SubscribeToChannel(channel string) (PubSubMessage, error) {
+func (ps *RedisPubSub) SubscribeToChannel(channel string) (pubsub.PubSubMessage, error) {
 	pubsub := ps.Client.Subscribe(context.Background(), channel)
 	_, err := pubsub.Receive(context.Background())
 	if err != nil {

lib/sse.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/labstack/echo/v4"
+	"goth.stack/lib/pubsub"
 )
 
 type SSEServerType struct {
@@ -58,7 +59,7 @@ func (s *SSEServerType) ClientCount(channel string) int {
 	return len(s.clients[channel])
 }
 
-func SendSSE(ctx context.Context, messageBroker PubSub, channel string, message string) error {
+func SendSSE(ctx context.Context, messageBroker pubsub.PubSub, channel string, message string) error {
 	// Create a channel to receive an error from the goroutine
 	errCh := make(chan error, 1)
 
@@ -102,7 +103,7 @@ func CreateTickerAndKeepAlive(c echo.Context, duration time.Duration) *time.Tick
 	return ticker
 }
 
-func HandleIncomingMessages(c echo.Context, pubsub PubSubMessage, client chan string) {
+func HandleIncomingMessages(c echo.Context, pubsub pubsub.PubSubMessage, client chan string) {
 	for {
 		select {
 		case <-c.Request().Context().Done():