Migrate from EventMachine to Async by henrikbjorn · Pull Request #2 · relatel/librevox

henrikbjorn · 2026-02-24T09:15:34Z

Why

Librevox was built on EventMachine, which is effectively unmaintained and incompatible with Ruby 4.0. This PR replaces the entire runtime with Ruby's fiber-based async ecosystem, making librevox work on modern Ruby while simplifying the concurrency model.

What changed

Runtime: EventMachine → Async

The core dependency shifts from eventmachine to three gems:

async — fiber-based concurrency (replaces EM reactor loop)
io-endpoint — TCP server/client (replaces EM.start_server/EM.connect)
io-stream — buffered streaming (replaces EM::Connection#receive_data)

The custom Base.new allocator that mimicked EM::Connection is replaced with plain initialize methods. Signal handling uses Async::Barrier for clean shutdown instead of EM.stop.

Architecture: Extract connection management from listeners

Previously, listeners mixed protocol I/O with session logic. Now:

Protocol::Connection — wraps the socket stream, handles read/write framing
Server — accepts inbound connections (outbound listener)
Client — manages connection + reconnect loop (inbound listener)
Listeners — pure session logic, receive a connection via constructor

Concurrency: Callbacks → Queues

The old callback model (application("playback", "file") { |r| ... }) is replaced with blocking calls that read naturally top-to-bottom:

# Before (callback)
application("playback", "sounds/welcome.wav") do
  application("hangup")
end

# After (blocking)
application "playback", "sounds/welcome.wav"
application "hangup"

Under the hood, each connection runs two fibers:

Read loop fiber — reads ESL frames, routes responses to queues, fires event hooks
Session fiber — runs session_initiated / user logic, blocks on queues

Three synchronization primitives coordinate them:

@reply_queue (Async::Queue) — command/reply responses from FreeSWITCH
@app_complete_queue (Async::Queue) — CHANNEL_EXECUTE_COMPLETE events
@command_mutex (Async::Semaphore(1)) — serializes ESL command sends

The mutex is held only for send + reply-ack, then released before waiting on @app_complete_queue. This allows command() calls from other fibers (e.g. event hook pollers calling api uuid_break) to proceed while an application() call waits for execution to complete.

Disconnect handling

When FreeSWITCH drops the connection, the read loop exits. The session task may be blocked on a queue dequeue with nothing to wake it. session_task.stop in ensure terminates it cleanly — no session_task.wait which would hang forever.

Tests: Bacon → Minitest

Migrated from Bacon (unmaintained) to Minitest
Split monolithic spec files into one test class per file
Tests use a MockConnection that captures send_data calls and drives receive_data directly
CI matrix updated to Ruby 3.3, 3.4, 4.0

Other

event-lock: true added to all sendmsg calls (prevents FreeSWITCH from executing apps out of order)
New park and multiset applications
update_session fixed to actually apply the uuid_dump response
Broken pipe on connection close handled gracefully

Replace the unmaintained Bacon test framework with Minitest::Test. Rename spec/ to test/ following Ruby conventions (*_test.rb suffix). Use prepend-based test mocking instead of class reopening to avoid method redefinition warnings. Drop RR and Bacon gem dependencies. Also: rewrite README with table of contents, add GitHub Actions CI workflow (Ruby 3.1–3.4), migrate from EventMachine to async I/O, and fix Ruby 4.0 frozen string warnings in Response.

- Replace ListenerTestMock send_data/initialize overrides with a MockConnection object passed to .new, so the real send_data writes to the mock connection instead of being monkey-patched - Split test files to one test class per file - Move connection logic into Inbound.start/Outbound.start class methods - Standardize on parentheses in all method definitions - Update examples to match current API

Remove Base.self.new/allocate/post_init pattern inherited from EventMachine. Each class now uses a standard initialize that accepts connection as the first argument.

async-io is deprecated. Replace with io-endpoint and io-stream which provide the same API. Remove the Librevox::Connection wrapper class since IO::Stream supports write(flush: true), read_partial, and close directly. Add logger as a gemspec dependency since it was removed from Ruby's stdlib.

Async already handles SIGINT/SIGTERM by cleanly stopping child tasks and raising Interrupt. Calling task.stop from a trap context crashes because async uses mutexes internally.

Docker Compose and other process managers send SIGTERM (not SIGINT) when stopping containers. Catch SignalException alongside Interrupt for graceful shutdown in all environments.

Add park helper to Applications. Update README dialplan section to show both calling styles: sequential (without blocks) and nested (with blocks for completion callbacks).

Without event-lock, FreeSWITCH returns the command reply immediately rather than waiting for the application to finish executing. This caused callbacks for bridge, playback, record etc. to fire prematurely.

require 'io/endpoint' does not load the tcp method; needs require 'io/endpoint/host_endpoint' instead.

@task

Replace stored @task module state with a local Async::Barrier, giving proper group lifecycle (wait/stop) for multi-listener setups. Introduce Runner class to keep the block form of Librevox.start working without shared state. Move endpoint creation outside the inbound reconnect loop.

Separate wire-protocol parsing from listener business logic by introducing Protocol::Connection (wraps IO::Stream with read_message/ write/close), Server (accepts outbound connections), and Client (connects inbound with reconnect). Listeners become pure session logic with thin self.start methods that delegate to Server/Client. Also unifies CommandSocket to use Protocol::Connection, extracts Runner to its own file, and switches Librevox.start to kwargs.

endpoint.accept passes (socket, address) — we were only accepting one argument, causing ArgumentError in production.

… close update_session now explicitly sets @session from the uuid_dump response before calling the user's block, so variable() sees fresh channel data after slow applications like bridge or play_and_get_digits. Connection#close guards against double-close and rescues EPIPE/ECONNRESET when the remote end has already hung up.

…and/reply The application queue now waits for the application to actually finish on the channel before firing callbacks. This fixes the issue where all application callbacks would fire immediately on the sendmsg ack, before any application had executed. Session is updated directly from the CHANNEL_EXECUTE_COMPLETE event content, which carries all channel variables — eliminating the uuid_dump roundtrip after each application. Setup commands (connect/myevents/linger) use a separate reply queue that still advances on command/reply.

…tive command() is the universal one-response primitive (command/reply or api/response). sendmsg() handles two responses (ack + EXECUTE_COMPLETE). Eliminates @reply_queue and send_command helper.

Replace array-based callback queues with Async::Queue instances, enabling flat sequential code in session_initiated instead of nested callbacks. Two fibers per connection: a read fiber (read_loop) dispatches messages to queues, a session fiber (run_session) issues commands and blocks until replies arrive. Async::Semaphore(1) serializes command access so event hook tasks can't cause reply misdelivery. - Remove all &block params from commands and applications modules - Extract run_session for both inbound (auth/events/filters) and outbound (connect/myevents/linger/session_initiated) - Server and Client spawn session task alongside read_loop - Wrap tests in Sync via AsyncTest module, yield_to_fibers for cooperative scheduling in test helpers

Remove session_task.wait from client/server — when the connection drops, the session task may be blocked on a queue dequeue forever. The ensure block's session_task.stop already handles both finished and blocked tasks. Release command_mutex before waiting on app_complete_queue in sendmsg, allowing other fibers to call command() while an application is blocked waiting for CHANNEL_EXECUTE_COMPLETE.

henrikbjorn added 26 commits February 10, 2026 09:57

Update CI matrix to non-EOL Ruby versions (3.3, 3.4, 4.0)

fa4d3d0

Add note about calling application directly for unwrapped apps

a9f76f2

Replace custom allocator with plain initialize methods

eff63a7

Remove Base.self.new/allocate/post_init pattern inherited from EventMachine. Each class now uses a standard initialize that accepts connection as the first argument.

Fix signal handling crash by removing trap handlers

fed1ae1

Async already handles SIGINT/SIGTERM by cleanly stopping child tasks and raising Interrupt. Calling task.stop from a trap context crashes because async uses mutexes internally.

Handle SIGTERM for clean shutdown in Docker/systemd/k8s

e03dba4

Docker Compose and other process managers send SIGTERM (not SIGINT) when stopping containers. Catch SignalException alongside Interrupt for graceful shutdown in all environments.

Add park application, document block vs sequential style in README

7bf0e08

Add park helper to Applications. Update README dialplan section to show both calling styles: sequential (without blocks) and nested (with blocks for completion callbacks).

Add event-lock: true to application sendmsg

ca11cad

Without event-lock, FreeSWITCH returns the command reply immediately rather than waiting for the application to finish executing. This caused callbacks for bridge, playback, record etc. to fire prematurely.

Fix missing require for IO::Endpoint.tcp

16baa4c

require 'io/endpoint' does not load the tcp method; needs require 'io/endpoint/host_endpoint' instead.

Use block form of endpoint.connect for automatic socket cleanup

a6024c5

Fix Server#accept to match io-endpoint callback signature

a9e5781

endpoint.accept passes (socket, address) — we were only accepting one argument, causing ArgumentError in production.

Prefix unused address parameter with underscore

60a1a0b

Use modern hash syntax for options

a490bb1

Add multiset application for setting multiple channel variables

10f418c

Require callback style in README, update application queue docs

134cc42

Note that set/multiset don't need callbacks in README

9dc8651

Extract send_command helper, test reply queue ordering

2523a98

Unify command and reply queues, extract sendmsg as two-response primi…

5e217c6

…tive command() is the universal one-response primitive (command/reply or api/response). sendmsg() handles two responses (ack + EXECUTE_COMPLETE). Eliminates @reply_queue and send_command helper.

henrikbjorn self-assigned this Feb 24, 2026

henrikbjorn merged commit ef22f80 into master Feb 24, 2026
3 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Migrate from EventMachine to Async#2

Migrate from EventMachine to Async#2
henrikbjorn merged 26 commits intomasterfrom
async-queues

henrikbjorn commented Feb 24, 2026 •

edited

Loading

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Conversation

henrikbjorn commented Feb 24, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Why

What changed

Runtime: EventMachine → Async

Architecture: Extract connection management from listeners

Concurrency: Callbacks → Queues

Disconnect handling

Tests: Bacon → Minitest

Other

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

henrikbjorn commented Feb 24, 2026 •

edited

Loading