grail/gralias
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Enha: simplify sse (worsened event recieving)

Browse Source
This commit is contained in:
Grail Finder
2025-07-12 20:33:41 +03:00
parent a934d07be3
commit 8f9865db3f
6 changed files with 37 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
broker/sse.go
View File

@@ -4,6 +4,7 @@ import (
"fmt" "fmt"
"log/slog" "log/slog"
"net/http" "net/http"
"os"
"time" "time"
) )
@@ -20,22 +21,18 @@ type (
Broker struct { Broker struct {
// Events are pushed to this channel by the main events-gathering routine // Events are pushed to this channel by the main events-gathering routine
Notifier NotifierChan Notifier NotifierChan
// New client connections log *slog.Logger
newClients chan NotifierChan
// Closed client connections
closingClients chan NotifierChan
// Client connections registry
clients map[NotifierChan]struct{}
} }
) )
func NewBroker() (broker *Broker) { func NewBroker() (broker *Broker) {
// Instantiate a broker // Instantiate a broker
return &Broker{ return &Broker{
Notifier: make(NotifierChan, 1), Notifier: make(NotifierChan, 100),
newClients: make(chan NotifierChan), log: slog.New(slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{
closingClients: make(chan NotifierChan), Level: slog.LevelDebug,
clients: make(map[NotifierChan]struct{}), AddSource: true,
})),
} }
} }
@@ -44,7 +41,6 @@ var Notifier *Broker
// for use in different packages // for use in different packages
func init() { func init() {
Notifier = NewBroker() Notifier = NewBroker()
go Notifier.Listen()
} }
func (broker *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) { func (broker *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
@@ -59,9 +55,6 @@ func (broker *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
} }
w.Header().Set("Access-Control-Allow-Origin", origin) w.Header().Set("Access-Control-Allow-Origin", origin)
w.Header().Set("Access-Control-Allow-Credentials", "true") w.Header().Set("Access-Control-Allow-Credentials", "true")
messageChan := make(NotifierChan, 10) // Buffered channel
broker.newClients <- messageChan
defer func() { broker.closingClients <- messageChan }()
ctx := r.Context() ctx := r.Context()
// browser can close sse on its own; ping every 2s to prevent // browser can close sse on its own; ping every 2s to prevent
heartbeat := time.NewTicker(2 * time.Second) heartbeat := time.NewTicker(2 * time.Second)
@@ -69,12 +62,14 @@ func (broker *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
for { for {
select { select {
case <-ctx.Done(): case <-ctx.Done():
broker.log.Debug("broker: got ctx done")
// Client disconnected // Client disconnected
return return
case event := <-messageChan: case event := <-broker.Notifier:
broker.log.Debug("got event", "event", event)
_, err := fmt.Fprintf(w, "event: %s\ndata: %s\n\n", event.EventName, event.Payload) _, err := fmt.Fprintf(w, "event: %s\ndata: %s\n\n", event.EventName, event.Payload)
if err != nil { if err != nil {
fmt.Println(err) broker.log.Error("failed to write event", "error", err)
// Client disconnected // Client disconnected
return return
} }
@@ -82,38 +77,10 @@ func (broker *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
case <-heartbeat.C: case <-heartbeat.C:
// Send SSE heartbeat comment // Send SSE heartbeat comment
if _, err := fmt.Fprint(w, ":\n\n"); err != nil { if _, err := fmt.Fprint(w, ":\n\n"); err != nil {
broker.log.Error("failed to write heartbeat", "error", err)
return // Client disconnected return // Client disconnected
} }
w.(http.Flusher).Flush() w.(http.Flusher).Flush()
} }
} }
} }
// Listen for new notifications and redistribute them to clients
func (broker *Broker) Listen() {
for {
select {
case s := <-broker.newClients:
// A new client has connected.
// Register their message channel
broker.clients[s] = struct{}{}
slog.Info("Client added", "clients listening", len(broker.clients))
case s := <-broker.closingClients:
// A client has dettached and we want to
// stop sending them messages.
delete(broker.clients, s)
slog.Info("Client removed", "clients listening", len(broker.clients))
case event := <-broker.Notifier:
// We got a new event from the outside!
// Send event to all connected clients
for clientMessageChan := range broker.clients {
select {
case clientMessageChan <- event:
case <-time.After(patience):
delete(broker.clients, clientMessageChan)
slog.Info("Client was removed", "clients listening", len(broker.clients))
}
}
}
}
}

6
components/base.html
View File

@@ -13,6 +13,12 @@
</head> </head>
<body> <body>
<div id="ancestor" hx-ext="sse" sse-connect="/sub/sse"> <div id="ancestor" hx-ext="sse" sse-connect="/sub/sse">
<script type="text/javascript">
document.body.addEventListener('htmx:sseError', function (e) {
// do something before the event data is swapped in
console.log(e)
})
</script>
<div id="main-content"> <div id="main-content">
{{template "main" .}} {{template "main" .}}
</div> </div>

1
components/index.html
View File

@@ -22,6 +22,7 @@
{{template "roomlist" .List}} {{template "roomlist" .List}}
</div> </div>
{{else}} {{else}}
<div id="sse-listener" sse-connect="/sub/sse" hx-trigger="sse:roomupdate_{{.State.RoomID}}" hx-get="/room" hx-target="#room-interier" hx-swap="none" style="display:none;"></div>
<div id="room"> <div id="room">
{{template "room" .}} {{template "room" .}}
</div> </div>

2
components/room.html
View File

@@ -1,5 +1,5 @@
{{define "room"}} {{define "room"}}
<div id="interier" hx-get="/" hx-trigger="sse:roomupdate_{{.State.RoomID}}" class=space-y-2> <div id="room-interier" class=space-y-2>
<div id="meta"> <div id="meta">
<p>Hello {{.State.Username}};</p> <p>Hello {{.State.Username}};</p>
<p>Room created by {{.Room.CreatorName}};</p> <p>Room created by {{.Room.CreatorName}};</p>

16
handlers/elements.go
View File

@@ -329,3 +329,19 @@ func HandleRemoveBot(w http.ResponseWriter, r *http.Request) {
} }
notify(models.NotifyRoomUpdatePrefix+fi.Room.ID, "") notify(models.NotifyRoomUpdatePrefix+fi.Room.ID, "")
} }
func HandleGetRoom(w http.ResponseWriter, r *http.Request) {
fi, err := getFullInfoByCtx(r.Context())
if err != nil {
abortWithError(w, err.Error())
return
}
tmpl, err := template.ParseGlob("components/*.html")
if err != nil {
abortWithError(w, err.Error())
return
}
if err := tmpl.ExecuteTemplate(w, "room", fi); err != nil {
log.Error("failed to execute template", "error", err)
}
}

1
main.go
View File

@@ -92,6 +92,7 @@ func ListenToRequests(port string) *http.Server {
mux.HandleFunc("GET /add-bot", handlers.HandleAddBot) mux.HandleFunc("GET /add-bot", handlers.HandleAddBot)
mux.HandleFunc("GET /remove-bot", handlers.HandleRemoveBot) mux.HandleFunc("GET /remove-bot", handlers.HandleRemoveBot)
mux.HandleFunc("GET /mark-card", handlers.HandleMarkCard) mux.HandleFunc("GET /mark-card", handlers.HandleMarkCard)
mux.HandleFunc("GET /room", handlers.HandleGetRoom)
// special // special
mux.HandleFunc("GET /renotify-bot", handlers.HandleRenotifyBot) mux.HandleFunc("GET /renotify-bot", handlers.HandleRenotifyBot)
// sse // sse