Вы не можете выбрать более 25 тем Темы должны начинаться с буквы или цифры, могут содержать дефисы(-) и должны содержать не более 35 символов.
 
 
 
 

296 строки
7.9 KiB

  1. package appcontext
  2. import (
  3. "fmt"
  4. "strings"
  5. "time"
  6. "github.com/AFASystems/presence/internal/pkg/model"
  7. "github.com/segmentio/kafka-go"
  8. )
  9. // AppState provides centralized access to application state
  10. type AppState struct {
  11. beacons model.BeaconsList
  12. httpResults model.HTTPResultList
  13. settings model.Settings
  14. beaconEvents model.BeaconEventList
  15. beaconsLookup map[string]struct{}
  16. latestList model.LatestBeaconsList
  17. kafkaReadersList model.KafkaReadersList
  18. kafkaWritersList model.KafkaWritersList
  19. }
  20. // NewAppState creates a new application context AppState with default values
  21. func NewAppState() *AppState {
  22. return &AppState{
  23. beacons: model.BeaconsList{
  24. Beacons: make(map[string]model.Beacon),
  25. },
  26. httpResults: model.HTTPResultList{
  27. Results: make(map[string]model.HTTPResult),
  28. },
  29. settings: model.Settings{
  30. Settings: model.SettingsVal{
  31. LocationConfidence: 4,
  32. LastSeenThreshold: 15,
  33. BeaconMetricSize: 30,
  34. HASendInterval: 5,
  35. HASendChangesOnly: false,
  36. RSSIEnforceThreshold: false,
  37. RSSIMinThreshold: 100,
  38. },
  39. },
  40. beaconEvents: model.BeaconEventList{
  41. Beacons: make(map[string]model.BeaconEvent),
  42. },
  43. beaconsLookup: make(map[string]struct{}),
  44. latestList: model.LatestBeaconsList{
  45. LatestList: make(map[string]model.Beacon),
  46. },
  47. kafkaReadersList: model.KafkaReadersList{
  48. KafkaReaders: make([]*kafka.Reader, 0),
  49. },
  50. kafkaWritersList: model.KafkaWritersList{
  51. KafkaWriters: make([]*kafka.Writer, 0),
  52. },
  53. }
  54. }
  55. func (m *AppState) AddKafkaWriter(kafkaUrl, topic string) *kafka.Writer {
  56. kafkaWriter := &kafka.Writer{
  57. Addr: kafka.TCP(kafkaUrl),
  58. Topic: topic,
  59. Balancer: &kafka.LeastBytes{},
  60. Async: false,
  61. RequiredAcks: kafka.RequireAll,
  62. BatchSize: 100,
  63. BatchTimeout: 10 * time.Millisecond,
  64. }
  65. m.kafkaWritersList.KafkaWritersLock.Lock()
  66. m.kafkaWritersList.KafkaWriters = append(m.kafkaWritersList.KafkaWriters, kafkaWriter)
  67. m.kafkaWritersList.KafkaWritersLock.Unlock()
  68. return kafkaWriter
  69. }
  70. func (m *AppState) CleanKafkaWriters() {
  71. fmt.Println("shutdown of kafka readers starts")
  72. for _, r := range m.kafkaWritersList.KafkaWriters {
  73. if err := r.Close(); err != nil {
  74. fmt.Printf("Error in closing kafka writer %v", err)
  75. }
  76. }
  77. fmt.Println("Kafka writers graceful shutdown complete")
  78. }
  79. func (m *AppState) AddKafkaReader(kafkaUrl, topic, groupID string) *kafka.Reader {
  80. brokers := strings.Split(kafkaUrl, ",")
  81. kafkaReader := kafka.NewReader(kafka.ReaderConfig{
  82. Brokers: brokers,
  83. GroupID: groupID,
  84. Topic: topic,
  85. MinBytes: 1,
  86. MaxBytes: 10e6,
  87. })
  88. m.kafkaReadersList.KafkaReadersLock.Lock()
  89. m.kafkaReadersList.KafkaReaders = append(m.kafkaReadersList.KafkaReaders, kafkaReader)
  90. m.kafkaReadersList.KafkaReadersLock.Unlock()
  91. return kafkaReader
  92. }
  93. func (m *AppState) CleanKafkaReaders() {
  94. for _, r := range m.kafkaReadersList.KafkaReaders {
  95. if err := r.Close(); err != nil {
  96. fmt.Printf("Error in closing kafka reader %v", err)
  97. }
  98. }
  99. fmt.Println("Kafka readers graceful shutdown complete")
  100. }
  101. // GetBeacons returns thread-safe access to beacons list
  102. func (m *AppState) GetBeacons() *model.BeaconsList {
  103. return &m.beacons
  104. }
  105. // GetSettings returns thread-safe access to settings
  106. func (m *AppState) GetSettings() *model.Settings {
  107. return &m.settings
  108. }
  109. // GetBeaconEvents returns thread-safe access to beacon events
  110. func (m *AppState) GetBeaconEvents() *model.BeaconEventList {
  111. return &m.beaconEvents
  112. }
  113. // GetBeaconsLookup returns thread-safe access to beacon lookup map
  114. func (m *AppState) GetBeaconsLookup() map[string]struct{} {
  115. return m.beaconsLookup
  116. }
  117. // GetLatestList returns thread-safe access to latest beacons list
  118. func (m *AppState) GetLatestList() *model.LatestBeaconsList {
  119. return &m.latestList
  120. }
  121. // AddBeaconToLookup adds a beacon ID to the lookup map
  122. func (m *AppState) AddBeaconToLookup(id string) {
  123. m.beaconsLookup[id] = struct{}{}
  124. }
  125. // RemoveBeaconFromLookup removes a beacon ID from the lookup map
  126. func (m *AppState) RemoveBeaconFromLookup(id string) {
  127. delete(m.beaconsLookup, id)
  128. }
  129. func (m *AppState) RemoveBeacon(id string) {
  130. m.beacons.Lock.Lock()
  131. delete(m.beacons.Beacons, id)
  132. m.beacons.Lock.Unlock()
  133. }
  134. func (m *AppState) RemoveHTTPResult(id string) {
  135. m.httpResults.Lock.Lock()
  136. delete(m.httpResults.Results, id)
  137. m.httpResults.Lock.Unlock()
  138. }
  139. // BeaconExists checks if a beacon exists in the lookup
  140. func (m *AppState) BeaconExists(id string) bool {
  141. _, exists := m.beaconsLookup[id]
  142. return exists
  143. }
  144. // GetBeacon returns a beacon by ID (thread-safe)
  145. func (m *AppState) GetBeacon(id string) (model.Beacon, bool) {
  146. m.beacons.Lock.RLock()
  147. defer m.beacons.Lock.RUnlock()
  148. beacon, exists := m.beacons.Beacons[id]
  149. return beacon, exists
  150. }
  151. // GetHTTPResult returns a beacon from HTTP results by ID (thread-safe)
  152. func (m *AppState) GetHTTPResult(id string) (model.HTTPResult, bool) {
  153. m.httpResults.Lock.RLock()
  154. defer m.httpResults.Lock.RUnlock()
  155. beacon, exists := m.httpResults.Results[id]
  156. return beacon, exists
  157. }
  158. // UpdateHTTPResult updates a beacon in the list (thread-safe)
  159. func (m *AppState) UpdateHTTPResult(id string, beacon model.HTTPResult) {
  160. m.httpResults.Lock.Lock()
  161. defer m.httpResults.Lock.Unlock()
  162. m.httpResults.Results[id] = beacon
  163. }
  164. // UpdateBeacon updates a beacon in the list (thread-safe)
  165. func (m *AppState) UpdateBeacon(id string, beacon model.Beacon) {
  166. m.beacons.Lock.Lock()
  167. defer m.beacons.Lock.Unlock()
  168. m.beacons.Beacons[id] = beacon
  169. }
  170. // GetBeaconEvent returns a beacon event by ID (thread-safe)
  171. func (m *AppState) GetBeaconEvent(id string) (model.BeaconEvent, bool) {
  172. m.beaconEvents.Lock.RLock()
  173. defer m.beaconEvents.Lock.RUnlock()
  174. event, exists := m.beaconEvents.Beacons[id]
  175. return event, exists
  176. }
  177. // UpdateBeaconEvent updates a beacon event in the list (thread-safe)
  178. func (m *AppState) UpdateBeaconEvent(id string, event model.BeaconEvent) {
  179. m.beaconEvents.Lock.Lock()
  180. defer m.beaconEvents.Lock.Unlock()
  181. m.beaconEvents.Beacons[id] = event
  182. }
  183. // GetLatestBeacon returns the latest beacon by ID (thread-safe)
  184. func (m *AppState) GetLatestBeacon(id string) (model.Beacon, bool) {
  185. m.latestList.Lock.RLock()
  186. defer m.latestList.Lock.RUnlock()
  187. beacon, exists := m.latestList.LatestList[id]
  188. return beacon, exists
  189. }
  190. // UpdateLatestBeacon updates the latest beacon in the list (thread-safe)
  191. func (m *AppState) UpdateLatestBeacon(id string, beacon model.Beacon) {
  192. m.latestList.Lock.Lock()
  193. defer m.latestList.Lock.Unlock()
  194. m.latestList.LatestList[id] = beacon
  195. }
  196. // GetAllBeacons returns a copy of all beacons
  197. func (m *AppState) GetAllBeacons() map[string]model.Beacon {
  198. m.beacons.Lock.RLock()
  199. defer m.beacons.Lock.RUnlock()
  200. beacons := make(map[string]model.Beacon)
  201. for id, beacon := range m.beacons.Beacons {
  202. beacons[id] = beacon
  203. }
  204. return beacons
  205. }
  206. // GetAllHttpResults returns a copy of all beacons
  207. func (m *AppState) GetAllHttpResults() map[string]model.HTTPResult {
  208. m.httpResults.Lock.RLock()
  209. defer m.httpResults.Lock.RUnlock()
  210. beacons := make(map[string]model.HTTPResult)
  211. for id, beacon := range m.httpResults.Results {
  212. beacons[id] = beacon
  213. }
  214. return beacons
  215. }
  216. // GetAllLatestBeacons returns a copy of all latest beacons
  217. func (m *AppState) GetAllLatestBeacons() map[string]model.Beacon {
  218. m.latestList.Lock.RLock()
  219. defer m.latestList.Lock.RUnlock()
  220. beacons := make(map[string]model.Beacon)
  221. for id, beacon := range m.latestList.LatestList {
  222. beacons[id] = beacon
  223. }
  224. return beacons
  225. }
  226. // GetBeaconCount returns the number of tracked beacons
  227. func (m *AppState) GetBeaconCount() int {
  228. m.beacons.Lock.RLock()
  229. defer m.beacons.Lock.RUnlock()
  230. return len(m.beacons.Beacons)
  231. }
  232. // GetSettingsValue returns current settings as a value
  233. func (m *AppState) GetSettingsValue() model.SettingsVal {
  234. m.settings.Lock.RLock()
  235. defer m.settings.Lock.RUnlock()
  236. return m.settings.Settings
  237. }
  238. // UpdateSettings updates the system settings (thread-safe)
  239. func (m *AppState) UpdateSettings(newSettings model.SettingsVal) {
  240. m.settings.Lock.Lock()
  241. defer m.settings.Lock.Unlock()
  242. m.settings.Settings = newSettings
  243. }