From c5d261250b3c3ba8886f98c02fbedd576e352a2e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Milan=20Obrtl=C3=ADk?= Date: Fri, 31 Jul 2026 12:24:51 +0200 Subject: [PATCH 1/3] Reuse the existing EventSource when an element is processed again htmx re-fires htmx:afterProcessNode whenever an element's attributes change, which is what a morph style swap does. Right before firing it, htmx wipes the element's internal data, so the extension lost its reference to the connection it had already opened and opened a second one. The first stayed open forever, holding a connection slot on the server and counting against the browser's per domain SSE limit. Keep the connection in a WeakMap that survives the internal data reset, so an element that is already connected to the same url keeps its stream instead of opening another one. When the element asks for a different connection, or drops sse-connect entirely, close the previous stream and rebind the descendants that were listening on it. --- src/sse/sse.js | 118 +++++++++++++++++++++------- src/sse/test/ext/sse.js | 168 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 257 insertions(+), 29 deletions(-) diff --git a/src/sse/sse.js b/src/sse/sse.js index 886e3f9..075967c 100644 --- a/src/sse/sse.js +++ b/src/sse/sse.js @@ -9,6 +9,19 @@ This extension adds support for Server Sent Events to htmx. See /www/extensions /** @type {import("../htmx").HtmxInternalApi} */ var api + /** + * sseSources keeps a reference to the EventSource that was created for an element, + * together with the attribute values it was created from. + * + * htmx fires htmx:afterProcessNode again every time an element's attributes change, + * and it wipes that element's internal data right before doing so. Without this map + * the extension has no reference left to the connection it opened earlier, so it + * would open a second one and leave the first one open forever. + * + * @type {WeakMap} + */ + var sseSources = new WeakMap() + htmx.defineExtension('sse', { /** @@ -42,16 +55,8 @@ This extension adds support for Server Sent Events to htmx. See /www/extensions var parent = evt.target || evt.detail.elt switch (name) { case 'htmx:beforeCleanupElement': - var internalData = api.getInternalData(parent) // Try to remove remove an EventSource when elements are removed - var source = internalData.sseEventSource - if (source) { - api.triggerEvent(parent, 'htmx:sseClose', { - source, - type: 'nodeReplaced', - }) - internalData.sseEventSource.close() - } + closeEventSource(parent, 'nodeReplaced') return @@ -125,6 +130,7 @@ This extension adds support for Server Sent Events to htmx. See /www/extensions // Register the new listener api.getInternalData(elt).sseEventListener = listener + api.getInternalData(elt).sseEventListenerSource = source source.addEventListener(sseEventName, listener) } } @@ -162,6 +168,7 @@ This extension adds support for Server Sent Events to htmx. See /www/extensions // Register the new listener api.getInternalData(elt).sseEventListener = listener + api.getInternalData(elt).sseEventListenerSource = source source.addEventListener(ts.trigger.slice(4), listener) }) } @@ -181,19 +188,49 @@ This extension adds support for Server Sent Events to htmx. See /www/extensions } // handle extension source creation attribute - if (api.getAttributeValue(elt, 'sse-connect')) { - var sseURL = api.getAttributeValue(elt, 'sse-connect') - if (sseURL == null) { - return - } - + var sseURL = api.getAttributeValue(elt, 'sse-connect') + if (sseURL) { ensureEventSource(elt, sseURL, retryCount) + } else { + // the element does not ask for a connection anymore, close the one it owned + closeEventSource(elt, 'sourceReplaced') } registerSSE(elt) } + /** + * ensureEventSource connects the given element to the given url. + * + * If the element already is connected to that url and the connection is still + * usable, then the existing EventSource is kept, and its reference is restored in + * the element's internal data, which htmx wipes every time it processes the element. + * + * @param {HTMLElement} elt + * @param {string} url + * @param {number} retryCount + */ function ensureEventSource(elt, url, retryCount) { + var closeAttribute = api.getAttributeValue(elt, 'sse-close') + var previous = sseSources.get(elt) + var replacedSource = false + + if (previous) { + if (previous.source.readyState !== EventSource.CLOSED) { + if (previous.url === url && previous.closeAttribute === closeAttribute) { + // Already connected, keep the stream instead of opening a second one + api.getInternalData(elt).sseEventSource = previous.source + return + } + + // The element asks for a different connection now, close the previous one + closeEventSource(elt, 'sourceReplaced') + } + + // Descendants are still listening on the source that is being replaced + replacedSource = true + } + var source = htmx.createEventSource(url) source.onerror = function(err) { @@ -219,20 +256,23 @@ This extension adds support for Server Sent Events to htmx. See /www/extensions source.onopen = function(evt) { api.triggerEvent(elt, 'htmx:sseOpen', { source }) - if (retryCount && retryCount > 0) { + if (replacedSource || (retryCount && retryCount > 0)) { const childrenToFix = elt.querySelectorAll("[sse-swap], [data-sse-swap], [hx-trigger], [data-hx-trigger]") for (let i = 0; i < childrenToFix.length; i++) { - registerSSE(childrenToFix[i]) + // children processed after this source was created listen on it already + if (api.getInternalData(childrenToFix[i]).sseEventListenerSource !== source) { + registerSSE(childrenToFix[i]) + } } + replacedSource = false // We want to increase the reconnection delay for consecutive failed attempts only retryCount = 0 } } api.getInternalData(elt).sseEventSource = source + sseSources.set(elt, { source, url, closeAttribute }) - - var closeAttribute = api.getAttributeValue(elt, "sse-close"); if (closeAttribute) { // close eventsource when this message is received source.addEventListener(closeAttribute, function() { @@ -254,20 +294,40 @@ This extension adds support for Server Sent Events to htmx. See /www/extensions */ function maybeCloseSSESource(elt) { if (!api.bodyContains(elt)) { - var source = api.getInternalData(elt).sseEventSource - if (source != undefined) { - api.triggerEvent(elt, 'htmx:sseClose', { - source, - type: 'nodeMissing', - }) - source.close() - // source = null - return true - } + return closeEventSource(elt, 'nodeMissing') } return false } + /** + * closeEventSource closes the EventSource that was created for the given element, + * if there still is one, and reports the reason it was closed. + * + * The source is looked up in sseSources as well, because htmx wipes an element's + * internal data whenever it processes that element again. + * + * @param {HTMLElement} elt + * @param {string} type + * @returns boolean whether a source was closed + */ + function closeEventSource(elt, type) { + var internalData = api.getInternalData(elt) + var tracked = sseSources.get(elt) + var source = internalData.sseEventSource || (tracked && tracked.source) + if (!source) { + return false + } + + api.triggerEvent(elt, 'htmx:sseClose', { + source, + type, + }) + source.close() + sseSources.delete(elt) + delete internalData.sseEventSource + return true + } + /** * @param {HTMLElement} elt diff --git a/src/sse/test/ext/sse.js b/src/sse/test/ext/sse.js index 3a20bfe..f1fc55b 100644 --- a/src/sse/test/ext/sse.js +++ b/src/sse/test/ext/sse.js @@ -66,9 +66,11 @@ describe('sse extension', function() { this.clock = sinon.useFakeTimers(); var test = this clearWorkArea() + this.eventSources = [] htmx.createEventSource = function(url) { var eventSource = mockEventSource() test.eventSource = eventSource + test.eventSources.push(eventSource) eventSource.connect(url) return eventSource } @@ -702,4 +704,170 @@ describe('sse extension', function() { byId('d1').innerHTML.should.equal('div1 updated') byId('d2').innerHTML.should.equal('div2 updated') }) + + // htmx fires htmx:afterProcessNode again whenever an element's attributes change, + // which is what a morph style swap does, so the extension has to cope with being + // asked to connect an element that is already connected. + it('reuses the existing EventSource when the element is processed again', function() { + var div = make('
' + + '
init
' + + '
') + this.clock.tick(1) + + var first = this.eventSource + + div.setAttribute('data-morphed', 'true') + htmx.process(div) + this.clock.tick(1) + + this.eventSources.should.be.lengthOf(1) + this.eventSource.should.equal(first) + first.readyState.should.equal(EventSource.OPEN) + div['htmx-internal-data'].sseEventSource.should.equal(first) + }) + + it('keeps children bound to the reused EventSource', function() { + var div = make('
' + + '
init
' + + '
') + this.clock.tick(1) + + var first = this.eventSource + + div.setAttribute('data-morphed', 'true') + htmx.process(div) + this.clock.tick(1) + + first._listeners.e1.should.be.lengthOf(1) + + this.eventSource.sendEvent('e1', 'Event 1') + byId('d1').innerHTML.should.equal('Event 1') + }) + + it('does not raise htmx:sseClose when the connection is reused', function() { + var div = make('
') + this.clock.tick(1) + + var closeCalled = false + htmx.on(div, 'htmx:sseClose', function() { + closeCalled = true + }) + + div.setAttribute('data-morphed', 'true') + htmx.process(div) + this.clock.tick(1) + + closeCalled.should.equal(false) + }) + + it('replaces the EventSource when sse-connect changes', function() { + var div = make('
' + + '
init
' + + '
') + this.clock.tick(1) + + var first = this.eventSource + var test = this + htmx.on(div, 'htmx:sseClose', function(evt) { + test.closeType = evt.detail.type + }) + + div.setAttribute('sse-connect', '/bar') + htmx.process(div) + this.clock.tick(1) + + first.readyState.should.equal(EventSource.CLOSED) + this.closeType.should.equal('sourceReplaced') + this.eventSource.should.not.equal(first) + this.eventSource.url.should.equal('/bar') + this.eventSource._listeners.e1.should.be.lengthOf(1) + + this.eventSource.sendEvent('e1', 'Event 1') + byId('d1').innerHTML.should.equal('Event 1') + }) + + it('does not register duplicate listeners when children are processed again with a new EventSource', function() { + var div = make('
' + + '
init
' + + '
') + this.clock.tick(1) + + var messageCount = 0 + div.addEventListener('htmx:sseMessage', function() { + messageCount++ + }) + + div.setAttribute('sse-connect', '/bar') + byId('d1').setAttribute('data-morphed', 'true') + htmx.process(div) + this.clock.tick(1) + + this.eventSource._listeners.e1.should.be.lengthOf(1) + + this.eventSource.sendEvent('e1', 'Event 1') + messageCount.should.equal(1) + }) + + it('closes the EventSource when sse-connect is removed', function() { + var div = make('
') + this.clock.tick(1) + + var first = this.eventSource + var test = this + htmx.on(div, 'htmx:sseClose', function(evt) { + test.closeType = evt.detail.type + }) + + div.removeAttribute('sse-connect') + htmx.process(div) + this.clock.tick(1) + + first.readyState.should.equal(EventSource.CLOSED) + this.closeType.should.equal('sourceReplaced') + this.eventSources.should.be.lengthOf(1) + }) + + it('does not open a duplicate EventSource when processed while a reconnect is pending', function() { + var div = make('
' + + '
init
' + + '
') + this.clock.tick(1) + + this.eventSource.simulateConnectionError() + + div.setAttribute('data-morphed', 'true') + htmx.process(div) + this.clock.tick(1) + + // the retry scheduled by the connection error must find the fresh connection + this.clock.tick(500) + + this.eventSources.should.be.lengthOf(2) + this.eventSource._listeners.e1.should.be.lengthOf(1) + + this.eventSource.sendEvent('e1', 'Event 1') + byId('d1').innerHTML.should.equal('Event 1') + }) + + it('opens a new EventSource after the connection was closed', function() { + var div = make('
' + + '
init
' + + '
') + this.clock.tick(1) + + var first = this.eventSource + first.sendEvent('close') + first.readyState.should.equal(EventSource.CLOSED) + + div.setAttribute('data-morphed', 'true') + htmx.process(div) + this.clock.tick(1) + + this.eventSources.should.be.lengthOf(2) + this.eventSource.should.not.equal(first) + this.eventSource._listeners.e1.should.be.lengthOf(1) + + this.eventSource.sendEvent('e1', 'Event 1') + byId('d1').innerHTML.should.equal('Event 1') + }) }) From c192076c24c18f6de11c07eb38bb1f131634b10d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Milan=20Obrtl=C3=ADk?= Date: Fri, 31 Jul 2026 12:56:47 +0200 Subject: [PATCH 2/3] Remove the listeners an element registered before registering it again registerSSE always added a listener and never removed the previous one, and it could not: htmx wipes the element's internal data before re-firing htmx:afterProcessNode, so sseEventListener no longer referred to what had been registered earlier. An element processed twice ended up listening for the same event twice and swapped every message twice. Track the registered listeners in a WeakMap that survives the internal data reset, and drop them at the start of registerSSE. This also makes the onopen rebinding safe for descendants that were already registered against the new connection in the same htmx.process pass. --- src/sse/sse.js | 64 +++++++++++++++++++++++++++++++++++------ src/sse/test/ext/sse.js | 45 +++++++++++++++++++++++++++++ 2 files changed, 101 insertions(+), 8 deletions(-) diff --git a/src/sse/sse.js b/src/sse/sse.js index 075967c..1b94cc8 100644 --- a/src/sse/sse.js +++ b/src/sse/sse.js @@ -22,6 +22,18 @@ This extension adds support for Server Sent Events to htmx. See /www/extensions */ var sseSources = new WeakMap() + /** + * sseListeners keeps the listeners that were registered for an element, so that they + * can be removed before the element is registered again. + * + * This is needed for the same reason as sseSources: htmx wipes an element's internal + * data before it re-fires htmx:afterProcessNode, so sseEventListener no longer refers + * to what was registered earlier, and the element would end up listening twice. + * + * @type {WeakMap>} + */ + var sseListeners = new WeakMap() + htmx.defineExtension('sse', { /** @@ -90,6 +102,10 @@ This extension adds support for Server Sent Events to htmx. See /www/extensions * @param {HTMLElement} elt */ function registerSSE(elt) { + // Drop what this element registered earlier, so that processing it again does not + // leave it listening for the same event twice + removeSSEListeners(elt) + // Add message handlers for every `sse-swap` attribute if (api.getAttributeValue(elt, 'sse-swap')) { // Find closest existing event source @@ -130,8 +146,7 @@ This extension adds support for Server Sent Events to htmx. See /www/extensions // Register the new listener api.getInternalData(elt).sseEventListener = listener - api.getInternalData(elt).sseEventListenerSource = source - source.addEventListener(sseEventName, listener) + addSSEListener(elt, source, sseEventName, listener) } } @@ -168,8 +183,7 @@ This extension adds support for Server Sent Events to htmx. See /www/extensions // Register the new listener api.getInternalData(elt).sseEventListener = listener - api.getInternalData(elt).sseEventListenerSource = source - source.addEventListener(ts.trigger.slice(4), listener) + addSSEListener(elt, source, ts.trigger.slice(4), listener) }) } } @@ -259,10 +273,7 @@ This extension adds support for Server Sent Events to htmx. See /www/extensions if (replacedSource || (retryCount && retryCount > 0)) { const childrenToFix = elt.querySelectorAll("[sse-swap], [data-sse-swap], [hx-trigger], [data-hx-trigger]") for (let i = 0; i < childrenToFix.length; i++) { - // children processed after this source was created listen on it already - if (api.getInternalData(childrenToFix[i]).sseEventListenerSource !== source) { - registerSSE(childrenToFix[i]) - } + registerSSE(childrenToFix[i]) } replacedSource = false // We want to increase the reconnection delay for consecutive failed attempts only @@ -299,6 +310,43 @@ This extension adds support for Server Sent Events to htmx. See /www/extensions return false } + /** + * addSSEListener starts listening for an event on the given source, and keeps track + * of the listener so that it can be removed again. + * + * @param {HTMLElement} elt + * @param {EventSource} source + * @param {string} name + * @param {Function} listener + */ + function addSSEListener(elt, source, name, listener) { + var listeners = sseListeners.get(elt) + if (!listeners) { + listeners = [] + sseListeners.set(elt, listeners) + } + + listeners.push({ source, name, listener }) + source.addEventListener(name, listener) + } + + /** + * removeSSEListeners stops listening for every event this element was registered for. + * + * @param {HTMLElement} elt + */ + function removeSSEListeners(elt) { + var listeners = sseListeners.get(elt) + if (!listeners) { + return + } + + for (var i = 0; i < listeners.length; i++) { + listeners[i].source.removeEventListener(listeners[i].name, listeners[i].listener) + } + sseListeners.delete(elt) + } + /** * closeEventSource closes the EventSource that was created for the given element, * if there still is one, and reports the reason it was closed. diff --git a/src/sse/test/ext/sse.js b/src/sse/test/ext/sse.js index f1fc55b..c1a37d3 100644 --- a/src/sse/test/ext/sse.js +++ b/src/sse/test/ext/sse.js @@ -786,6 +786,51 @@ describe('sse extension', function() { byId('d1').innerHTML.should.equal('Event 1') }) + it('does not register duplicate listeners when the element is processed again', function() { + // the same element owns the connection and is a swap target + var div = make('
initial
') + this.clock.tick(1) + + var first = this.eventSource + var messageCount = 0 + div.addEventListener('htmx:sseMessage', function() { + messageCount++ + }) + + div.setAttribute('data-morphed', 'true') + htmx.process(div) + this.clock.tick(1) + + first._listeners.e1.should.be.lengthOf(1) + + this.eventSource.sendEvent('e1', 'Event 1') + messageCount.should.equal(1) + byId('d1').innerHTML.should.equal('Event 1') + }) + + it('does not register duplicate listeners when a child is processed again', function() { + var div = make('
' + + '
initial
' + + '
') + this.clock.tick(1) + + var first = this.eventSource + var messageCount = 0 + div.addEventListener('htmx:sseMessage', function() { + messageCount++ + }) + + byId('d1').setAttribute('data-morphed', 'true') + htmx.process(div) + this.clock.tick(1) + + first._listeners.e1.should.be.lengthOf(1) + + this.eventSource.sendEvent('e1', 'Event 1') + messageCount.should.equal(1) + byId('d1').innerHTML.should.equal('Event 1') + }) + it('does not register duplicate listeners when children are processed again with a new EventSource', function() { var div = make('
' + '
init
' + From 692460bcd57af452ea3ce9eebf7931441a9f6495 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Milan=20Obrtl=C3=ADk?= Date: Tue, 1 Sep 2026 08:35:07 +0200 Subject: [PATCH 3/3] Test that an outer reconnect leaves a nested stream's listeners alone The onopen rebinding walks the whole subtree with querySelectorAll, but registerSSE resolves every match against its own closest source. A descendant that sits under a nested sse-connect is therefore rebound onto the inner stream, which never dropped and is still holding the listener it registered at startup. Every reconnect of the outer stream stacked one more, and every message rendered one more time. Dropping the listeners at the start of registerSSE already covers this, but the nested fixtures only reach the initial subscription, so nothing would catch the day that call is removed as redundant. Reported against #191 by jeffothy, who hit it in production: an app shell holding a user scoped stream, a chat panel nested inside holding its own, and messages doubling every time the device woke up. --- src/sse/test/ext/sse.js | 31 +++++++++++++++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/src/sse/test/ext/sse.js b/src/sse/test/ext/sse.js index c1a37d3..9f9036a 100644 --- a/src/sse/test/ext/sse.js +++ b/src/sse/test/ext/sse.js @@ -915,4 +915,35 @@ describe('sse extension', function() { this.eventSource.sendEvent('e1', 'Event 1') byId('d1').innerHTML.should.equal('Event 1') }) + + it('does not stack listeners on a nested EventSource when the outer one reconnects', function() { + make('
' + + '
' + + '
init
' + + '
' + + '
') + this.clock.tick(1) + + var outer = this.eventSources[0] + var inner = this.eventSources[1] + inner.url.should.equal('/inner') + inner._listeners.e1.should.be.lengthOf(1) + + var messageCount = 0 + byId('d1').addEventListener('htmx:sseMessage', function() { + messageCount++ + }) + + // the outer stream drops and reconnects; the nested one never dropped + outer.simulateConnectionError() + this.clock.tick(500) + this.clock.tick(1) + + inner.readyState.should.equal(EventSource.OPEN) + inner._listeners.e1.should.be.lengthOf(1) + + inner.sendEvent('e1', 'Event 1') + messageCount.should.equal(1) + byId('d1').innerHTML.should.equal('Event 1') + }) })