atridad/goth.stack
Template
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Updated package structure and cleaned up logging

Browse Source
This commit is contained in:
Atridad Lahiji
2024-02-06 23:05:22 -07:00
parent 74e03bda74
commit ec59d59763
5 changed files with 24 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
lib/localpubsub.go → lib/pubsub/adapters/localpubsub.go
View File

@ -1,10 +1,11 @@
package lib
package adapters
import (
"context"
"sync"
"time"
"goth.stack/lib"
"goth.stack/lib/pubsub"
)
@ -28,22 +29,22 @@ func (ps *LocalPubSub) SubscribeToChannel(channel string) (pubsub.PubSubMessage,
ch := make(chan pubsub.Message, 100)
ps.subscribers[channel] = append(ps.subscribers[channel], ch)
LogInfo.Printf("[PUBSUB/LOCAL] Subscribed to channel %s", channel)
lib.LogInfo.Printf("[PUBSUB/LOCAL] Subscribed to channel %s\n", channel)
return &LocalPubSubMessage{messages: ch}, nil
}
func (ps *LocalPubSub) PublishToChannel(channel string, message string) error {
ps.lock.Lock() // Changed from RLock to Lock
defer ps.lock.Unlock() // Changed from RUnlock to Unlock
ps.lock.Lock()
defer ps.lock.Unlock()
if subscribers, ok := ps.subscribers[channel]; ok {
LogInfo.Printf("[PUBSUB/LOCAL] Publishing message to channel %s: %s", channel, message)
lib.LogInfo.Printf("\n[PUBSUB/LOCAL] Publishing message to channel %s: %s\n", channel, message)
for _, ch := range subscribers {
ch <- pubsub.Message{Payload: message}
}
} else {
LogWarning.Printf("[PUBSUB/LOCAL] No subscribers for channel %s", channel)
lib.LogWarning.Printf("\n[PUBSUB/LOCAL] No subscribers for channel %s\n", channel)
}
return nil
@ -57,7 +58,7 @@ func (m *LocalPubSubMessage) ReceiveMessage(ctx context.Context) (*pubsub.Messag
return nil, ctx.Err()
case msg := <-m.messages:
// A message has been received. Send it to the client.
LogInfo.Printf("[PUBSUB/LOCAL] Received message: %s", msg.Payload)
lib.LogInfo.Printf("\n[PUBSUB/LOCAL] Received message: %s\n", msg.Payload)
return &msg, nil
case <-time.After(30 * time.Second):
// No message has been received for 30 seconds. Send a keep-alive message.

12
lib/redispubsub.go → lib/pubsub/adapters/redispubsub.go
View File

@ -1,12 +1,12 @@
package lib
package adapters
import (
"context"
"os"
"github.com/fatih/color"
"github.com/joho/godotenv"
"github.com/redis/go-redis/v9"
"goth.stack/lib"
"goth.stack/lib/pubsub"
)
@ -30,7 +30,7 @@ func NewRedisClient() *redis.Client {
redis_host := os.Getenv("REDIS_HOST")
redis_password := os.Getenv("REDIS_PASSWORD")
LogInfo.Printf("Connecting to Redis at %s", redis_host)
lib.LogInfo.Printf("\n[PUBSUB/REDIS]Connecting to Redis at %s\n", redis_host)
RedisClient = redis.NewClient(&redis.Options{
Addr: redis_host,
Password: redis_password,
@ -45,7 +45,7 @@ func (m *RedisPubSubMessage) ReceiveMessage(ctx context.Context) (*pubsub.Messag
if err != nil {
return nil, err
}
LogInfo.Printf("[PUBSUB/REDIS] Received message: %s", msg.Payload)
lib.LogInfo.Printf("\n[PUBSUB/REDIS] Received message: %s\n", msg.Payload)
return &pubsub.Message{Payload: msg.Payload}, nil
}
@ -55,7 +55,7 @@ func (ps *RedisPubSub) SubscribeToChannel(channel string) (pubsub.PubSubMessage,
if err != nil {
return nil, err
}
LogInfo.Printf("[PUBSUB/REDIS] Subscribed to channel %s", channel)
lib.LogInfo.Printf("\n[PUBSUB/REDIS] Subscribed to channel %s\n", channel)
return &RedisPubSubMessage{pubsub: pubsub}, nil
}
@ -65,6 +65,6 @@ func (r *RedisPubSub) PublishToChannel(channel string, message string) error {
if err != nil {
return err
}
color.Cyan("[PUBSUB/REDIS] Publishing message to channel %s: %s", channel, message)
lib.LogInfo.Printf("\n[PUBSUB/REDIS] Publishing message to channel %s: %s\n", channel, message)
return nil
}