v0.24.0

Caller-owns-data contract: Writable#send no longer deep-freezes or
binary-coerces message parts, and Connection#receive_message returns
mutable arrays of mutable strings. Non-string parts now NoMethodError
at the wire layer instead of being silently .to_s-coerced; nil parts
no longer become empty binary frames.

Reactor fast path: when a socket was bound/connected from an Async
fiber, #send and #receive skip Reactor.run entirely and call the
engine directly, with Async::Task#with_timeout only when a timeout is
configured. The shared IO thread is used only for non-Async callers.

Inproc PUSH/PULL roughly doubles (+105% to +128% single peer, +63% to
+111% three peers). IPC gains 5-17%. TCP is unchanged -- already
OS/syscall-bound.

Author: paddor

CHANGELOG.md

@@ -1,5 +1,41 @@
 # Changelog
 
+## 0.24.0 — 2026-04-18
+
+### Changed
+
+- **Caller owns message parts.** `Writable#send` no longer deep-freezes or
+  binary-coerces the caller's input. The contract is now libzmq-style:
+  don't mutate parts after sending. `#receive` likewise returns mutable
+  arrays of mutable strings. This removes a full-payload allocation per
+  message (`.b.freeze`) on the send path and a per-frame freeze on the
+  receive path.
+
+- **No more implicit `#to_s` / nil coercion.** Passing a non-string part
+  (e.g. Integer, Symbol, nil) will raise `NoMethodError` at the wire layer
+  instead of being silently converted. The `EMPTY_PART` constant is gone.
+
+- **Reactor fast path for `#send` / `#receive`.** When the socket was
+  bound/connected from an Async fiber, hot-path I/O skips `Reactor.run`
+  entirely and calls the engine directly (with an `Async::Task#with_timeout`
+  wrapper only when a timeout is configured). The shared IO thread is used
+  only when the socket was created from a non-Async thread.
+
+### Performance
+
+Combined effect of caller-owns-data + Reactor fast path on inproc:
+
+- PUSH/PULL inproc 1-peer: **+105% to +128%** msg/s across payload sizes
+- PUSH/PULL inproc 3-peer: **+63% to +111%** msg/s
+- PUSH/PULL ipc: +5% to +17%
+- TCP numbers unchanged (OS/syscall-dominated)
+
+### Removed
+
+- `Writable#freeze_message` and `#frozen_binary` private helpers.
+- `Writable::EMPTY_PART` constant.
+
+
 ## 0.23.1 — 2026-04-18
 
 ### Fixed

lib/omq/client_server.rb

@@ -60,7 +60,7 @@ def initialize(endpoints = nil, linger: Float::INFINITY, backend: nil)
     # @return [self]
     #
     def send_to(routing_id, message)
-      parts = [routing_id.b.freeze, message.b.freeze]
+      parts = [routing_id, message]
       Reactor.run(timeout: @options.write_timeout) { @engine.enqueue_send(parts) }
       self
     end

lib/omq/engine.rb

@@ -100,6 +100,7 @@ def all_peers_gone    = @lifecycle.all_peers_gone
     def parent_task       = @lifecycle.parent_task
     def barrier           = @lifecycle.barrier
     def closed?           = @lifecycle.closed?
+    def on_io_thread?     = @lifecycle.on_io_thread
 
 
     # Enables or disables auto-reconnect for dropped connections.

lib/omq/engine/recv_pump.rb

@@ -94,7 +94,7 @@ def start_with_transform(parent, transform)
 
             while count < FAIRNESS_MESSAGES && bytes < FAIRNESS_BYTES
               msg = conn.receive_message
-              msg = transform.call(msg).freeze
+              msg = transform.call(msg)
 
               # Emit the verbose trace BEFORE enqueueing so the monitor
               # fiber is woken before the application fiber -- the

lib/omq/peer.rb

@@ -38,7 +38,7 @@ def initialize(endpoints = nil, linger: Float::INFINITY, backend: nil)
     # @return [self]
     #
     def send_to(routing_id, message)
-      parts = [routing_id.b.freeze, message.b.freeze]
+      parts = [routing_id, message]
       Reactor.run(timeout: @options.write_timeout) { @engine.enqueue_send(parts) }
       self
     end

lib/omq/radio_dish.rb

@@ -42,7 +42,7 @@ def initialize(endpoints = nil, linger: Float::INFINITY, on_mute: :drop_newest,
     # @return [self]
     #
     def publish(group, body)
-      parts = [group.b.freeze, body.b.freeze]
+      parts = [group, body]
       Reactor.run timeout: @options.write_timeout do
         @engine.enqueue_send(parts)
       end

lib/omq/readable.rb

@@ -14,7 +14,11 @@ module Readable
     # @raise [IO::TimeoutError] if read_timeout exceeded
     #
     def receive
-      Reactor.run timeout: @options.read_timeout do |task|
+      if @engine.on_io_thread?
+        Reactor.run(timeout: @options.read_timeout) { @engine.dequeue_recv }
+      elsif (timeout = @options.read_timeout)
+        Async::Task.current.with_timeout(timeout, IO::TimeoutError) { @engine.dequeue_recv }
+      else
         @engine.dequeue_recv
       end
     end

lib/omq/transport/inproc/pipe.rb

@@ -208,7 +208,7 @@ def close
 
         def apply_transform(parts)
           if @direct_recv_transform
-            @direct_recv_transform.call(parts).freeze
+            @direct_recv_transform.call(parts)
           else
             parts
           end

lib/omq/transport/udp.rb

@@ -200,7 +200,7 @@ def receive_message
             next unless parts
             group, body = parts
             next unless @groups.include?(group.b)
-            return [group.b.freeze, body.b.freeze]
+            return [group, body]
           rescue IO::WaitReadable
             @socket.wait_readable
             retry

lib/omq/version.rb

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