v0.27.0 - transport-supplied ZMTP Connection class

Transport modules may define .connection_class to substitute their
own Protocol::ZMTP::Connection-shaped class. ConnectionLifecycle
queries it (with a respond_to? fallback) so existing transports —
built-in or third-party — keep working unchanged.

Enables plugin transports whose wire shape differs from ZMTP/3.1
(e.g. omq-websocket per RFC 45) to plug in without forking the
engine.

Author: paddor

CHANGELOG.md

@@ -1,5 +1,18 @@
 # Changelog
 
+## 0.27.0 — 2026-04-20
+
+### Added
+
+- **Transport-supplied ZMTP Connection class.** Transport modules may
+  now define `.connection_class` to substitute their own
+  `Protocol::ZMTP::Connection`-shaped class. `ConnectionLifecycle`
+  reads it (with a `respond_to?` fallback to
+  `Protocol::ZMTP::Connection`) so existing transports — built-in or
+  third-party — keep working unchanged. Enables plugin transports
+  whose wire shape differs from ZMTP/3.1 (e.g. ZeroMQ-over-WebSocket
+  per RFC 45) to plug in without forking the engine.
+
 ## 0.26.2 — 2026-04-20
 
 ### Fixed

lib/omq/engine.rb

@@ -635,7 +635,8 @@ def transport_for(endpoint)
     def spawn_connection(io, as_server:, endpoint: nil)
       @lifecycle.barrier&.async(transient: true, annotation: "conn #{endpoint}") do
         done      = Async::Promise.new
-        lifecycle = ConnectionLifecycle.new(self, endpoint: endpoint, done: done)
+        transport = endpoint ? transport_for(endpoint) : nil
+        lifecycle = ConnectionLifecycle.new(self, endpoint: endpoint, done: done, transport: transport)
         lifecycle.handshake!(io, as_server: as_server)
         done.wait
       rescue Async::Stop, Async::Cancel

lib/omq/engine/connection_lifecycle.rb

@@ -66,13 +66,19 @@ class InvalidTransition < RuntimeError
       # @param engine [Engine]
       # @param endpoint [String, nil]
       # @param done [Async::Promise, nil] resolved when connection is lost
+      # @param transport [Module, nil] transport module that produced +io+;
+      #   queried for {.connection_class} so plugins (e.g. WebSocket) can
+      #   substitute their own ZMTP-shaped connection class. Falls back to
+      #   {Protocol::ZMTP::Connection} when nil or when the transport
+      #   doesn't define +connection_class+.
       #
-      def initialize(engine, endpoint: nil, done: nil)
-        @engine   = engine
-        @endpoint = endpoint
-        @done     = done
-        @state    = :new
-        @conn     = nil
+      def initialize(engine, endpoint: nil, done: nil, transport: nil)
+        @engine    = engine
+        @endpoint  = endpoint
+        @done      = done
+        @transport = transport
+        @state     = :new
+        @conn      = nil
 
         # Nest the per-connection barrier under the socket-level barrier
         # so every pump spawned via +@barrier.async+ is also tracked by
@@ -90,7 +96,8 @@ def initialize(engine, endpoint: nil, done: nil)
       #
       def handshake!(io, as_server:)
         transition!(:handshaking)
-        conn = Protocol::ZMTP::Connection.new io,
+        conn_class = @transport.respond_to?(:connection_class) ? @transport.connection_class : Protocol::ZMTP::Connection
+        conn = conn_class.new io,
           socket_type:      @engine.socket_type.to_s,
           identity:         @engine.options.identity,
           as_server:        as_server,

lib/omq/transport/ipc.rb

@@ -15,6 +15,15 @@ module IPC
 
 
       class << self
+        # ZMTP connection class used for IPC-accepted/dialed peers.
+        #
+        # @return [Class]
+        #
+        def connection_class
+          Protocol::ZMTP::Connection
+        end
+
+
         # Creates a bound IPC listener.
         #
         # @param endpoint [String] e.g. "ipc:///tmp/my.sock" or "ipc://@abstract"

lib/omq/transport/tcp.rb

@@ -13,6 +13,15 @@ module TCP
 
 
       class << self
+        # ZMTP connection class used for TCP-accepted/dialed peers.
+        #
+        # @return [Class]
+        #
+        def connection_class
+          Protocol::ZMTP::Connection
+        end
+
+
         # Creates a bound TCP listener.
         #
         # @param endpoint [String] e.g. "tcp://127.0.0.1:5555" or "tcp://*:0"

lib/omq/version.rb

@@ -1,5 +1,5 @@
 # frozen_string_literal: true
 
 module OMQ
-  VERSION = "0.26.2"
+  VERSION = "0.27.0"
 end

test/omq/transport/tcp_test.rb

@@ -3,6 +3,11 @@
 require_relative "../../test_helper"
 
 describe "TCP transport" do
+  it "advertises Protocol::ZMTP::Connection as its connection_class" do
+    assert_equal Protocol::ZMTP::Connection, OMQ::Transport::TCP.connection_class
+  end
+
+
   it "PAIR over TCP with ephemeral port" do
     Async do
       server = OMQ::PAIR.new