Extract connection message routing

NEXT_STEPS.md

@@ -20,6 +20,7 @@ Estado al cierre:
   - `arts/sess/engine-session.ts` delega snapshot, generation, dispatch, persistencia y commits en `session-state.ts`.
   - `arts/sess/engine-session.ts` delega la resolucion de revoke local/global/degradado en `session-revoke.ts`.
   - `arts/conn/connection.ts` delega la programacion de reconnect y exhaustion en `connection-reconnect-runtime.ts`.
+  - `arts/conn/connection.ts` delega decode/routing de frames entrantes en `connection-message-router.ts`.
   - `arts/timr/engine-timers.ts` delega fan-out de listeners y diagnostico de listeners en `timer-events.ts`.
 - Integracion total ampliada: `Auth.signOut()` valida anonimizacion, invalidacion de `Permissions` y evento `Cache.invalidate`.
 - Tanda focalizada verde: `npx vitest run src/arts/conn src/libs/cach src/arts/cach src/svrs/cach src/arts/auth src/svrs/auth src/libs/auth src/arts/aapp/test/ecosystem.integration.test.ts` -> 19 archivos, 82 tests.

src/arts/conn/connection-message-router.ts

@@ -0,0 +1,65 @@
+import {
+	CONNECTION_DIAGNOSTIC_EVENTS,
+	CONNECTION_FRAME_TYPE_PONG
+} from './consts.ts';
+import { emitConnectionDiagnostic, type ConnectionDiagnostics } from './diagnostics.ts';
+import { assertConnectionFrame } from './serializer.ts';
+import type { ConnectionAckRegistry } from './acks.ts';
+import type { ConnectionChannelRegistry } from './channel-registry.ts';
+import type { ConnectionEventBus } from './connection-events.ts';
+import type { ConnectionStateTracker } from './connection-state.ts';
+import type { ConnectionHeartbeat } from './heartbeat.ts';
+import type {
+	ConnectionChannelMap,
+	ConnectionFrame,
+	ConnectionMessageMeta,
+	ConnectionSerializer
+} from './types.ts';
+
+export interface RouteConnectionMessageInput<TChannels extends ConnectionChannelMap> {
+	readonly name: string;
+	readonly raw: string | ArrayBuffer;
+	readonly serializer: ConnectionSerializer;
+	readonly state: ConnectionStateTracker;
+	readonly heartbeat: ConnectionHeartbeat;
+	readonly acks: ConnectionAckRegistry;
+	readonly channels: ConnectionChannelRegistry<TChannels>;
+	readonly events: ConnectionEventBus;
+	readonly diagnostics: ConnectionDiagnostics;
+}
+
+export function routeConnectionMessage<TChannels extends ConnectionChannelMap>(
+	input: RouteConnectionMessageInput<TChannels>
+): void {
+	const receivedAt = input.state.touchMessage();
+	input.heartbeat.received();
+	let frame: ConnectionFrame;
+	try {
+		frame = input.serializer.decode(input.raw);
+		assertConnectionFrame(frame);
+	} catch (err) {
+		input.state.setError(err);
+		emitConnectionDiagnostic(input.diagnostics, CONNECTION_DIAGNOSTIC_EVENTS.FRAME_DECODE_FAILED, {
+			error: err
+		});
+		return;
+	}
+
+	if (frame.replyTo !== undefined) {
+		input.acks.resolveFromFrame(frame);
+		return;
+	}
+	if (frame.type === CONNECTION_FRAME_TYPE_PONG) return;
+
+	const meta: ConnectionMessageMeta = {
+		connection: input.name,
+		topic: frame.topic,
+		receivedAt,
+		generation: input.state.generation
+	};
+
+	if (frame.topic !== undefined) {
+		input.channels.receive(frame.topic, frame, meta);
+	}
+	input.events.emitGlobal(frame, meta);
+}

src/arts/conn/connection.ts

@@ -3,6 +3,7 @@ import { createConnectionAckRegistry } from './acks.ts';
 import { wireBrowserReconnect } from './browser-reconnect.ts';
 import { createConnectionChannelRegistry } from './channel-registry.ts';
 import { createConnectionEventBus } from './connection-events.ts';
+import { routeConnectionMessage } from './connection-message-router.ts';
 import { scheduleConnectionReconnect } from './connection-reconnect-runtime.ts';
 import { createConnectionRequestRuntime } from './connection-requests.ts';
 import { createConnectionStateTracker } from './connection-state.ts';
@@ -17,7 +18,6 @@ import {
 	CONNECTION_CONNECT_REASON_AUTH_FAILED,
 	CONNECTION_CONNECT_REASON_DISPOSED,
 	CONNECTION_CONNECT_REASON_TRANSPORT_ERROR,
-	CONNECTION_FRAME_TYPE_PONG,
 	CONNECTION_DIAGNOSTIC_EVENTS,
 	CONNECTION_STATE_CLOSED,
 	CONNECTION_STATE_CLOSING,
@@ -39,7 +39,7 @@ import { createConnectionHeartbeat } from './heartbeat.ts';
 import { createConnectionReconnectPolicy } from './reconnect.ts';
 import { wireConnectionSession } from './session-wiring.ts';
 import { createConnectionSender } from './sender.ts';
-import { assertConnectionFrame, createFrame } from './serializer.ts';
+import { createFrame } from './serializer.ts';
 import { jsonConnectionSerializer } from './serializers/json.ts';
 import { attachConnectionTransport } from './transport-wiring.ts';
 import type { Logger } from '$libs/logr';
@@ -50,8 +50,6 @@ import type {
 	ConnectionChannelOptions,
 	ConnectionCloseEvent,
 	ConnectionConnectResult,
-	ConnectionFrame,
-	ConnectionMessageMeta,
 	ConnectionOptions,
 	ConnectionSessionSource,
 	ConnectionState,
@@ -143,37 +141,17 @@ export function createConnection<TChannels extends ConnectionChannelMap = Connec
 	}
 
 	function handleTransportMessage(raw: string | ArrayBuffer): void {
-		const receivedAt = stateTracker.touchMessage();
-		heartbeat.received();
-		let frame: ConnectionFrame;
-		try {
-			frame = serializer.decode(raw);
-			assertConnectionFrame(frame);
-		} catch (err) {
-			stateTracker.setError(err);
-			emitConnectionDiagnostic(diagnostics, CONNECTION_DIAGNOSTIC_EVENTS.FRAME_DECODE_FAILED, {
-				error: err
-			});
-			return;
-		}
-
-		if (frame.replyTo !== undefined) {
-			acks.resolveFromFrame(frame);
-			return;
-		}
-		if (frame.type === CONNECTION_FRAME_TYPE_PONG) return;
-
-		const meta: ConnectionMessageMeta = {
-			connection: name,
-			topic: frame.topic,
-			receivedAt,
-			generation: stateTracker.generation
-		};
-
-		if (frame.topic !== undefined) {
-			channelRegistry.receive(frame.topic, frame, meta);
-		}
-		events.emitGlobal(frame, meta);
+		routeConnectionMessage({
+			name,
+			raw,
+			serializer,
+			state: stateTracker,
+			heartbeat,
+			acks,
+			channels: channelRegistry,
+			events,
+			diagnostics
+		});
 	}
 
 	function handleTransportClose(event: ConnectionCloseEvent): void {