atridad/goth.stack
Template
1
0
Fork 0
Code Activity

Refactor LocalPubSub PublishToChannel method

Browse source
This commit is contained in:
Atridad Lahiji 2024-02-07 01:14:17 -07:00
parent ec59d59763
commit b04c1eede7
No known key found for this signature in database
1 changed files with 26 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

44
lib/pubsub/adapters/localpubsub.go
View file

@ -35,38 +35,23 @@ func (ps *LocalPubSub) SubscribeToChannel(channel string) (pubsub.PubSubMessage,
} }
func (ps *LocalPubSub) PublishToChannel(channel string, message string) error { func (ps *LocalPubSub) PublishToChannel(channel string, message string) error {
subscribers, ok := ps.subscribers[channel]
if !ok {
lib.LogWarning.Printf("\n[PUBSUB/LOCAL] No subscribers for channel %s\n", channel)
return nil
}
ps.lock.Lock() ps.lock.Lock()
defer ps.lock.Unlock() defer ps.lock.Unlock()
if subscribers, ok := ps.subscribers[channel]; ok {
lib.LogInfo.Printf("\n[PUBSUB/LOCAL] Publishing message to channel %s: %s\n", channel, message) lib.LogInfo.Printf("\n[PUBSUB/LOCAL] Publishing message to channel %s: %s\n", channel, message)
for _, ch := range subscribers { for _, ch := range subscribers {
ch <- pubsub.Message{Payload: message} ch <- pubsub.Message{Payload: message}
} }
} else {
lib.LogWarning.Printf("\n[PUBSUB/LOCAL] No subscribers for channel %s\n", channel)
}
return nil return nil
} }
func (m *LocalPubSubMessage) ReceiveMessage(ctx context.Context) (*pubsub.Message, error) {
for {
select {
case <-ctx.Done():
// The client has disconnected. Stop trying to send messages.
return nil, ctx.Err()
case msg := <-m.messages:
// A message has been received. Send it to the client.
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.
return &pubsub.Message{Payload: "keep-alive"}, nil
}
}
}
func (ps *LocalPubSub) UnsubscribeFromChannel(channel string, ch <-chan pubsub.Message) { func (ps *LocalPubSub) UnsubscribeFromChannel(channel string, ch <-chan pubsub.Message) {
ps.lock.Lock() ps.lock.Lock()
defer ps.lock.Unlock() defer ps.lock.Unlock()
@ -86,3 +71,20 @@ func (ps *LocalPubSub) UnsubscribeFromChannel(channel string, ch <-chan pubsub.M
ps.subscribers[channel] = subscribers ps.subscribers[channel] = subscribers
} }
} }
func (m *LocalPubSubMessage) ReceiveMessage(ctx context.Context) (*pubsub.Message, error) {
for {
select {
case <-ctx.Done():
// The client has disconnected. Stop trying to send messages.
return nil, ctx.Err()
case msg := <-m.messages:
// A message has been received. Send it to the client.
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.
return &pubsub.Message{Payload: "keep-alive"}, nil
}
}
}