pubsub.rb

pubsub.rb

class Topic
  attr_reader :name
  def initialize(name)
    @name = name
    @clients = {}
  end
  def publish(message)
    @clients.values.each {|client| client.receive(@name, message)}
  end
  def subscribe(client)
    puts "[client:#{client.name}] subscribe #{@name}"
    @clients[client.name] = client
  end
end
class Client
  attr_reader :name
  def initialize(name)
    @name = name
    @q = Queue.new
    @t = Thread.new do
      while result = @q.pop
        yield(*result)
      end
      p "#{@name} terminated"
    end
  end
  def subscribe(topic)
    topic.subscribe(self)
    self
  end
  def receive(topic_name, message)
    @q << [topic_name, message]
  end
  def close
    @q.push(nil)
    @t.join
    @q.close
  end
end

topic = Topic.new('general')
topics = ['general', 'random', 'free_talk', 'alert', 'activity', 'times'].map {|name| Topic.new(name)}
clients =
  (0...8).map do |i|
  client = Client.new(i) do |topic_name, message|
    puts "##{topic_name} [client:#{i}] #{message}"
  end
  4.times do
    client.subscribe(topics[rand(topics.size)])
  end
  client
end
topics.shuffle.each.with_index do |topic, i|
  topic.publish("message #{i}")
  sleep rand
end
clients.each(&:close)