[pgmq-framework] Implement Listener class with LISTEN/NOTIFY support

[mensfeld]

Summary

Implement a full Listener class for the planned pgmq-framework gem that uses PostgreSQL's LISTEN/NOTIFY mechanism for push-based message arrival detection.

Context

The low-level primitives are now available in pgmq-ruby:

• enable_notify_insert(queue_name, throttle_interval_ms:) - enables notifications
• disable_notify_insert(queue_name) - disables notifications

These wrap the PGMQ extension's notification throttling feature (PR #445).

Proposed Implementation

Listener Class

listener = PGMQ::Framework::Listener.new(client, queues: ['orders', 'notifications'])

listener.on_message do |queue_name|
  messages = client.read_batch(queue_name, vt: 30, qty: 10)
  messages.each { |msg| process(msg) }
end

listener.start  # Blocks, listening for notifications

Features to Implement

1. LISTEN Management

   • Automatic LISTEN on configured queues
   • Handle multiple queues with a single connection

2. Notification Handling

   • Use PG::Connection#wait_for_notify for blocking mode
   • Support async mode with set_notice_receiver for non-blocking

3. Connection Management

   • Dedicated connection for LISTEN (can't share with pooled connections)
   • Automatic reconnection on connection loss
   • Graceful shutdown

4. Concurrency Support

   • Thread-based listener
   • Fiber/async compatible (Ruby 3.0+ Fiber Scheduler)
   • Ractor compatibility (future)

5. Integration

   • Callback-based API
   • Block-based API
   • Integration with framework's worker system

Technical Considerations

• wait_for_notify blocks the connection, requiring a dedicated connection
• PostgreSQL NOTIFY payloads are limited to 8000 bytes (we only need queue name)
• The pg gem's wait_for_notify returns [channel, pid, payload]
• Need to handle the case where notifications are missed during reconnection

References

• PGMQ PR #445 - Notification throttling
• pg gem - wait_for_notify
• PostgreSQL LISTEN/NOTIFY

---

This issue tracks a feature for the planned pgmq-framework gem, not pgmq-ruby.