fix(channel): flip S1 S4 S5 S7 S8 and fix S9 drain

changelog.d/channel-hardener-s1-s8.changed.md

@@ -0,0 +1,5 @@
+- Channel `publish()` throws `Wheels.Channel.PublishFailed` when the database INSERT fails instead of returning `persisted:false`
+- `$getChannelEngine()` throws `Wheels.Channel.UnknownAdapter` for a name that is not `memory` or `database`
+- The memory channel engine replays retained events after `lastEventId` on subscribe
+- `channelSSETag(events=)` registers `addEventListener` for each named event so they are not dropped by `onmessage`
+- Empty and whitespace-only channel names throw `Wheels.Channel.InvalidName`

changelog.d/channel-hardener-s9.fixed.md

@@ -0,0 +1 @@
+- Memory channel drain no longer `clear()`s the live buffer, so events published during the drain loop are not dropped

vendor/wheels/Channel.cfc

@@ -26,9 +26,17 @@ component {
 	public Channel function init() {
 		// channel -> ConcurrentHashMap of subscriberId -> {callback, createdAt}
 		variables.channels = CreateObject("java", "java.util.concurrent.ConcurrentHashMap").init();
+		variables.eventLog = CreateObject("java", "java.util.concurrent.ConcurrentHashMap").init();
+		variables.maxEventLogSize = 100;
 		return this;
 	}
 
+	public void function $assertChannelName(required string channel) {
+		if (!Len(Trim(arguments.channel))) {
+			throw(type = "Wheels.Channel.InvalidName", message = "Channel name cannot be empty.");
+		}
+	}
+
 	/**
 	 * Subscribe to a channel with a callback function.
 	 *
@@ -42,6 +50,7 @@ component {
 		required any callback,
 		string id = CreateUUID()
 	) {
+		$assertChannelName(arguments.channel);
 		// Ensure channel map exists (putIfAbsent is atomic)
 		variables.channels.putIfAbsent(
 			arguments.channel,
@@ -73,6 +82,7 @@ component {
 		required string data,
 		string id = CreateUUID()
 	) {
+		$assertChannelName(arguments.channel);
 		local.timestamp = Now();
 		local.eventPayload = {
 			id: arguments.id,
@@ -81,6 +91,7 @@ component {
 			data: arguments.data,
 			timestamp: local.timestamp
 		};
+		$appendEventLog(arguments.channel, local.eventPayload);
 
 		local.subscriberCount = 0;
 		local.subscribers = variables.channels.get(arguments.channel);
@@ -176,6 +187,46 @@ component {
 	 */
 	public void function removeChannel(required string channel) {
 		variables.channels.remove(arguments.channel);
+		variables.eventLog.remove(arguments.channel);
+	}
+
+	/**
+	 * Return retained events on a channel after lastEventId.
+	 * If lastEventId is not in the retained window, return the retained tail.
+	 */
+	public array function replay(required string channel, required string lastEventId) {
+		$assertChannelName(arguments.channel);
+		local.out = [];
+		local.log = variables.eventLog.get(arguments.channel);
+		if (IsNull(local.log)) {
+			return local.out;
+		}
+		local.snapshot = local.log.toArray();
+		local.seen = false;
+		for (local.evt in local.snapshot) {
+			if (local.seen) {
+				ArrayAppend(local.out, local.evt);
+			}
+			if (local.evt.id == arguments.lastEventId) {
+				local.seen = true;
+			}
+		}
+		if (!local.seen) {
+			return local.snapshot;
+		}
+		return local.out;
+	}
+
+	private void function $appendEventLog(required string channel, required struct eventPayload) {
+		variables.eventLog.putIfAbsent(
+			arguments.channel,
+			CreateObject("java", "java.util.concurrent.ConcurrentLinkedQueue").init()
+		);
+		local.log = variables.eventLog.get(arguments.channel);
+		local.log.offer(arguments.eventPayload);
+		while (local.log.size() > variables.maxEventLogSize) {
+			local.log.poll();
+		}
 	}
 
 }

vendor/wheels/channel/DatabaseAdapter.cfc

@@ -32,6 +32,12 @@ component {
 		return this;
 	}
 
+	public void function $assertChannelName(required string channel) {
+		if (!Len(Trim(arguments.channel))) {
+			throw(type = "Wheels.Channel.InvalidName", message = "Channel name cannot be empty.");
+		}
+	}
+
 	/**
 	 * Publish an event to the database.
 	 *
@@ -47,6 +53,7 @@ component {
 		required string data,
 		string id = CreateUUID()
 	) {
+		$assertChannelName(arguments.channel);
 		$ensureEventsTable();
 		$maybeCleanup();
 
@@ -76,12 +83,10 @@ component {
 				type="error",
 				file="wheels_channels"
 			);
-			return {
-				id: arguments.id,
-				channel: arguments.channel,
-				event: arguments.event,
-				persisted: false
-			};
+			throw(
+				type = "Wheels.Channel.PublishFailed",
+				message = "Failed to persist channel event on [#arguments.channel#]: #e.message#"
+			);
 		}
 	}
 
@@ -98,6 +103,7 @@ component {
 		string lastEventId = "",
 		date since = DateAdd("n", -5, Now())
 	) {
+		$assertChannelName(arguments.channel);
 		$ensureEventsTable();
 
 		// If lastEventId is provided, find its timestamp and get events at or after it,

vendor/wheels/controller/channels.cfc

@@ -41,6 +41,7 @@ component {
 		numeric timeout = 300,
 		numeric heartbeatInterval = 15
 	) {
+		$assertChannelName(arguments.channel);
 		// Auto-detect Last-Event-ID from request header
 		if (!Len(arguments.lastEventId)) {
 			try {
@@ -107,6 +108,7 @@ component {
 		string action = "stream",
 		string events = ""
 	) {
+		$assertChannelName(arguments.channel);
 		// Build URL
 		if (Len(arguments.route)) {
 			local.url = urlFor(route = arguments.route);
@@ -125,16 +127,76 @@ component {
 			local.url &= "&events=" & EncodeForURL(arguments.events);
 		}
 
+		local.listeners = "src.onmessage = relay;";
+		if (Len(arguments.events)) {
+			for (local.evtName in ListToArray(arguments.events)) {
+				local.trimmed = Trim(local.evtName);
+				if (Len(local.trimmed) && CompareNoCase(local.trimmed, "message")) {
+					local.listeners &= Chr(10) & "	src.addEventListener('#JSStringFormat(local.trimmed)#', relay);";
+				}
+			}
+		}
+
 		return "<script>
 (function(){
 	var src = new EventSource('#JSStringFormat(local.url)#');
-	src.onmessage = function(e) {
+	function relay(e) {
 		document.dispatchEvent(new CustomEvent('wheels:sse', {detail: {data: e.data, event: e.type, id: e.lastEventId}}));
-	};
+	}
+	#local.listeners#
 })();
 </script>";
 	}
 
+	public boolean function $isChannelBufferItem(required any item) {
+		if (IsStruct(arguments.item)) {
+			return true;
+		}
+		if (IsSimpleValue(arguments.item) && !IsBoolean(arguments.item)) {
+			return true;
+		}
+		return false;
+	}
+
+	public void function $assertChannelName(required string channel) {
+		if (!Len(Trim(arguments.channel))) {
+			throw(type = "Wheels.Channel.InvalidName", message = "Channel name cannot be empty.");
+		}
+	}
+
+	public array function $drainChannelBuffer(required any buffer) {
+		local.events = [];
+		try {
+			local.next = arguments.buffer.poll();
+			while (!IsNull(local.next)) {
+				if (!$isChannelBufferItem(local.next)) {
+					break;
+				}
+				ArrayAppend(local.events, local.next);
+				local.next = arguments.buffer.poll();
+			}
+			if (ArrayLen(local.events)) {
+				return local.events;
+			}
+		} catch (any e) {
+		}
+		while (true) {
+			if (!arguments.buffer.size()) {
+				break;
+			}
+			try {
+				local.item = arguments.buffer.remove(JavaCast("int", 0));
+			} catch (any drainError) {
+				break;
+			}
+			if (!$isChannelBufferItem(local.item)) {
+				break;
+			}
+			ArrayAppend(local.events, local.item);
+		}
+		return local.events;
+	}
+
 	/**
 	 * Internal: Memory-adapter subscription loop.
 	 * Subscribes to the Channel singleton, buffers events in a synchronized
@@ -150,10 +212,7 @@ component {
 		local.writer = initSSEStream();
 		local.engine = $getChannelEngine("memory");
 
-		// Thread-safe event buffer using a synchronized list
-		local.buffer = CreateObject("java", "java.util.Collections").synchronizedList(
-			CreateObject("java", "java.util.ArrayList").init()
-		);
+		local.buffer = CreateObject("java", "java.util.concurrent.ConcurrentLinkedQueue").init();
 
 		// Subscribe with a callback that buffers events
 		local.subscriberId = local.engine.subscribe(
@@ -163,10 +222,20 @@ component {
 				if (ArrayLen(eventFilter) && !ArrayFind(eventFilter, event.event)) {
 					return;
 				}
-				buffer.add(event);
+				buffer.offer(event);
 			}
 		);
 
+		if (Len(arguments.lastEventId)) {
+			local.replayed = local.engine.replay(channel = arguments.channel, lastEventId = arguments.lastEventId);
+			for (local.replayEvt in local.replayed) {
+				if (ArrayLen(arguments.eventFilter) && !ArrayFind(arguments.eventFilter, local.replayEvt.event)) {
+					continue;
+				}
+				local.buffer.offer(local.replayEvt);
+			}
+		}
+
 		try {
 			local.startTime = GetTickCount() / 1000;
 			local.lastHeartbeat = local.startTime;
@@ -179,15 +248,8 @@ component {
 				}
 
 				// Drain buffer and send events
-				local.size = local.buffer.size();
-				if (local.size > 0) {
-					// Snapshot and clear
-					local.events = [];
-					for (local.i = 1; local.i <= local.size; local.i++) {
-						ArrayAppend(local.events, local.buffer.get(local.i - 1));
-					}
-					local.buffer.clear();
-
+				local.events = $drainChannelBuffer(local.buffer);
+				if (ArrayLen(local.events)) {
 					for (local.evt in local.events) {
 						sendSSEEvent(
 							writer = local.writer,

vendor/wheels/global/routing.cfm

@@ -71,15 +71,21 @@
 			return application.wheels.channelDatabaseEngine;
 		}
 
-		// Default: memory adapter
-		if (!StructKeyExists(application, "wheels") || !StructKeyExists(application.wheels, "channelEngine")) {
-			lock name="wheelsChannelEngine" timeout="10" {
-				if (!StructKeyExists(application, "wheels") || !StructKeyExists(application.wheels, "channelEngine")) {
-					application.wheels.channelEngine = CreateObject("component", "wheels.Channel").init();
+		if (local.adapterType == "memory") {
+			if (!StructKeyExists(application, "wheels") || !StructKeyExists(application.wheels, "channelEngine")) {
+				lock name="wheelsChannelEngine" timeout="10" {
+					if (!StructKeyExists(application, "wheels") || !StructKeyExists(application.wheels, "channelEngine")) {
+						application.wheels.channelEngine = CreateObject("component", "wheels.Channel").init();
+					}
 				}
 			}
+			return application.wheels.channelEngine;
 		}
-		return application.wheels.channelEngine;
+
+		throw(
+			type = "Wheels.Channel.UnknownAdapter",
+			message = "Unknown channel adapter [#local.adapterType#]. Use memory or database."
+		);
 	}
 
 

vendor/wheels/tests/_assets/channel/BrokenDatasourceAdapter.cfc

@@ -0,0 +1,10 @@
+component extends="wheels.channel.DatabaseAdapter" {
+
+	public BrokenDatasourceAdapter function init(boolean tableVerified = true) {
+		super.init();
+		variables.$datasource = "wheels-channel-hardener-missing-ds";
+		variables.tableVerified = arguments.tableVerified;
+		return this;
+	}
+
+}

vendor/wheels/tests/_assets/channel/MidLoopPublishBuffer.cfc

@@ -0,0 +1,43 @@
+component {
+
+	public MidLoopPublishBuffer function init() {
+		variables.items = ["a", "b"];
+		variables.published = false;
+		variables.cleared = false;
+		return this;
+	}
+
+	public numeric function size() {
+		return ArrayLen(variables.items);
+	}
+
+	public any function get(required numeric idx) {
+		local.value = variables.items[arguments.idx + 1];
+		$publishMidLoop();
+		return local.value;
+	}
+
+	public any function remove(required numeric idx) {
+		local.value = variables.items[arguments.idx + 1];
+		ArrayDeleteAt(variables.items, arguments.idx + 1);
+		$publishMidLoop();
+		return local.value;
+	}
+
+	public void function clear() {
+		variables.cleared = true;
+		variables.items = [];
+	}
+
+	public boolean function wasCleared() {
+		return variables.cleared;
+	}
+
+	private void function $publishMidLoop() {
+		if (!variables.published) {
+			variables.published = true;
+			ArrayAppend(variables.items, "c");
+		}
+	}
+
+}

vendor/wheels/tests/_assets/channel/SseWriterFake.cfc

@@ -0,0 +1,25 @@
+component {
+
+	public SseWriterFake function init() {
+		variables.chunks = [];
+		variables.checks = 0;
+		return this;
+	}
+
+	public void function write(required string text) {
+		ArrayAppend(variables.chunks, arguments.text);
+	}
+
+	public void function flush() {
+	}
+
+	public boolean function checkError() {
+		variables.checks = variables.checks + 1;
+		return variables.checks > 1;
+	}
+
+	public array function chunks() {
+		return variables.chunks;
+	}
+
+}