diff --git a/main.go b/main.go index d72eb10..082dc6e 100644 --- a/main.go +++ b/main.go @@ -155,13 +155,19 @@ func (s *Server) handleConnections(w http.ResponseWriter, r *http.Request) { } } +func broadcastSingleRecord(s *Server, message Message) { + for client := range s.clients { + sendMessageToClient(client, message) + } +} + func sendMessageToClient(client *Client, message Message) { var msgJSON, err = json.Marshal(message) if err != nil { log.Println("Error marshalling json:", err) return } - sendToClient(client, msgJSON) + go sendToClient(client, msgJSON) } func sendToClient(client *Client, msgJSON []byte) { @@ -172,36 +178,6 @@ func sendToClient(client *Client, msgJSON []byte) { } } -func broadcastAllRecords(s *Server) { - allRecords, err := fetchAllRecords() - if err != nil { - log.Println("Error fetching all records:", err) - } - if len(allRecords) > 0 { - for _, msgContent := range allRecords { - fmt.Printf("Broadcasting %s to %d clients\n", msgContent, len(s.clients)) - for client := range s.clients { - sendToClient(client, []byte(msgContent)) - } - } - } -} - -func broadcastRemovedRecords(s *Server, message Message) { - fmt.Printf("Broadcasting removal: %s,%s,%s,%s to %d clients\n", message.Id, message.Status, message.Name, message.Timestamp, len(s.clients)) - for client := range s.clients { - sendMessageToClient(client, message) - } -} - -func broadcastExpiredRecords(s *Server, removed string) { - var message = Message{Id: removed, Name: "", Image: "", Status: "removed", Timestamp: ""} - fmt.Printf("Broadcasting expiration: %s to %d clients\n", message, len(s.clients)) - for client := range s.clients { - sendMessageToClient(client, message) - } -} - func sendAllRecordsToClient(client *Client) { allRecords, err := fetchAllRecords() if err != nil { @@ -209,15 +185,13 @@ func sendAllRecordsToClient(client *Client) { } fmt.Printf("Broadcasting %d records to client\n", len(allRecords)) var message Message - client.mu.Lock() - defer client.mu.Unlock() for _, msgContent := range allRecords { err = json.Unmarshal([]byte(msgContent), &message) if err != nil { log.Println("Unable to marshal JSON due to: ", err) } fmt.Printf("Broadcasting %s,%s,%s,%s\n", message.Id, message.Status, message.Name, message.Timestamp) - sendToClient(client, []byte(msgContent)) + go sendToClient(client, []byte(msgContent)) } } @@ -226,7 +200,7 @@ func (s *Server) listenForExpirationEvents() { defer func(ps *redis.PubSub) { err := ps.Close() if err != nil { - + log.Println("Error closing redis pubsub:", err) } }(ps) @@ -235,8 +209,9 @@ func (s *Server) listenForExpirationEvents() { expiredID := msg.Payload fmt.Println(expiredID) // Broadcast expiration event - broadcastExpiredRecords(s, expiredID) - fmt.Printf("Done Broadcasting Expiration\n") + var message = Message{Id: expiredID, Name: "", Image: "", Status: "removed", Timestamp: ""} + fmt.Printf("Broadcasting expiration: %s to %d clients\n", message, len(s.clients)) + broadcastSingleRecord(s, message) } } @@ -256,7 +231,6 @@ func fetchAllRecords() (map[string]string, error) { } records[key] = value } - return records, nil } @@ -333,26 +307,24 @@ func setState(w http.ResponseWriter, r *http.Request, s *Server) { http.Error(w, "Failed to delete key from Redis", http.StatusInternalServerError) return } - broadcastRemovedRecords(s, message) - w.WriteHeader(http.StatusOK) - return + fmt.Printf("Broadcasting removal: %s,%s,%s,%s to %d clients\n", message.Id, message.Status, message.Name, message.Timestamp, len(s.clients)) + } else { + msgJSON, err := json.Marshal(message) + if err != nil { + http.Error(w, "Failed to encode message", http.StatusInternalServerError) + return + } + timeout := 20 * time.Second + err = redisClient.Set(ctx, message.Id, msgJSON, timeout).Err() + if err != nil { + log.Printf("Failed to set key in Redis: %v", err) + http.Error(w, "Failed to store data", http.StatusInternalServerError) + return + } + fmt.Printf("Broadcasting message: %s,%s,%s,%s to %d clients\n", message.Id, message.Status, message.Name, message.Timestamp, len(s.clients)) } - msgJSON, err := json.Marshal(message) - if err != nil { - http.Error(w, "Failed to encode message", http.StatusInternalServerError) - return - } - - timeout := 20 * time.Second - err = redisClient.Set(ctx, message.Id, msgJSON, timeout).Err() - if err != nil { - log.Printf("Failed to set key in Redis: %v", err) - http.Error(w, "Failed to store data", http.StatusInternalServerError) - return - } - - broadcastAllRecords(s) + broadcastSingleRecord(s, message) w.WriteHeader(http.StatusOK) }