Each event is named after a fragment of the page, and carries its HTML. htmx’s SSE extension swaps it into the element of that name.
- A stream subscribes, and sends every fragment: the baseline.
- The store changes the runs, and notifies the topic runs.
- The stream gathers changes for a moment, renders the fragments that depend on runs, and sends only those whose HTML is new.
The stream gathers changes for a moment, renders the fragments that depend on runs, and sends only those whose HTML is new.
The baseline comes first, because the page can be older than a change that the stream did not hear of. So a stream needs no event IDs, and a reconnect loses nothing.
ctx, stop := context.WithCancel(context.Background())
defer stop()
store := &Store{}
stream := func(w http.ResponseWriter, r *http.Request) {
_ = sse.Serve(w, r, &store.notifier, func(changed []string) ([]sse.Event, error) {
if changed == nil {
changed = []string{"runs"}
}
var es []sse.Event
for _, t := range changed {
es = append(es, store.fragment(t))
}
return es, nil
}, sse.Options{Done: ctx.Done()})
}
srv := httptest.NewServer(localweb.SameOrigin(localweb.OriginPolicy{})(http.HandlerFunc(stream)))
defer srv.Close()
resp, err := http.Get(srv.URL)
if err != nil {
fmt.Println(err)
return
}
defer resp.Body.Close()
events := localwebtest.NewReader(resp.Body)
e, _ := events.Next()
fmt.Println(e.Name, e.Data)
store.AddRun("create")
e, _ = events.Next()
fmt.Println(e.Name, e.Data)runs <ul>0 runs</ul>
runs <ul>1 runs</ul>localweb/sse/example_test.go. CI runs it and checks its output.| Name | What it does |
|---|---|
sse.Notifier, sse.Subscription | the topics that change, told to each subscriber once |
sse.Serve, sse.Options | the stream: the baseline, then what changed |
sse.Event, sse.Format | one event; a data line for each line of HTML |
The stream ends when localweb.Serve starts to stop, so it does not hold
the stop for the grace period.