Contents

require 'set'
require 'concurrent/actor/context'

module Concurrent
  module Actor
    module Utils

      # Allows to build pub/sub easily.
      # @example news
      #   news_channel = Concurrent::Actor::Utils::Broadcast.spawn :news
      #
      #   2.times do |i|
      #     Concurrent::Actor::Utils::AdHoc.spawn "listener-#{i}" do
      #       news_channel << :subscribe
      #       -> message { puts message }
      #     end
      #   end
      #
      #   news_channel << 'Ruby rocks!'
      #   # prints: 'Ruby rocks!' twice
      class Broadcast < RestartingContext

        def initialize
          @receivers = Set.new
        end

        def on_message(message)
          case message
          when :subscribe
            if envelope.sender.is_a? Reference
              @receivers.add envelope.sender
              true
            else
              false
            end
          when :unsubscribe
            !!@receivers.delete(envelope.sender)
          when :subscribed?
            @receivers.include? envelope.sender
          else
            filtered_receivers.each { |r| r << message }
          end
        end

        # override to define different behaviour, filtering etc
        def filtered_receivers
          @receivers
        end
      end

    end
  end
end

Version data entries

3 entries across 3 versions & 1 rubygems

Version	Path
concurrent-ruby-edge-0.7.2	lib/concurrent-ruby-edge/concurrent/actor/utils/broadcast.rb
concurrent-ruby-edge-0.7.1	lib/concurrent-ruby-edge/concurrent/actor/utils/broadcast.rb
concurrent-ruby-edge-0.7.0	lib/concurrent-ruby-edge/concurrent/actor/utils/broadcast.rb