This commit is contained in:
parent
c812af90ad
commit
497abfe83f
2 changed files with 62 additions and 24 deletions
58
api/sse.go
58
api/sse.go
|
@ -2,6 +2,7 @@ package api
|
|||
|
||||
import (
|
||||
"fmt"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"atri.dad/lib"
|
||||
|
@ -26,51 +27,66 @@ func SSE(c echo.Context) error {
|
|||
// Use the request context, which is cancelled when the client disconnects
|
||||
ctx := c.Request().Context()
|
||||
|
||||
c.Response().Header().Set(echo.HeaderContentType, "text/event-stream")
|
||||
c.Response().Header().Set(echo.HeaderConnection, "keep-alive")
|
||||
c.Response().Header().Set(echo.HeaderCacheControl, "no-cache")
|
||||
// Set headers for SSE
|
||||
c.Response().Header().Set("Content-Type", "text/event-stream")
|
||||
c.Response().Header().Set("Cache-Control", "no-cache, no-store, must-revalidate")
|
||||
c.Response().Header().Set("Connection", "keep-alive")
|
||||
c.Response().Header().Set("Pragma", "no-cache")
|
||||
c.Response().Header().Set("Expires", "0")
|
||||
c.Response().Header().Set("X-Accel-Buffering", "no")
|
||||
|
||||
// Get origin from request
|
||||
origin := c.Request().Header.Get(echo.HeaderOrigin)
|
||||
// Only allow specific origins
|
||||
// Set CORS headers
|
||||
origin := c.Request().Header.Get("Origin")
|
||||
if origin == "https://atri.dad" || origin == "http://localhost:3000" {
|
||||
c.Response().Header().Set(echo.HeaderAccessControlAllowOrigin, origin)
|
||||
c.Response().Header().Set(echo.HeaderAccessControlAllowCredentials, "true")
|
||||
c.Response().Header().Set("Access-Control-Allow-Origin", origin)
|
||||
c.Response().Header().Set("Access-Control-Allow-Credentials", "true")
|
||||
} else {
|
||||
// Allow any origin as fallback for testing
|
||||
c.Response().Header().Set("Access-Control-Allow-Origin", "*")
|
||||
}
|
||||
|
||||
// Create a channel to receive messages from the lib.SSEServer
|
||||
// Set response status
|
||||
c.Response().WriteHeader(http.StatusOK)
|
||||
c.Response().Flush()
|
||||
|
||||
// Create a channel to receive messages
|
||||
clientChan := make(chan string)
|
||||
|
||||
// Add the client to the lib.SSEServer
|
||||
// Add the client to the SSEServer
|
||||
lib.SSEServer.AddClient(channel, clientChan)
|
||||
|
||||
defer func() {
|
||||
// Remove the client from the lib.SSEServer when the connection is closed
|
||||
lib.SSEServer.RemoveClient(channel, clientChan)
|
||||
}()
|
||||
// Clean up when the connection is closed
|
||||
defer lib.SSEServer.RemoveClient(channel, clientChan)
|
||||
|
||||
// Create a ticker that fires every 15 seconds
|
||||
ticker := time.NewTicker(30 * time.Second)
|
||||
// Write the initial message
|
||||
initialMsg := ": connected\n\n"
|
||||
if _, err := c.Response().Write([]byte(initialMsg)); err != nil {
|
||||
return err
|
||||
}
|
||||
c.Response().Flush()
|
||||
|
||||
// Create a ticker for keep-alive messages
|
||||
ticker := time.NewTicker(15 * time.Second)
|
||||
defer ticker.Stop()
|
||||
|
||||
// Event loop
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
// If the client has disconnected, stop the loop
|
||||
// Client disconnected
|
||||
return nil
|
||||
case <-ticker.C:
|
||||
// Every 30 seconds, send a comment to keep the connection alive
|
||||
if _, err := c.Response().Write([]byte(": keep-alive\n\n")); err != nil {
|
||||
// Send keep-alive comment
|
||||
if _, err := c.Response().Write([]byte(": ping\n\n")); err != nil {
|
||||
return err
|
||||
}
|
||||
c.Response().Flush()
|
||||
case msg := <-clientChan:
|
||||
// Handle incoming messages from the lib.SSEServer
|
||||
// Format the SSE message
|
||||
data := fmt.Sprintf("data: %s\n\n", msg)
|
||||
if _, err := c.Response().Write([]byte(data)); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
c.Response().Flush()
|
||||
}
|
||||
}
|
||||
|
|
|
@ -7,7 +7,7 @@ Atridad Lahiji // Tools // SSE Demo
|
|||
{{end}}
|
||||
|
||||
{{define "description"}}
|
||||
A demo of my SSE implimentation.
|
||||
A demo of my SSE implementation.
|
||||
{{end}}
|
||||
|
||||
{{define "head"}}
|
||||
|
@ -17,9 +17,9 @@ A demo of my SSE implimentation.
|
|||
<h2 class="text-2xl font-extrabold tracking-tight text-white sm:text-[2rem]">Server Sent Events Demo</h2>
|
||||
|
||||
<p class="text-lg">This page demonstrates the use of the <a href="https://htmx.org/extensions/sse/">HTMX SSE
|
||||
Extention</a> to receive Server Sent Events on the "default" channel.</p>
|
||||
Extension</a> to receive Server Sent Events on the "default" channel.</p>
|
||||
<p class="text-lg">Any events received on the "default" channel will appear below:</p>
|
||||
<div hx-ext="sse" sse-connect="/api/sse" sse-swap="message">
|
||||
<div id="sse-target" hx-ext="sse" sse-connect="/api/sse" sse-swap="message">
|
||||
Waiting for SSE Message...
|
||||
</div>
|
||||
|
||||
|
@ -37,4 +37,26 @@ A demo of my SSE implimentation.
|
|||
|
||||
{{define "foot"}}
|
||||
<script src="/public/js/htmx.sse.js"></script>
|
||||
<script>
|
||||
// Custom EventSource factory with better retry handling
|
||||
htmx.createEventSource = function(url) {
|
||||
console.log("Creating EventSource for:", url);
|
||||
|
||||
// Create EventSource with credentials
|
||||
const source = new EventSource(url, { withCredentials: true });
|
||||
|
||||
// Log connection state changes
|
||||
source.onopen = function() {
|
||||
console.log("SSE connection opened successfully");
|
||||
document.getElementById('sse-target').innerHTML = "Connected to SSE. Waiting for messages...";
|
||||
};
|
||||
|
||||
source.onerror = function(err) {
|
||||
console.error("SSE connection error:", err);
|
||||
document.getElementById('sse-target').innerHTML = "Connection error. Retrying...";
|
||||
};
|
||||
|
||||
return source;
|
||||
};
|
||||
</script>
|
||||
{{end}}
|
||||
|
|
Loading…
Add table
Reference in a new issue