CodeMirror/ntfy
1
0
Fork 0
mirror of https://github.com/binwiederhier/ntfy.git synced 2024-06-27 02:20:56 +12:00
Code Issues Projects Releases Wiki Activity

Make publishing asynchronous

Browse source
This commit is contained in:
Philipp Heckel 2021-11-09 14:48:25 -05:00
parent cc752cf797
commit c0ee174b13
1 changed files with 13 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
server/topic.go
View file

@ -28,6 +28,7 @@ func newTopic(id string, last time.Time) *topic {
} }
} }
// Subscribe subscribes to this topic
func (t *topic) Subscribe(s subscriber) int { func (t *topic) Subscribe(s subscriber) int {
t.mu.Lock() t.mu.Lock()
defer t.mu.Unlock() defer t.mu.Unlock()
@ -37,24 +38,29 @@ func (t *topic) Subscribe(s subscriber) int {
return subscriberID return subscriberID
} }
// Unsubscribe removes the subscription from the list of subscribers
func (t *topic) Unsubscribe(id int) { func (t *topic) Unsubscribe(id int) {
t.mu.Lock() t.mu.Lock()
defer t.mu.Unlock() defer t.mu.Unlock()
delete(t.subscribers, id) delete(t.subscribers, id)
} }
// Publish asynchronously publishes to all subscribers
func (t *topic) Publish(m *message) error { func (t *topic) Publish(m *message) error {
t.mu.Lock() go func() {
defer t.mu.Unlock() t.mu.Lock()
t.last = time.Now() defer t.mu.Unlock()
for _, s := range t.subscribers { t.last = time.Now()
if err := s(m); err != nil { for _, s := range t.subscribers {
log.Printf("error publishing message to subscriber") if err := s(m); err != nil {
log.Printf("error publishing message to subscriber")
}
} }
} }()
return nil return nil
} }
// Subscribers returns the number of subscribers to this topic
func (t *topic) Subscribers() int { func (t *topic) Subscribers() int {
t.mu.Lock() t.mu.Lock()
defer t.mu.Unlock() defer t.mu.Unlock()