Implements pollUpdates

server/server.go

@@ -20,6 +20,9 @@ type Server struct {
 	conversations []model.Conversation
 	authTokens    []model.AuthToken
 	messageStore  map[string][]model.Message
+	updateItems   map[int]model.UpdateItem
+	updateChannel chan []model.UpdateItem
+	updateItemSeq int
 }
 
 type AuthError struct {
@@ -44,6 +47,9 @@ func NewServer() *Server {
 		conversations: []model.Conversation{},
 		authTokens:    []model.AuthToken{},
 		messageStore:  make(map[string][]model.Message),
+		updateItems:   make(map[int]model.UpdateItem),
+		updateChannel: make(chan []model.UpdateItem),
+		updateItemSeq: 0,
 	}
 }
 
@@ -157,9 +163,52 @@ func (s *Server) ReceiveMessage(conversation *model.Conversation, message model.
 	conversation.Date = message.Date
 	conversation.UnreadCount += 1
 
+	// Enqueue Update
+	s.EnqueueUpdateItem(model.UpdateItem{
+		Conversation: conversation,
+		Message:      &message,
+	})
+
 	log.Info().EmbedObject(message).Msgf("Received message from conversation %s", conversation.Guid)
 }
 
+func (s *Server) EnqueueUpdateItem(item model.UpdateItem) {
+	log.Info().EmbedObject(&item).Msg("Enqueuing update item")
+
+	s.updateItemSeq += 1
+	item.MessageSequenceNumber = s.updateItemSeq
+
+	s.updateItems[s.updateItemSeq] = item
+
+	// Publish to channel
+	select {
+	case s.updateChannel <- []model.UpdateItem{item}:
+	default: // Nobody listening
+	}
+}
+
+func (s *Server) FetchUpdates(since int) []model.UpdateItem {
+	items := []model.UpdateItem{}
+	for i := since; i <= s.updateItemSeq; i++ {
+		if val, ok := s.updateItems[i]; ok {
+			items = append(items, val)
+		}
+	}
+
+	return items
+}
+
+func (s *Server) FetchUpdatesBlocking(since int) []model.UpdateItem {
+	if since < 0 || since >= s.updateItemSeq {
+		// Wait for updates
+		log.Info().Msgf("Waiting for updates since %d", since)
+		items := <-s.updateChannel
+		return items
+	} else {
+		return s.FetchUpdates(since)
+	}
+}
+
 // Private
 
 func (s *Server) registerAuthToken(token *model.AuthToken) {