public/pogdark-api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ws-refactor #3

Merged
whysman merged 4 commits from ws-refactor into master 2025-02-26 17:00:28 +00:00
Conversation 0 Commits 4 Files Changed 1 +90 -125
1 changed files with 90 additions and 125 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

93
main.go
View File

@ -32,6 +32,7 @@ type Message struct {
Name string `json:"Name"` Name string `json:"Name"`
Image string `json:"Image"` Image string `json:"Image"`
Status string `json:"Status"` Status string `json:"Status"`
Theme string `json:"Theme"`
Timestamp string `json:"Timestamp"` Timestamp string `json:"Timestamp"`
} }
@ -44,6 +45,7 @@ type Config struct {
type Client struct { type Client struct {
conn *websocket.Conn conn *websocket.Conn
mu sync.Mutex
} }
type Server struct { type Server struct {
@ -110,7 +112,7 @@ func (s *Server) addClient(client *Client) {
s.mu.Lock() s.mu.Lock()
defer s.mu.Unlock() defer s.mu.Unlock()
s.clients[client] = true s.clients[client] = true
broadcastAllRecordsToClient(client) sendAllRecordsToClient(client)
} }
func (s *Server) removeClient(client *Client) { func (s *Server) removeClient(client *Client) {
@ -154,62 +156,30 @@ func (s *Server) handleConnections(w http.ResponseWriter, r *http.Request) {
} }
} }
func broadcastAllRecords(s *Server) { func broadcastSingleRecord(s *Server, message Message) {
allRecords, err := fetchAllRecords()
if err != nil {
log.Println("Error fetching all records:", err)
}
if len(allRecords) > 0 {
var message Message
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 to %d clients\n", message.Id, message.Status, message.Name, message.Timestamp, len(s.clients))
for client := range s.clients { for client := range s.clients {
if err := client.conn.WriteMessage(websocket.TextMessage, []byte(msgContent)); err != nil { sendMessageToClient(client, message)
log.Println("Failed to broadcast update:", err)
}
}
}
} }
} }
func broadcastRemovedRecords(s *Server, message Message) { func sendMessageToClient(client *Client, message Message) {
var msgJSON, err = json.Marshal(message) var msgJSON, err = json.Marshal(message)
if err != nil { if err != nil {
log.Println("Error marshalling json:", err) log.Println("Error marshalling json:", err)
return return
} }
fmt.Printf("Broadcasting removal: %s,%s,%s,%s to %d clients\n", message.Id, message.Status, message.Name, message.Timestamp, len(s.clients)) go sendToClient(client, msgJSON)
}
for client := range s.clients { func sendToClient(client *Client, msgJSON []byte) {
client.mu.Lock()
defer client.mu.Unlock()
if err := client.conn.WriteMessage(websocket.TextMessage, msgJSON); err != nil { if err := client.conn.WriteMessage(websocket.TextMessage, msgJSON); err != nil {
log.Println("Failed to broadcast update:", err) log.Println("Failed to send message to client:", err)
return
}
} }
} }
func broadcastExpiredRecords(s *Server, removed string) { func sendAllRecordsToClient(client *Client) {
var message = Message{Id: removed, Name: "", Image: "", Status: "removed", Timestamp: ""}
var msgJSON, err = json.Marshal(message)
if err != nil {
log.Println("Error marshalling json:", err)
return
}
fmt.Printf("Broadcasting expiration: %s to %d clients\n", string(msgJSON), len(s.clients))
for client := range s.clients {
if err := client.conn.WriteMessage(websocket.TextMessage, msgJSON); err != nil {
log.Println("Failed to broadcast update:", err)
return
}
}
}
func broadcastAllRecordsToClient(c *Client) {
allRecords, err := fetchAllRecords() allRecords, err := fetchAllRecords()
if err != nil { if err != nil {
log.Println("Error fetching all records:", err) log.Println("Error fetching all records:", err)
@ -221,10 +191,8 @@ func broadcastAllRecordsToClient(c *Client) {
if err != nil { if err != nil {
log.Println("Unable to marshal JSON due to: ", err) 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) fmt.Printf("Broadcasting %s,%s,%s,%s,%s\n", message.Id, message.Status, message.Name, message.Theme, message.Timestamp)
if err := c.conn.WriteMessage(websocket.TextMessage, []byte(msgContent)); err != nil { go sendToClient(client, []byte(msgContent))
log.Println("Failed to broadcast update:", err)
}
} }
} }
@ -233,7 +201,7 @@ func (s *Server) listenForExpirationEvents() {
defer func(ps *redis.PubSub) { defer func(ps *redis.PubSub) {
err := ps.Close() err := ps.Close()
if err != nil { if err != nil {
log.Println("Error closing redis pubsub:", err)
} }
}(ps) }(ps)
@ -242,8 +210,9 @@ func (s *Server) listenForExpirationEvents() {
expiredID := msg.Payload expiredID := msg.Payload
fmt.Println(expiredID) fmt.Println(expiredID)
// Broadcast expiration event // Broadcast expiration event
broadcastExpiredRecords(s, expiredID) var message = Message{Id: expiredID, Name: "", Image: "", Status: "removed", Theme: "", Timestamp: ""}
fmt.Printf("Done Broadcasting Expiration\n") fmt.Printf("Broadcasting expiration: %s to %d clients\n", message, len(s.clients))
broadcastSingleRecord(s, message)
} }
} }
@ -263,12 +232,10 @@ func fetchAllRecords() (map[string]string, error) {
} }
records[key] = value records[key] = value
} }
return records, nil return records, nil
} }
/* func getState(w http.ResponseWriter, r *http.Request) {
func getState(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost { if r.Method != http.MethodPost {
http.Error(w, "Only POST method is allowed", http.StatusMethodNotAllowed) http.Error(w, "Only POST method is allowed", http.StatusMethodNotAllowed)
return return
@ -284,7 +251,7 @@ func fetchAllRecords() (map[string]string, error) {
http.Error(w, "Invalid JSON format", http.StatusBadRequest) http.Error(w, "Invalid JSON format", http.StatusBadRequest)
return return
} }
fmt.Println(request)
if request.Id == "" { if request.Id == "" {
fmt.Println("Missing or empty Id field") fmt.Println("Missing or empty Id field")
http.Error(w, "Missing or empty Id field", http.StatusBadRequest) http.Error(w, "Missing or empty Id field", http.StatusBadRequest)
@ -319,8 +286,8 @@ func fetchAllRecords() (map[string]string, error) {
if err != nil { if err != nil {
http.Error(w, "Failed to encode response", http.StatusInternalServerError) http.Error(w, "Failed to encode response", http.StatusInternalServerError)
} }
} }
*/
func setState(w http.ResponseWriter, r *http.Request, s *Server) { func setState(w http.ResponseWriter, r *http.Request, s *Server) {
if r.Method != http.MethodPost { if r.Method != http.MethodPost {
http.Error(w, "Only POST method is allowed", http.StatusMethodNotAllowed) http.Error(w, "Only POST method is allowed", http.StatusMethodNotAllowed)
@ -340,17 +307,13 @@ func setState(w http.ResponseWriter, r *http.Request, s *Server) {
http.Error(w, "Failed to delete key from Redis", http.StatusInternalServerError) http.Error(w, "Failed to delete key from Redis", http.StatusInternalServerError)
return return
} }
broadcastRemovedRecords(s, message) fmt.Printf("Broadcasting removal: %s,%s,%s,%s to %d clients\n", message.Id, message.Status, message.Name, message.Timestamp, len(s.clients))
w.WriteHeader(http.StatusOK) } else {
return
}
msgJSON, err := json.Marshal(message) msgJSON, err := json.Marshal(message)
if err != nil { if err != nil {
http.Error(w, "Failed to encode message", http.StatusInternalServerError) http.Error(w, "Failed to encode message", http.StatusInternalServerError)
return return
} }
timeout := 20 * time.Second timeout := 20 * time.Second
err = redisClient.Set(ctx, message.Id, msgJSON, timeout).Err() err = redisClient.Set(ctx, message.Id, msgJSON, timeout).Err()
if err != nil { if err != nil {
@ -358,8 +321,10 @@ func setState(w http.ResponseWriter, r *http.Request, s *Server) {
http.Error(w, "Failed to store data", http.StatusInternalServerError) http.Error(w, "Failed to store data", http.StatusInternalServerError)
return return
} }
fmt.Printf("Broadcasting message: %s,%s,%s,%s,%s to %d clients\n", message.Id, message.Status, message.Name, message.Theme, message.Timestamp, len(s.clients))
}
broadcastAllRecords(s) broadcastSingleRecord(s, message)
w.WriteHeader(http.StatusOK) w.WriteHeader(http.StatusOK)
} }
@ -420,7 +385,7 @@ func main() {
// Register routes on the mux router // Register routes on the mux router
router.HandleFunc("/ws", server.handleConnections).Methods("GET") router.HandleFunc("/ws", server.handleConnections).Methods("GET")
//router.HandleFunc("/get", getState).Methods("POST") router.HandleFunc("/get", getState).Methods("POST")
router.HandleFunc("/set", func(w http.ResponseWriter, r *http.Request) { router.HandleFunc("/set", func(w http.ResponseWriter, r *http.Request) {
setState(w, r, server) setState(w, r, server)
}).Methods("POST") }).Methods("POST")