MCorange/wings
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Hopefully fix a possible race condition when Unsubscribing from the EventBus while an event is being Published

Browse Source
This commit is contained in:
Matthew Penner 2020-07-29 23:18:11 -06:00
parent 373dbd355e
commit 5036077152
2 changed files with 23 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
router/websocket/listeners.go
View File

@ -26,9 +26,9 @@ func (h *Handler) ListenForExpiration(ctx context.Context) {
jwt := h.GetJwt()
if jwt != nil {
if jwt.ExpirationTime.Unix()-time.Now().Unix() <= 0 {
h.SendJson(&Message{Event: TokenExpiredEvent})
_ = h.SendJson(&Message{Event: TokenExpiredEvent})
} else if jwt.ExpirationTime.Unix()-time.Now().Unix() <= 180 {
h.SendJson(&Message{Event: TokenExpiringEvent})
_ = h.SendJson(&Message{Event: TokenExpiringEvent})
}
}
}
@ -63,7 +63,7 @@ func (h *Handler) ListenForServerEvents(ctx context.Context) {
close(eventChannel)
default:
h.SendJson(&Message{
_ = h.SendJson(&Message{
Event: d.Topic,
Args: []string{d.Data},
})

38
server/events.go
View File

@ -46,30 +46,32 @@ func (s *Server) Events() *EventBus {
// Publish data to a given topic.
func (e *EventBus) Publish(topic string, data string) {
e.RLock()
defer e.RUnlock()
go func() {
e.RLock()
defer e.RUnlock()
t := topic
// Some of our topics for the socket support passing a more specific namespace,
// such as "backup completed:1234" to indicate which specific backup was completed.
//
// In these cases, we still need to the send the event using the standard listener
// name of "backup completed".
if strings.Contains(topic, ":") {
parts := strings.SplitN(topic, ":", 2)
t := topic
// Some of our topics for the socket support passing a more specific namespace,
// such as "backup completed:1234" to indicate which specific backup was completed.
//
// In these cases, we still need to the send the event using the standard listener
// name of "backup completed".
if strings.Contains(topic, ":") {
parts := strings.SplitN(topic, ":", 2)
if len(parts) == 2 {
t = parts[0]
if len(parts) == 2 {
t = parts[0]
}
}
}
if ch, ok := e.subscribers[t]; ok {
go func(data Event, cs []chan Event) {
for _, channel := range cs {
if ch, ok := e.subscribers[t]; ok {
data := Event{Data: data, Topic: topic}
for _, channel := range ch {
channel <- data
}
}(Event{Data: data, Topic: topic}, ch)
}
}
}()
}
func (e *EventBus) PublishJson(topic string, data interface{}) error {