Pub/Sub with Queue and a write-pump Thread

Before the code: messaging and replication, in plain English

Picture a group chat. One person publishes a message. The chat app knows who is in the room and forwards the message to each of them. That is publish and subscribe, and it is the simplest piece of Redis that is not a key-value store. Replication is the same idea pointed at a backup machine: the leader sends every write to a follower, who quietly replays them. Same plumbing, different audience. In Ruby we wire this with one Queue per subscriber and a small pump Thread that drains each Queue to the socket.

Fan-out and follow-the-leader

A publisher pushes once. Each subscriber has its own Queue. A pump Thread drains it to the socket.

Quiz: Quiz

Loading practice…

Queue is Ruby's thread-safe MPMC channel. Publishers push, the pump pops and writes. Slow subscribers stay isolated; publishers never block on a slow socket.

07-pubsub/server.rb
ruby
class Subscriber
  attr_reader :queue, :subs
  def initialize(conn)
    @conn = conn
    @queue = Queue.new
    @subs = Set.new
  end
  def deliver(payload); @queue.push(payload); end
  def pump
    Thread.new(@conn) do |conn|
      while (msg = @queue.pop)
        break if msg == :stop
        begin; conn.write(msg); rescue IOError; break; end
      end
    end
  end
  def stop; @queue.push(:stop); end
end

Subscriber class wraps the Queue and the pump Thread.

07-pubsub/server.rb
ruby
when 'PUBLISH'
  return encode_error("ERR PUBLISH needs channel and message") if rest.size != 2
  subs = $ch_mu.synchronize { $channels[rest[0]].dup }
  subs.each { |s| s.deliver(encode_message(rest[0], rest[1])) }
  encode_int(subs.size)

PUBLISH grabs the subscriber set under $ch_mu, then delivers outside the lock.

Why a pump Thread? Because a slow subscriber's TCP send buffer fills up and conn.write blocks. Without the pump, one slow subscriber blocks the publisher, starving every other subscriber. The pump decouples them.

Quiz: Quiz

Loading practice…