session_manager.go 4.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161
  1. package state
  2. import (
  3. "context"
  4. "log/slog"
  5. "sync"
  6. "github.com/mk6i/retro-aim-server/wire"
  7. )
  8. // InMemorySessionManager handles the lifecycle of a user session and provides
  9. // synchronized message relay between sessions in the session pool. An
  10. // InMemorySessionManager is safe for concurrent use by multiple goroutines.
  11. type InMemorySessionManager struct {
  12. store map[string]*Session
  13. mapMutex sync.RWMutex
  14. logger *slog.Logger
  15. }
  16. // NewInMemorySessionManager creates a new instance of InMemorySessionManager.
  17. func NewInMemorySessionManager(logger *slog.Logger) *InMemorySessionManager {
  18. return &InMemorySessionManager{
  19. logger: logger,
  20. store: make(map[string]*Session),
  21. }
  22. }
  23. // RelayToAll relays a message to all sessions in the session pool.
  24. func (s *InMemorySessionManager) RelayToAll(ctx context.Context, msg wire.SNACMessage) {
  25. s.mapMutex.RLock()
  26. defer s.mapMutex.RUnlock()
  27. for _, sess := range s.store {
  28. s.maybeRelayMessage(ctx, msg, sess)
  29. }
  30. }
  31. // RelayToAllExcept relays a message to all session in the pool except for one
  32. // particular session.
  33. func (s *InMemorySessionManager) RelayToAllExcept(ctx context.Context, except *Session, msg wire.SNACMessage) {
  34. s.mapMutex.RLock()
  35. defer s.mapMutex.RUnlock()
  36. for _, sess := range s.store {
  37. if sess == except {
  38. continue
  39. }
  40. s.maybeRelayMessage(ctx, msg, sess)
  41. }
  42. }
  43. // RelayToScreenName relays a message to a session with a matching screen name.
  44. func (s *InMemorySessionManager) RelayToScreenName(ctx context.Context, screenName string, msg wire.SNACMessage) {
  45. sess := s.RetrieveByScreenName(screenName)
  46. if sess == nil {
  47. s.logger.WarnContext(ctx, "can't send notification because user is not online", "recipient", screenName, "message", msg)
  48. return
  49. }
  50. s.maybeRelayMessage(ctx, msg, sess)
  51. }
  52. // RelayToScreenNames relays a message to sessions with matching screenNames.
  53. func (s *InMemorySessionManager) RelayToScreenNames(ctx context.Context, screenNames []string, msg wire.SNACMessage) {
  54. for _, sess := range s.retrieveByScreenNames(screenNames) {
  55. s.maybeRelayMessage(ctx, msg, sess)
  56. }
  57. }
  58. func (s *InMemorySessionManager) maybeRelayMessage(ctx context.Context, msg wire.SNACMessage, sess *Session) {
  59. switch sess.RelayMessage(msg) {
  60. case SessSendClosed:
  61. s.logger.WarnContext(ctx, "can't send notification because the user's session is closed", "recipient", sess.ScreenName(), "message", msg)
  62. case SessQueueFull:
  63. s.logger.WarnContext(ctx, "can't send notification because queue is full", "recipient", sess.ScreenName(), "message", msg)
  64. sess.Close()
  65. }
  66. }
  67. // AddSession adds a new session to the pool. It replaces an existing session
  68. // with a matching screen name, ensuring that each screen name is unique in the
  69. // pool.
  70. func (s *InMemorySessionManager) AddSession(screenName string) *Session {
  71. s.mapMutex.Lock()
  72. defer s.mapMutex.Unlock()
  73. // Only allow one session at a time per screen name. A session may already
  74. // exist because:
  75. // 1) the user is signing on using an already logged-on screen name.
  76. // 2) the session might be orphaned due to an undetected client
  77. // disconnection.
  78. for _, sess := range s.store {
  79. if screenName == sess.ScreenName() {
  80. sess.Close()
  81. delete(s.store, screenName)
  82. break
  83. }
  84. }
  85. sess := NewSession()
  86. sess.SetScreenName(screenName)
  87. s.store[sess.ScreenName()] = sess
  88. return sess
  89. }
  90. // RemoveSession takes a session out of the session pool.
  91. func (s *InMemorySessionManager) RemoveSession(sess *Session) {
  92. s.mapMutex.Lock()
  93. defer s.mapMutex.Unlock()
  94. delete(s.store, sess.ScreenName())
  95. }
  96. // RetrieveSession finds a session with a matching sessionID. Returns nil if
  97. // session is not found.
  98. func (s *InMemorySessionManager) RetrieveSession(sessionID string) *Session {
  99. s.mapMutex.RLock()
  100. defer s.mapMutex.RUnlock()
  101. return s.store[sessionID]
  102. }
  103. // RetrieveByScreenName find a session with a matching screen name. Returns nil
  104. // if session is not found.
  105. func (s *InMemorySessionManager) RetrieveByScreenName(screenName string) *Session {
  106. s.mapMutex.RLock()
  107. defer s.mapMutex.RUnlock()
  108. for _, sess := range s.store {
  109. if screenName == sess.ScreenName() {
  110. return sess
  111. }
  112. }
  113. return nil
  114. }
  115. func (s *InMemorySessionManager) retrieveByScreenNames(screenNames []string) []*Session {
  116. s.mapMutex.RLock()
  117. defer s.mapMutex.RUnlock()
  118. var ret []*Session
  119. for _, sn := range screenNames {
  120. for _, sess := range s.store {
  121. if sn == sess.ScreenName() {
  122. ret = append(ret, sess)
  123. }
  124. }
  125. }
  126. return ret
  127. }
  128. // Empty returns true if the session pool contains 0 sessions.
  129. func (s *InMemorySessionManager) Empty() bool {
  130. s.mapMutex.RLock()
  131. defer s.mapMutex.RUnlock()
  132. return len(s.store) == 0
  133. }
  134. // AllSessions returns all sessions in the session pool.
  135. func (s *InMemorySessionManager) AllSessions() []*Session {
  136. s.mapMutex.RLock()
  137. defer s.mapMutex.RUnlock()
  138. var sessions []*Session
  139. for _, sess := range s.store {
  140. sessions = append(sessions, sess)
  141. }
  142. return sessions
  143. }