1. 1
2. 2
3. 3
4. 4
5. 5
6. 6
7. 7
8. 8
9. 9
10. 10
11. 11
12. 12
13. 13
14. 14
15. 15
16. 16
17. 17
18. 18
19. 19
20. 20
21. 21
22. 22
23. 23
24. 24
25. 25
26. 26
27. 27
28. 28
29. 29
30. 30
31. 31
32. 32
33. 33
34. 34
35. 35
36. 36
37. 37
38. 38
39. 39
40. 40
41. 41
42. 42
43. 43
44. 44
45. 45
46. 46
47. 47
48. 48
49. 49
50. 50
51. 51
52. 52
53. 53
54. 54
55. 55
56. 56
57. 57
58. 58
59. 59
60. 60
61. 61
62. 62
63. 63
64. 64
65. 65
66. 66
67. 67
68. 68
69. 69
70. 70
71. 71
72. 72
73. 73
74. 74
75. 75
76. 76
77. 77
78. 78

// Copyright 2020 The Gitea Authors. All rights reserved.
// Use of this source code is governed by a MIT-style
// license that can be found in the LICENSE file.

package eventsource

import "sync"

// Messenger is a per uid message store
type Messenger struct {
	mutex    sync.Mutex
	uid      int64
	channels []chan *Event
}

// NewMessenger creates a messenger for a particular uid
func NewMessenger(uid int64) *Messenger {
	return &Messenger{
		uid:      uid,
		channels: [](chan *Event){},
	}
}

// Register returns a new chan []byte
func (m *Messenger) Register() <-chan *Event {
	m.mutex.Lock()
	// TODO: Limit the number of messengers per uid
	channel := make(chan *Event, 1)
	m.channels = append(m.channels, channel)
	m.mutex.Unlock()
	return channel
}

// Unregister removes the provider chan []byte
func (m *Messenger) Unregister(channel <-chan *Event) bool {
	m.mutex.Lock()
	defer m.mutex.Unlock()
	for i, toRemove := range m.channels {
		if channel == toRemove {
			m.channels = append(m.channels[:i], m.channels[i+1:]...)
			close(toRemove)
			break
		}
	}
	return len(m.channels) == 0
}

// UnregisterAll removes all chan []byte
func (m *Messenger) UnregisterAll() {
	m.mutex.Lock()
	defer m.mutex.Unlock()
	for _, channel := range m.channels {
		close(channel)
	}
	m.channels = nil
}

// SendMessage sends the message to all registered channels
func (m *Messenger) SendMessage(message *Event) {
	m.mutex.Lock()
	defer m.mutex.Unlock()
	for i := range m.channels {
		channel := m.channels[i]
		select {
		case channel <- message:
		default:
		}
	}
}

// SendMessageBlocking sends the message to all registered channels and ensures it gets sent
func (m *Messenger) SendMessageBlocking(message *Event) {
	m.mutex.Lock()
	defer m.mutex.Unlock()
	for i := range m.channels {
		m.channels[i] <- message
	}
}