websocket stuff wip

examples/sse-with-state/.dockerignore

@@ -0,0 +1,11 @@
+# Project exclude paths
+/tmp/
+node_modules/
+dist/
+js/dist
+js/node_modules
+go.work
+go.work.sum
+.idea
+!framework/assets/dist
+__htmgo

examples/sse-with-state/.gitignore

@@ -0,0 +1,6 @@
+/assets/dist
+tmp
+node_modules
+.idea
+__htmgo
+dist

examples/sse-with-state/Dockerfile

@@ -0,0 +1,38 @@
+# Stage 1: Build the Go binary
+FROM golang:1.23-alpine AS builder
+
+RUN apk update
+RUN apk add git
+RUN apk add curl
+
+# Set the working directory inside the container
+WORKDIR /app
+
+# Copy go.mod and go.sum files
+COPY go.mod go.sum ./
+
+# Download and cache the Go modules
+RUN go mod download
+
+# Copy the source code into the container
+COPY . .
+
+# Build the Go binary for Linux
+RUN GOPRIVATE=github.com/maddalax GOPROXY=direct go run github.com/maddalax/htmgo/cli/htmgo@latest build
+
+
+# Stage 2: Create the smallest possible image
+FROM gcr.io/distroless/base-debian11
+
+# Set the working directory inside the container
+WORKDIR /app
+
+# Copy the Go binary from the    builder stage
+COPY --from=builder /app/dist .
+
+# Expose the necessary port (replace with your server port)
+EXPOSE 3000
+
+
+# Command to run the binary
+CMD ["./sse-with-state"]

examples/sse-with-state/Taskfile.yml

@@ -0,0 +1,20 @@
+version: '3'
+
+tasks:
+  run:
+    cmds:
+      - go run github.com/maddalax/htmgo/cli/htmgo@latest run
+    silent: true
+
+  build:
+    cmds:
+      - go run github.com/maddalax/htmgo/cli/htmgo@latest build
+
+  docker:
+    cmds:
+      - docker build .
+
+  watch:
+    cmds:
+      - go run github.com/maddalax/htmgo/cli/htmgo@latest watch
+    silent: true

examples/sse-with-state/assets.go

@@ -0,0 +1,13 @@
+//go:build !prod
+// +build !prod
+
+package main
+
+import (
+	"io/fs"
+	"sse-with-state/internal/embedded"
+)
+
+func GetStaticAssets() fs.FS {
+	return embedded.NewOsFs()
+}

examples/sse-with-state/assets/css/input.css

@@ -0,0 +1,3 @@
+@tailwind base;
+@tailwind components;
+@tailwind utilities;

examples/sse-with-state/assets_prod.go

@@ -0,0 +1,16 @@
+//go:build prod
+// +build prod
+
+package main
+
+import (
+	"embed"
+	"io/fs"
+)
+
+//go:embed assets/dist/*
+var staticAssets embed.FS
+
+func GetStaticAssets() fs.FS {
+	return staticAssets
+}

examples/sse-with-state/event/listener.go

@@ -0,0 +1,124 @@
+package event
+
+import (
+	"fmt"
+	"github.com/maddalax/htmgo/framework/h"
+	"github.com/maddalax/htmgo/framework/service"
+	"github.com/puzpuzpuz/xsync/v3"
+	"sse-with-state/internal"
+	"sse-with-state/sse"
+	"sse-with-state/state"
+)
+
+type HandlerData struct {
+	SessionId state.SessionId
+	Socket    *sse.SocketConnection
+	Manager   *sse.SocketManager
+}
+
+type Handler func(data HandlerData)
+
+type handlerWrapper struct {
+	handler   Handler
+	sessionId state.SessionId
+}
+
+type ServerSideEvent struct {
+	Event     string
+	Payload   map[string]any
+	SessionId state.SessionId
+}
+
+var Map = xsync.NewMapOf[string, handlerWrapper]()
+var ServerSideEventMap = xsync.NewMapOf[string, *xsync.MapOf[string, handlerWrapper]]()
+var socketMessageListener = make(chan sse.SocketEvent, 100)
+var serverSideMessageListener = make(chan ServerSideEvent, 100)
+
+func AddServerSideHandler(ctx *h.RequestContext, id string, event string, handler Handler) {
+	sessionId := state.GetSessionId(ctx)
+
+	wrapper := handlerWrapper{
+		handler:   handler,
+		sessionId: sessionId,
+	}
+
+	handlers, ok := ServerSideEventMap.Load(event)
+	if !ok {
+		ServerSideEventMap.Store(event, xsync.NewMapOf[string, handlerWrapper]())
+	}
+
+	handlers, _ = ServerSideEventMap.Load(event)
+	handlers.Store(id, wrapper)
+
+	fmt.Printf("added server side handler for %s, %v\n", event, handlers)
+}
+
+func AddHandler(ctx *h.RequestContext, event string, handler Handler) *h.AttributeMapOrdered {
+	handlerId := fmt.Sprintf("event_%s_%s", event, internal.RandSeq(30))
+	for {
+		_, ok := Map.Load(handlerId)
+		if ok {
+			handlerId = fmt.Sprintf("event_%s_%s", event, internal.RandSeq(30))
+		} else {
+			break
+		}
+	}
+	sessionId := state.GetSessionId(ctx)
+	Map.Store(handlerId, handlerWrapper{
+		handler:   handler,
+		sessionId: sessionId,
+	})
+	return h.AttributePairs(
+		"data-handler-id", handlerId,
+		"data-handler-event", event,
+	)
+}
+
+func PushServerSideEvent(sessionId state.SessionId, event string) {
+	serverSideMessageListener <- ServerSideEvent{
+		Event:     event,
+		Payload:   make(map[string]any),
+		SessionId: sessionId,
+	}
+}
+
+func PushElement(data HandlerData, el *h.Element) {
+	data.Manager.SendHtml(data.Socket.Id, h.Render(el))
+}
+
+func StartListener(locator *service.Locator) {
+	manager := service.Get[sse.SocketManager](locator)
+	manager.Listen(socketMessageListener)
+	go func() {
+		for {
+			select {
+			case sevent := <-serverSideMessageListener:
+				handlers, ok := ServerSideEventMap.Load(sevent.Event)
+				if ok {
+					handlers.Range(func(key string, value handlerWrapper) bool {
+						go value.handler(HandlerData{
+							SessionId: sevent.SessionId,
+							Socket:    manager.Get(string(sevent.SessionId)),
+							Manager:   manager,
+						})
+						return true
+					})
+				}
+			case event := <-socketMessageListener:
+				if event.Type == sse.MessageEvent {
+					handlerId := event.Payload["id"].(string)
+					eventName := event.Payload["event"].(string)
+					cb, ok := Map.Load(handlerId)
+					if ok {
+						fmt.Printf("calling %s handler for session: %s\n", eventName, cb.sessionId)
+						go cb.handler(HandlerData{
+							SessionId: cb.sessionId,
+							Socket:    manager.Get(event.SocketId),
+							Manager:   manager,
+						})
+					}
+				}
+			}
+		}
+	}()
+}

examples/sse-with-state/go.mod

@@ -0,0 +1,14 @@
+module sse-with-state
+
+go 1.23.0
+
+require github.com/maddalax/htmgo/framework v0.0.0-20241006162137-150c87b4560b
+
+require (
+	github.com/go-chi/chi/v5 v5.1.0 // indirect
+	github.com/gobwas/httphead v0.1.0 // indirect
+	github.com/gobwas/pool v0.2.1 // indirect
+	github.com/gobwas/ws v1.4.0 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	golang.org/x/sys v0.6.0 // indirect
+)

examples/sse-with-state/go.sum

@@ -0,0 +1,24 @@
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/go-chi/chi/v5 v5.1.0 h1:acVI1TYaD+hhedDJ3r54HyA6sExp3HfXq7QWEEY/xMw=
+github.com/go-chi/chi/v5 v5.1.0/go.mod h1:DslCQbL2OYiznFReuXYUmQ2hGd1aDpCnlMNITLSKoi8=
+github.com/gobwas/httphead v0.1.0 h1:exrUm0f4YX0L7EBwZHuCF4GDp8aJfVeBrlLQrs6NqWU=
+github.com/gobwas/httphead v0.1.0/go.mod h1:O/RXo79gxV8G+RqlR/otEwx4Q36zl9rqC5u12GKvMCM=
+github.com/gobwas/pool v0.2.1 h1:xfeeEhW7pwmX8nuLVlqbzVc7udMDrwetjEv+TZIz1og=
+github.com/gobwas/pool v0.2.1/go.mod h1:q8bcK0KcYlCgd9e7WYLm9LpyS+YeLd8JVDW6WezmKEw=
+github.com/gobwas/ws v1.4.0 h1:CTaoG1tojrh4ucGPcoJFiAQUAsEWekEWvLy7GsVNqGs=
+github.com/gobwas/ws v1.4.0/go.mod h1:G3gNqMNtPppf5XUz7O4shetPpcZ1VJ7zt18dlUeakrc=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/maddalax/htmgo/framework v0.0.0-20241006162137-150c87b4560b h1:LzZTNwIGe0RHiEJZlpnpN8GRnKg2lCZppMX+JIyeF/g=
+github.com/maddalax/htmgo/framework v0.0.0-20241006162137-150c87b4560b/go.mod h1:HYKI49Pb6oyY2opSJdTt145B1vWgfWIDohvlolynv80=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
+github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+golang.org/x/net v0.29.0 h1:5ORfpBpCs4HzDYoodCDBbwHzdR5UrLBZ3sOnUJmFoHo=
+golang.org/x/net v0.29.0/go.mod h1:gLkgy8jTGERgjzMic6DS9+SP0ajcu6Xu3Orq/SpETg0=
+golang.org/x/sys v0.6.0 h1:MVltZSvRTcU2ljQOhs94SXPftV6DCNnZViHeQps87pQ=
+golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

examples/sse-with-state/internal/embedded/os.go

@@ -0,0 +1,17 @@
+package embedded
+
+import (
+	"io/fs"
+	"os"
+)
+
+type OsFs struct {
+}
+
+func (receiver OsFs) Open(name string) (fs.File, error) {
+	return os.Open(name)
+}
+
+func NewOsFs() OsFs {
+	return OsFs{}
+}

examples/sse-with-state/internal/random.go

@@ -0,0 +1,13 @@
+package internal
+
+import "math/rand"
+
+var letters = []rune("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ")
+
+func RandSeq(n int) string {
+	b := make([]rune, n)
+	for i := range b {
+		b[i] = letters[rand.Intn(len(letters))]
+	}
+	return string(b)
+}

examples/sse-with-state/main.go

@@ -0,0 +1,39 @@
+package main
+
+import (
+	"github.com/maddalax/htmgo/framework/h"
+	"github.com/maddalax/htmgo/framework/service"
+	"io/fs"
+	"net/http"
+	"sse-with-state/__htmgo"
+	"sse-with-state/event"
+	"sse-with-state/sse"
+)
+
+func main() {
+	locator := service.NewLocator()
+
+	service.Set[sse.SocketManager](locator, service.Singleton, func() *sse.SocketManager {
+		return sse.NewSocketManager()
+	})
+
+	event.StartListener(locator)
+
+	h.Start(h.AppOpts{
+		ServiceLocator: locator,
+		LiveReload:     true,
+		Register: func(app *h.App) {
+			sub, err := fs.Sub(GetStaticAssets(), "assets/dist")
+
+			if err != nil {
+				panic(err)
+			}
+
+			http.FileServerFS(sub)
+
+			app.Router.Handle("/public/*", http.StripPrefix("/public", http.FileServerFS(sub)))
+			app.Router.Handle("/ws/test", sse.HandleWs())
+			__htmgo.Register(app.Router)
+		},
+	})
+}

examples/sse-with-state/pages/index.go

@@ -0,0 +1,41 @@
+package pages
+
+import (
+	"fmt"
+	"github.com/maddalax/htmgo/framework/h"
+	"sse-with-state/partials"
+	"sse-with-state/state"
+)
+
+func IndexPage(ctx *h.RequestContext) *h.Page {
+	state.NewState(ctx)
+	return h.NewPage(
+		RootPage(
+			h.Div(
+				h.Attribute("ws-connect", fmt.Sprintf("/ws/test")),
+				h.Class("flex flex-col gap-4 items-center pt-24 min-h-screen bg-neutral-100"),
+				h.H3(h.Id("intro-text"), h.Text("Repeater Example"), h.Class("text-2xl")),
+
+				partials.CounterForm(ctx, partials.CounterProps{Id: "counter-1"}),
+
+				partials.Repeater(ctx, partials.RepeaterProps{
+					Id: "repeater-1",
+					AddButton: h.Button(
+						h.Text("+ Add Item"),
+					),
+					RemoveButton: func(index int, children ...h.Ren) *h.Element {
+						return h.Button(
+							h.Text("Remove"),
+							h.Children(children...),
+						)
+					},
+					Item: func(index int) *h.Element {
+						return h.Input("text",
+							h.Class("border border-gray-300 rounded p-2"),
+							h.Value(fmt.Sprintf("item %d", index)))
+					},
+				}),
+			),
+		),
+	)
+}

examples/sse-with-state/pages/root.go

@@ -0,0 +1,21 @@
+package pages
+
+import (
+	"github.com/maddalax/htmgo/framework/h"
+)
+
+func RootPage(children ...h.Ren) h.Ren {
+	return h.Html(
+		h.HxExtension(h.BaseExtensions()),
+		h.Head(
+			h.Link("/public/main.css", "stylesheet"),
+			h.Script("/public/htmgo.js"),
+		),
+		h.Body(
+			h.Div(
+				h.Class("flex flex-col gap-2 bg-white h-full"),
+				h.Fragment(children...),
+			),
+		),
+	)
+}

examples/sse-with-state/partials/click.go

@@ -0,0 +1,19 @@
+package partials
+
+import (
+	"github.com/maddalax/htmgo/framework/h"
+	"sse-with-state/event"
+)
+
+func OnClick(ctx *h.RequestContext, handler event.Handler) *h.AttributeMapOrdered {
+	return event.AddHandler(ctx, "click", handler)
+}
+
+func OnServerSideEvent(ctx *h.RequestContext, id string, eventName string, handler event.Handler) h.Ren {
+	event.AddServerSideHandler(ctx, id, eventName, handler)
+	return h.Empty()
+}
+
+func OnMouseOver(ctx *h.RequestContext, handler event.Handler) *h.AttributeMapOrdered {
+	return event.AddHandler(ctx, "mouseover", handler)
+}

examples/sse-with-state/partials/index.go

@@ -0,0 +1,80 @@
+package partials
+
+import (
+	"github.com/maddalax/htmgo/framework/h"
+	"sse-with-state/event"
+	"sse-with-state/state"
+)
+
+func UseState[T any](sessionId state.SessionId, key string, initial T) (func() T, func(T)) {
+	var get = func() T {
+		return state.Get[T](sessionId, key, initial)
+	}
+	var set = func(value T) {
+		state.Set(sessionId, key, value)
+	}
+	return get, set
+}
+
+type Counter struct {
+	Count     func() int
+	Increment func()
+}
+
+func UseCounter(sessionId state.SessionId, id string) Counter {
+	get, set := UseState(sessionId, id, 0)
+
+	var increment = func() {
+		set(get() + 1)
+	}
+
+	return Counter{
+		Count:     get,
+		Increment: increment,
+	}
+}
+
+type CounterProps struct {
+	Id string
+}
+
+func CounterForm(ctx *h.RequestContext, props CounterProps) *h.Element {
+	if props.Id == "" {
+		props.Id = h.GenId()
+	}
+	counter := UseCounter(state.GetSessionId(ctx), props.Id)
+	return h.Div(
+		h.Attribute("hx-swap", "none"),
+		h.Class("flex flex-col gap-3 items-center"),
+		h.Id(props.Id),
+		h.P(
+			h.Id("counter-text-"+props.Id),
+			h.AttributePairs(
+				"id", "counter",
+				"class", "text-xl",
+				"name", "count",
+				"text", "count",
+			),
+			h.TextF("Count: %d", counter.Count()),
+		),
+		h.Button(
+			h.Class("bg-rose-400 hover:bg-rose-500 text-white font-bold py-2 px-4 rounded"),
+			h.Type("submit"),
+			h.Text("Increment"),
+			OnServerSideEvent(ctx, props.Id, "increment", func(data event.HandlerData) {
+				counter.Increment()
+				event.PushElement(data, CounterForm(ctx, props))
+			}),
+			//OnMouseOver(ctx, func(data event.HandlerData) {
+			//	counter.Increment()
+			//	updated := CounterForm(ctx, props)
+			//	event.PushElement(data, updated)
+			//}),
+			//OnClick(ctx, func(data event.HandlerData) {
+			//	counter.Increment()
+			//	updated := CounterForm(ctx, props)
+			//	event.PushElement(data, updated)
+			//}),
+		),
+	)
+}

examples/sse-with-state/partials/repeater.go

@@ -0,0 +1,77 @@
+package partials
+
+import (
+	"fmt"
+	"github.com/maddalax/htmgo/framework/h"
+	"sse-with-state/event"
+)
+
+type RepeaterProps struct {
+	Item         func(index int) *h.Element
+	RemoveButton func(index int, children ...h.Ren) *h.Element
+	AddButton    *h.Element
+	DefaultItems []*h.Element
+	Id           string
+	currentIndex int
+}
+
+func (props *RepeaterProps) itemId(index int) string {
+	return fmt.Sprintf("%s-repeater-item-%d", props.Id, index)
+}
+
+func (props *RepeaterProps) addButtonId() string {
+	return fmt.Sprintf("%s-repeater-add-button", props.Id)
+}
+
+func repeaterItem(ctx *h.RequestContext, item *h.Element, index int, props *RepeaterProps) *h.Element {
+	id := props.itemId(index)
+	return h.Div(
+		h.Class("flex gap-2 items-center"),
+		h.Id(id),
+		item,
+		props.RemoveButton(index,
+			h.ClassIf(index == 0, "opacity-0 disabled"),
+			h.If(index == 0, h.Disabled()),
+			OnClick(ctx, func(data event.HandlerData) {
+				props.currentIndex--
+				event.PushElement(data,
+					h.Div(
+						h.Attribute("hx-swap-oob", fmt.Sprintf("delete:#%s", id)),
+						repeaterItem(
+							ctx, item, index, props,
+						),
+					),
+				)
+			}),
+		),
+	)
+}
+
+func Repeater(ctx *h.RequestContext, props RepeaterProps) *h.Element {
+	if props.Id == "" {
+		props.Id = h.GenId()
+	}
+	return h.Div(
+		h.Class("flex flex-col gap-2"),
+		h.List(props.DefaultItems, func(item *h.Element, index int) *h.Element {
+			return repeaterItem(ctx, item, index, &props)
+		}),
+		h.Div(
+			h.Id(props.addButtonId()),
+			h.Class("flex justify-center"),
+			props.AddButton,
+			OnClick(ctx, func(data event.HandlerData) {
+				event.PushServerSideEvent(data.SessionId, "increment")
+				event.PushElement(data,
+					h.Div(
+						h.Attribute("hx-swap-oob", "beforebegin:#"+props.addButtonId()),
+						repeaterItem(
+							ctx, props.Item(props.currentIndex), props.currentIndex, &props,
+						),
+					),
+				)
+				props.currentIndex++
+			}),
+		),
+	)
+}

examples/sse-with-state/sse/handler.go

@@ -0,0 +1,203 @@
+package sse
+
+import (
+	"encoding/json"
+	"fmt"
+	"github.com/gobwas/ws"
+	"github.com/gobwas/ws/wsutil"
+	"github.com/maddalax/htmgo/framework/h"
+	"github.com/maddalax/htmgo/framework/service"
+	"log/slog"
+	"net/http"
+	"sync"
+	"time"
+)
+
+func HandleWs() http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		cc := r.Context().Value(h.RequestContextKey).(*h.RequestContext)
+		locator := cc.ServiceLocator()
+		manager := service.Get[SocketManager](locator)
+
+		sessionCookie, _ := r.Cookie("state")
+		sessionId := ""
+
+		if sessionCookie != nil {
+			sessionId = sessionCookie.Value
+		}
+
+		if sessionId == "" {
+			w.WriteHeader(http.StatusUnauthorized)
+			return
+		}
+
+		conn, _, _, err := ws.UpgradeHTTP(r, w)
+		if err != nil {
+			slog.Info("failed to upgrade", slog.String("error", err.Error()))
+			return
+		}
+		/*
+			Large buffer in case the client disconnects while we are writing
+			we don't want to block the writer
+		*/
+		done := make(chan bool, 1000)
+		writer := make(WriterChan, 1000)
+
+		wg := sync.WaitGroup{}
+
+		manager.Add("all", sessionId, writer, done)
+
+		/*
+		 * This goroutine is responsible for writing messages to the client
+		 */
+		wg.Add(1)
+		go func() {
+			defer manager.Disconnect(sessionId)
+			defer wg.Done()
+
+			defer func() {
+				fmt.Printf("empting channels\n")
+				for len(writer) > 0 {
+					<-writer
+				}
+				for len(done) > 0 {
+					<-done
+				}
+			}()
+
+			ticker := time.NewTicker(5 * time.Second)
+			defer ticker.Stop()
+
+			for {
+				select {
+				case <-done:
+					fmt.Printf("closing connection: \n")
+					return
+				case <-ticker.C:
+					manager.Ping(sessionId)
+				case message := <-writer:
+					err = wsutil.WriteServerMessage(conn, ws.OpText, []byte(message))
+					if err != nil {
+						return
+					}
+				}
+			}
+		}()
+
+		/*
+		 * This goroutine is responsible for reading messages from the client
+		 */
+		go func() {
+			defer conn.Close()
+			for {
+				msg, op, err := wsutil.ReadClientData(conn)
+				if err != nil {
+					return
+				}
+				if op != ws.OpText {
+					return
+				}
+				m := make(map[string]any)
+				err = json.Unmarshal(msg, &m)
+				if err != nil {
+					return
+				}
+				manager.OnMessage(sessionId, m)
+			}
+		}()
+
+		wg.Wait()
+	}
+}
+
+func Handle() http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		// Set the necessary headers
+		w.Header().Set("Content-Type", "text/event-stream")
+		w.Header().Set("Cache-Control", "no-cache")
+		w.Header().Set("Connection", "keep-alive")
+		w.Header().Set("Access-Control-Allow-Origin", "*") // Optional for CORS
+
+		cc := r.Context().Value(h.RequestContextKey).(*h.RequestContext)
+		locator := cc.ServiceLocator()
+		manager := service.Get[SocketManager](locator)
+
+		sessionCookie, _ := r.Cookie("state")
+		sessionId := ""
+
+		if sessionCookie != nil {
+			sessionId = sessionCookie.Value
+		}
+
+		ctx := r.Context()
+
+		/*
+			Large buffer in case the client disconnects while we are writing
+			we don't want to block the writer
+		*/
+		done := make(chan bool, 1000)
+		writer := make(WriterChan, 1000)
+
+		wg := sync.WaitGroup{}
+		wg.Add(1)
+
+		/*
+		 * This goroutine is responsible for writing messages to the client
+		 */
+		go func() {
+			defer wg.Done()
+			defer manager.Disconnect(sessionId)
+
+			defer func() {
+				fmt.Printf("empting channels\n")
+				for len(writer) > 0 {
+					<-writer
+				}
+				for len(done) > 0 {
+					<-done
+				}
+			}()
+
+			ticker := time.NewTicker(5 * time.Second)
+			defer ticker.Stop()
+
+			for {
+				select {
+				case <-ctx.Done():
+					return
+				case <-done:
+					fmt.Printf("closing connection: \n")
+					return
+				case <-ticker.C:
+					manager.Ping(sessionId)
+				case message := <-writer:
+					_, err := fmt.Fprintf(w, message)
+					if err != nil {
+						done <- true
+					} else {
+						flusher, ok := w.(http.Flusher)
+						if ok {
+							flusher.Flush()
+						}
+					}
+				}
+			}
+		}()
+
+		/**
+		 * This goroutine is responsible for adding the client to the room
+		 */
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if sessionId == "" {
+				manager.writeCloseRaw(writer, "no session")
+				return
+			}
+
+			manager.Add("all", sessionId, writer, done)
+		}()
+
+		wg.Wait()
+	}
+}

examples/sse-with-state/sse/manager.go

@@ -0,0 +1,235 @@
+package sse
+
+import (
+	"fmt"
+	"github.com/puzpuzpuz/xsync/v3"
+	"strings"
+	"time"
+)
+
+type EventType string
+type WriterChan chan string
+type DoneChan chan bool
+
+const (
+	ConnectedEvent    EventType = "connected"
+	DisconnectedEvent EventType = "disconnected"
+	MessageEvent      EventType = "message"
+)
+
+type SocketEvent struct {
+	SocketId string
+	RoomId   string
+	Type     EventType
+	Payload  map[string]any
+}
+
+type CloseEvent struct {
+	Code   int
+	Reason string
+}
+
+type SocketConnection struct {
+	Id     string
+	RoomId string
+	Done   DoneChan
+	Writer WriterChan
+}
+
+type SocketManager struct {
+	sockets   *xsync.MapOf[string, *xsync.MapOf[string, SocketConnection]]
+	idToRoom  *xsync.MapOf[string, string]
+	listeners []chan SocketEvent
+}
+
+func NewSocketManager() *SocketManager {
+	return &SocketManager{
+		sockets:  xsync.NewMapOf[string, *xsync.MapOf[string, SocketConnection]](),
+		idToRoom: xsync.NewMapOf[string, string](),
+	}
+}
+
+func (manager *SocketManager) ForEachSocket(roomId string, cb func(conn SocketConnection)) {
+	sockets, ok := manager.sockets.Load(roomId)
+	if !ok {
+		return
+	}
+	sockets.Range(func(id string, conn SocketConnection) bool {
+		cb(conn)
+		return true
+	})
+}
+
+func (manager *SocketManager) Listen(listener chan SocketEvent) {
+	if manager.listeners == nil {
+		manager.listeners = make([]chan SocketEvent, 0)
+	}
+	if listener != nil {
+		manager.listeners = append(manager.listeners, listener)
+	}
+}
+
+func (manager *SocketManager) dispatch(event SocketEvent) {
+	fmt.Printf("dispatching event: %s\n", event.Type)
+	done := make(chan struct{}, 1)
+	go func() {
+		for {
+			select {
+			case <-done:
+				fmt.Printf("dispatched event: %s\n", event.Type)
+				return
+			case <-time.After(5 * time.Second):
+				fmt.Printf("havent dispatched event after 5s, chan blocked: %s\n", event.Type)
+			}
+		}
+	}()
+	for _, listener := range manager.listeners {
+		listener <- event
+	}
+	done <- struct{}{}
+}
+
+func (manager *SocketManager) OnMessage(id string, message map[string]any) {
+	socket := manager.Get(id)
+	if socket == nil {
+		return
+	}
+	manager.dispatch(SocketEvent{
+		SocketId: id,
+		Type:     MessageEvent,
+		Payload:  message,
+		RoomId:   socket.RoomId,
+	})
+}
+
+func (manager *SocketManager) Add(roomId string, id string, writer WriterChan, done DoneChan) {
+	manager.idToRoom.Store(id, roomId)
+
+	sockets, ok := manager.sockets.LoadOrCompute(roomId, func() *xsync.MapOf[string, SocketConnection] {
+		return xsync.NewMapOf[string, SocketConnection]()
+	})
+
+	sockets.Store(id, SocketConnection{
+		Id:     id,
+		Writer: writer,
+		RoomId: roomId,
+		Done:   done,
+	})
+
+	s, ok := sockets.Load(id)
+	if !ok {
+		return
+	}
+
+	manager.dispatch(SocketEvent{
+		SocketId: s.Id,
+		Type:     ConnectedEvent,
+		RoomId:   s.RoomId,
+		Payload:  map[string]any{},
+	})
+
+	fmt.Printf("User %s connected to %s\n", id, roomId)
+}
+
+func (manager *SocketManager) OnClose(id string) {
+	socket := manager.Get(id)
+	if socket == nil {
+		return
+	}
+	manager.dispatch(SocketEvent{
+		SocketId: id,
+		Type:     DisconnectedEvent,
+		RoomId:   socket.RoomId,
+		Payload:  map[string]any{},
+	})
+	manager.sockets.Delete(id)
+}
+
+func (manager *SocketManager) CloseWithMessage(id string, message string) {
+	conn := manager.Get(id)
+	if conn != nil {
+		defer manager.OnClose(id)
+		manager.writeText(*conn, message)
+		conn.Done <- true
+	}
+}
+
+func (manager *SocketManager) Disconnect(id string) {
+	conn := manager.Get(id)
+	if conn != nil {
+		manager.OnClose(id)
+		conn.Done <- true
+	}
+}
+
+func (manager *SocketManager) Get(id string) *SocketConnection {
+	roomId, ok := manager.idToRoom.Load(id)
+	if !ok {
+		return nil
+	}
+	sockets, ok := manager.sockets.Load(roomId)
+	if !ok {
+		return nil
+	}
+	conn, ok := sockets.Load(id)
+	return &conn
+}
+
+func (manager *SocketManager) Ping(id string) {
+	conn := manager.Get(id)
+	if conn != nil {
+		manager.writeText(*conn, "ping")
+	}
+}
+
+func (manager *SocketManager) writeCloseRaw(writer WriterChan, message string) {
+	manager.writeTextRaw(writer, message)
+}
+
+func (manager *SocketManager) writeTextRaw(writer WriterChan, message string) {
+	timeout := 3 * time.Second
+	select {
+	case writer <- message:
+	case <-time.After(timeout):
+		fmt.Printf("could not send %s to channel after %s\n", message, timeout)
+	}
+}
+
+func (manager *SocketManager) writeText(socket SocketConnection, message string) {
+	if socket.Writer == nil {
+		return
+	}
+	manager.writeTextRaw(socket.Writer, message)
+}
+
+func (manager *SocketManager) BroadcastText(roomId string, message string, predicate func(conn SocketConnection) bool) {
+	sockets, ok := manager.sockets.Load(roomId)
+
+	if !ok {
+		return
+	}
+
+	sockets.Range(func(id string, conn SocketConnection) bool {
+		if predicate(conn) {
+			manager.writeText(conn, message)
+		}
+		return true
+	})
+}
+
+func (manager *SocketManager) SendHtml(id string, message string) {
+	conn := manager.Get(id)
+	minified := strings.ReplaceAll(message, "\n", "")
+	minified = strings.ReplaceAll(minified, "\t", "")
+	minified = strings.TrimSpace(minified)
+	if conn != nil {
+		manager.writeText(*conn, minified)
+	}
+}
+
+func (manager *SocketManager) SendText(id string, message string) {
+	conn := manager.Get(id)
+	if conn != nil {
+		manager.writeText(*conn, message)
+	}
+}

examples/sse-with-state/state/state.go

@@ -0,0 +1,67 @@
+package state
+
+import (
+	"github.com/google/uuid"
+	"github.com/maddalax/htmgo/framework/h"
+	"github.com/puzpuzpuz/xsync/v3"
+	"net/http"
+)
+
+type SessionId string
+
+var cache = xsync.NewMapOf[SessionId, *xsync.MapOf[string, any]]()
+
+type State struct {
+	SessionId SessionId
+}
+
+func NewState(ctx *h.RequestContext) *State {
+	id := GetSessionId(ctx)
+	cache.Store(id, xsync.NewMapOf[string, any]())
+	return &State{
+		SessionId: id,
+	}
+}
+
+func GetSessionId(ctx *h.RequestContext) SessionId {
+	stateCookie, err := ctx.Request.Cookie("state")
+	sessionId := ""
+	if err == nil {
+		sessionId = stateCookie.Value
+	} else {
+		sessionId = uuid.NewString()
+	}
+
+	c := http.Cookie{
+		Name:  "state",
+		Value: sessionId,
+	}
+	ctx.Response.Header().Set("Set-Cookie", c.String())
+
+	return SessionId(sessionId)
+}
+
+func Update[T any](sessionId SessionId, key string, compute func(prev T) T) T {
+	actual := Get[T](sessionId, key, *new(T))
+	next := compute(actual)
+	Set(sessionId, key, next)
+	return next
+}
+
+func Get[T any](sessionId SessionId, key string, fallback T) T {
+	actual, _ := cache.LoadOrCompute(sessionId, func() *xsync.MapOf[string, any] {
+		return xsync.NewMapOf[string, any]()
+	})
+	value, exists := actual.Load(key)
+	if exists {
+		return value.(T)
+	}
+	return fallback
+}
+
+func Set(sessionId SessionId, key string, value any) {
+	actual, _ := cache.LoadOrCompute(sessionId, func() *xsync.MapOf[string, any] {
+		return xsync.NewMapOf[string, any]()
+	})
+	actual.Store(key, value)
+}

examples/sse-with-state/tailwind.config.js

@@ -0,0 +1,5 @@
+/** @type {import('tailwindcss').Config} */
+module.exports = {
+  content: ["**/*.go"],
+  plugins: [],
+};

framework/assets/js/htmgo.ts

@@ -6,7 +6,8 @@ import "./htmxextensions/response-targets";
 import "./htmxextensions/mutation-error";
 import "./htmxextensions/livereload"
 import "./htmxextensions/htmgo";
-import "./htmxextensions/sse"
+import "./htmxextensions/ws"
+import "./htmxextensions/ws-event-handler"
 
 // @ts-ignore
 window.htmx = htmx;

framework/assets/js/htmxextensions/sse.ts

@@ -1,72 +0,0 @@
-import htmx from 'htmx.org'
-import {removeAssociatedScripts} from "./htmgo";
-
-let api : any = null;
-let processed = new Set<string>()
-
-htmx.defineExtension("sse", {
-    init: function (apiRef) {
-        api = apiRef;
-    },
-    // @ts-ignore
-    onEvent: function (name, evt) {
-        const target = evt.target;
-        if(!(target instanceof HTMLElement)) {
-            return
-        }
-
-        if(name === 'htmx:beforeCleanupElement') {
-            removeAssociatedScripts(target);
-        }
-
-        if(name === 'htmx:beforeProcessNode') {
-            const elements = document.querySelectorAll('[sse-connect]');
-            for (let element of Array.from(elements)) {
-                const url = element.getAttribute("sse-connect")!;
-                if(url && !processed.has(url)) {
-                    connectEventSource(element, url)
-                    processed.add(url)
-                }
-            }
-        }
-    }
-})
-
-function connectEventSource(ele: Element, url: string) {
-    if(!url) {
-        return
-    }
-    console.info('Connecting to EventSource', url)
-    const eventSource = new EventSource(url);
-
-    eventSource.addEventListener("close", function(event) {
-        htmx.trigger(ele, "htmx:sseClose", {event: event});
-    })
-
-    eventSource.onopen = function(event) {
-        htmx.trigger(ele, "htmx:sseOpen", {event: event});
-    }
-
-    eventSource.onerror = function(event) {
-        htmx.trigger(ele, "htmx:sseError", {event: event});
-        if (eventSource.readyState == EventSource.CLOSED) {
-            htmx.trigger(ele, "htmx:sseClose", {event: event});
-        }
-    }
-
-    eventSource.onmessage = function(event) {
-        const settleInfo = api.makeSettleInfo(ele);
-        htmx.trigger(ele, "htmx:sseBeforeMessage", {event: event});
-        const response = event.data
-        const fragment = api.makeFragment(response) as DocumentFragment;
-        const children = Array.from(fragment.children);
-        for (let child of children) {
-            api.oobSwap(api.getAttributeValue(child, 'hx-swap-oob') || 'true', child, settleInfo);
-            // support htmgo eval__ scripts
-            if(child.tagName === 'SCRIPT' && child.id.startsWith("__eval")) {
-                document.body.appendChild(child);
-            }
-        }
-        htmx.trigger(ele, "htmx:sseAfterMessage", {event: event});
-    }
-}

framework/assets/js/htmxextensions/ws-event-handler.ts

@@ -0,0 +1,44 @@
+import {ws} from "./ws";
+
+window.onload = addWsEventHandlers;
+
+export function addWsEventHandlers() {
+    console.log('add ws event handlers')
+    const observer = new MutationObserver(register)
+    observer.observe(document.body, {childList: true, subtree: true})
+
+    let added = new Set<string>();
+
+    function register() {
+        let ids = new Set<string>();
+        document.querySelectorAll("[data-handler-id]").forEach(element => {
+            const id = element.getAttribute("data-handler-id");
+            const event = element.getAttribute("data-handler-event");
+
+            if(id == null || event == null) {
+                return;
+            }
+
+            ids.add(id);
+            if (added.has(id)) {
+                return;
+            }
+            added.add(id);
+            element.addEventListener(event, (e) => {
+                console.log('sending event', id, event, ws)
+                if(ws != null && ws.readyState === WebSocket.OPEN) {
+                    ws.send(JSON.stringify({id, event}));
+                }
+            });
+        })
+        for (let id of added) {
+            if (!ids.has(id)) {
+                added.delete(id);
+            }
+        }
+        console.log('size', added.size)
+    }
+
+    register()
+}
+

framework/assets/js/htmxextensions/ws.ts

@@ -0,0 +1,89 @@
+import htmx from 'htmx.org'
+import {removeAssociatedScripts} from "./htmgo";
+import {addWsEventHandlers} from "./ws-event-handler";
+
+let api : any = null;
+let processed = new Set<string>()
+export let ws: WebSocket | null = null;
+
+htmx.defineExtension("ws", {
+    init: function (apiRef) {
+        api = apiRef;
+    },
+    // @ts-ignore
+    onEvent: function (name, evt) {
+        const target = evt.target;
+        if(!(target instanceof HTMLElement)) {
+            return
+        }
+
+        if(name === 'htmx:beforeCleanupElement') {
+            removeAssociatedScripts(target);
+        }
+
+        if(name === 'htmx:beforeProcessNode') {
+            const elements = document.querySelectorAll('[ws-connect]');
+            for (let element of Array.from(elements)) {
+                const url = element.getAttribute("ws-connect")!;
+                if(url && !processed.has(url)) {
+                    connectWs(element, url)
+                    processed.add(url)
+                }
+            }
+        }
+    }
+})
+
+function exponentialBackoff(attempt: number, baseDelay = 100, maxDelay = 10000) {
+    // Exponential backoff: baseDelay * (2 ^ attempt) with jitter
+    const jitter = Math.random(); // Adding randomness to prevent collisions
+    return Math.min(baseDelay * Math.pow(2, attempt) * jitter, maxDelay);
+}
+
+function connectWs(ele: Element, url: string, attempt: number = 0) {
+    if(!url) {
+        return
+    }
+    if(!url.startsWith('ws://') && !url.startsWith('wss://')) {
+        const isSecure = window.location.protocol === 'https:'
+        url = (isSecure ? 'wss://' : 'ws://') + window.location.host + url
+    }
+    console.info('connecting to ws', url)
+    ws = new WebSocket(url);
+
+    ws.addEventListener("close", function(event) {
+        htmx.trigger(ele, "htmx:wsClose", {event: event});
+        const delay = exponentialBackoff(attempt);
+        console.info(`ws closed, reconnecting in ${delay}ms`)
+        setTimeout(() => {
+            connectWs(ele, url, attempt + 1)
+        }, delay)
+    })
+
+    ws.addEventListener("open", function(event) {
+        htmx.trigger(ele, "htmx:wsOpen", {event: event});
+    })
+
+    ws.addEventListener("error", function(event) {
+        htmx.trigger(ele, "htmx:wsError", {event: event});
+    })
+
+    ws.addEventListener("message", function(event) {
+        console.debug('ws message:', event.data)
+        const settleInfo = api.makeSettleInfo(ele);
+        htmx.trigger(ele, "htmx:wsBeforeMessage", {event: event});
+        const response = event.data
+        const fragment = api.makeFragment(response) as DocumentFragment;
+        const children = Array.from(fragment.children);
+        for (let child of children) {
+            api.oobSwap(api.getAttributeValue(child, 'hx-swap-oob') || 'true', child, settleInfo);
+            // support htmgo eval__ scripts
+            if(child.tagName === 'SCRIPT' && child.id.startsWith("__eval")) {
+                document.body.appendChild(child);
+            }
+        }
+        htmx.trigger(ele, "htmx:wsAfterMessage", {event: event});
+    })
+
+    return ws
+}

framework/h/app.go

@@ -20,7 +20,7 @@ type RequestContext struct {
 	Response          http.ResponseWriter
 	locator           *service.Locator
 	isBoosted         bool
-	currentBrowserUrl string
+	CurrentBrowserUrl string
 	hxPromptResponse  string
 	isHxRequest       bool
 	hxTargetId        string
@@ -80,7 +80,7 @@ const RequestContextKey = "htmgo.request.context"
 func populateHxFields(cc *RequestContext) {
 	cc.isBoosted = cc.Request.Header.Get(hx.BoostedHeader) == "true"
 	cc.isBoosted = cc.Request.Header.Get(hx.BoostedHeader) == "true"
-	cc.currentBrowserUrl = cc.Request.Header.Get(hx.CurrentUrlHeader)
+	cc.CurrentBrowserUrl = cc.Request.Header.Get(hx.CurrentUrlHeader)
 	cc.hxPromptResponse = cc.Request.Header.Get(hx.PromptResponseHeader)
 	cc.isHxRequest = cc.Request.Header.Get(hx.RequestHeader) == "true"
 	cc.hxTargetId = cc.Request.Header.Get(hx.TargetIdHeader)

framework/h/extensions.go

@@ -3,7 +3,7 @@ package h
 import "strings"
 
 func BaseExtensions() string {
-	extensions := []string{"path-deps", "response-targets", "mutation-error", "htmgo", "sse"}
+	extensions := []string{"path-deps", "response-targets", "mutation-error", "htmgo", "sse", "ws"}
 	if IsDevelopment() {
 		extensions = append(extensions, "livereload")
 	}

framework/h/header.go

@@ -16,7 +16,7 @@ func PushUrlHeader(url string) *Headers {
 }
 
 func PushQsHeader(ctx *RequestContext, qs *Qs) *Headers {
-	parsed, err := url.Parse(ctx.currentBrowserUrl)
+	parsed, err := url.Parse(ctx.CurrentBrowserUrl)
 	if err != nil {
 		return NewHeaders()
 	}

framework/h/qs.go

@@ -51,7 +51,7 @@ func (q *Qs) ToString() string {
 func GetQueryParam(ctx *RequestContext, key string) string {
 	value, ok := ctx.Request.URL.Query()[key]
 	if value == nil || !ok {
-		current := ctx.currentBrowserUrl
+		current := ctx.CurrentBrowserUrl
 		if current != "" {
 			u, err := url.Parse(current)
 			if err == nil {

framework/h/tag.go

@@ -2,6 +2,7 @@ package h
 
 import (
 	"fmt"
+	"github.com/maddalax/htmgo/framework/internal/util"
 	"strconv"
 )
 
@@ -153,6 +154,10 @@ func Div(children ...Ren) *Element {
 	return Tag("div", children...)
 }
 
+func GenId() string {
+	return util.RandSeq(6)
+}
+
 func Article(children ...Ren) *Element {
 	return Tag("article", children...)
 }