Class: Farce::Abstract::Publisher Abstract

Inherits:
Object
  • Object
show all
Includes:
Internal::Noncopyable, Internal::Freeze::Unfreezable
Defined in:
lib/farce/abstract/publisher.rb

Overview

This class is abstract.

Shared superclass for all publisher implementations.

Implements a publisher-subscriber (pub-sub) pattern.

See Also:

Instance Method Summary collapse

Methods included from Internal::Noncopyable

#duplicable?

Constructor Details

#initialize(*subscribers) ⇒ Publisher

Returns a new instance of Publisher.

Parameters:

  • subscribers (Array) —

    the initial subscribers to add to the publisher, see #subscribe



17
18
19
20
21
 
# File 'lib/farce/abstract/publisher.rb', line 17

def initialize(*subscribers)
  @closed = Farce::Flag.new
  super()
  subscribers.each { subscribe(it) }
end

Instance Method Details

#closed? ⇒ Boolean

Returns whether the publisher is closed.

Returns:

  • (Boolean) —

    whether the publisher is closed.



24
 
# File 'lib/farce/abstract/publisher.rb', line 24

def closed? = @closed.value

#publish(payload = nil) ⇒ self Also known as: send, <<, broadcast

Publishes the given payload to all current subscribers.

Full bounded queues drop new messages. Failed publishing attempts to closed or frozen receivers cancel their subscriptions.

Parameters:

  • payload (BasicObject) (defaults to: nil) —

    the payload to publish

Returns:

  • (self) —

    the publisher itself



33
34
35
36
37
 
# File 'lib/farce/abstract/publisher.rb', line 33

def publish(payload = nil)
  raise "Cannot publish to a closed publisher" if closed?
  map.each_value { it.publish(payload) }
  self
end

#subscribe(subscriber) ⇒ Subscription #subscribe(*args, mode: nil, scheduler: nil, capacity: nil) {|*args, payload| ... } ⇒ Subscription Also known as: monitor

Returns the created subscription.

Overloads:

  • #subscribe(subscriber) ⇒ Subscription

    Adds a subscriber to the publisher. Register additional subscriber classes with Subscription.define. Publisher subscribers forward messages synchronously to their own subscribers. Avoid subscribing publishers in a cycle.

    Examples:

    Forward messages to another publisher

    upstream.subscribe(downstream)
upstream.publish(:message) # also publishes to downstream's subscribers

    Parameters:

    • subscriber (BasicObject) —

      The subscriber to add. Standard-library queues are supported by unshared and local publishers and retain the original payload. A full Thread::SizedQueue drops new messages. Publishing to a closed queue cancels its subscription.

      Calls one of the following methods:

      The concurrent-ruby integration supports these subscribers with unshared and local publishers:

      • Concurrent::AtomicReference#value=
      • Concurrent::AtomicFixnum#increment (without arguments)
      • Concurrent::AtomicBoolean#make_true
      • Concurrent::Event#set (without arguments)
      • Concurrent::CountDownLatch#count_down (without arguments)

      Set the default mode, auto_local, capacity, and similar options via initializers for these classes to influence how data transfer is handled.

      Support for additional subscribers can be added via Subscription.define.

  • #subscribe(*args, mode: nil, scheduler: nil, capacity: nil) {|*args, payload| ... } ⇒ Subscription

    Adds a subscription callback. Messages run in order on the selected scheduler. A full queue drops new messages. Unsubscribing drains queued messages and stops the callback.

    Examples:

    counter = Farce::Counter.new

publisher.subscribe(counter) do |counter, payload|
  counter.increment
  puts "Message ##{counter}: #{payload.inspect}"
end

    Parameters:

    • args (Array) —

      additional arguments to pass to the subscriber

    • mode (Symbol, nil) (defaults to: nil) —

      The transfer mode to use

    • scheduler (#schedule, nil) (defaults to: nil) —

      The scheduler to use for running the block

    • capacity (Integer, nil) (defaults to: nil) —

      the maximum queued messages, or nil for an unbounded queue

    Yields:

    • (*args, payload) —

      the block to be called with the given arguments and payload

    Yield Parameters:

    • args (Array) —

      the additional arguments passed to the block

    • payload (Object) —

      the payload sent via #publish

Returns:



108
 
# File 'lib/farce/abstract/publisher.rb', line 108

def subscribe(...) = block_given? ? create_subscription(...) : add_subscription(...)

#subscribed?(subscriber) ⇒ Boolean

Returns whether the given subscriber is currently subscribed.

Returns:

  • (Boolean) —

    whether the given subscriber is currently subscribed.



124
 
# File 'lib/farce/abstract/publisher.rb', line 124

def subscribed?(subscriber) = !!subscription_for(subscriber)&.subscribed?

#subscription_for(subscriber) ⇒ Subscription?

Returns the subscription for the given subscriber, or nil if none exists.

Parameters:

  • subscriber (BasicObject) —

    the subscriber to retrieve the subscription for

Returns:

  • (Subscription, nil) —

    the subscription for the given subscriber, or nil if none exists.



113
114
115
116
117
118
119
120
121
 
# File 'lib/farce/abstract/publisher.rb', line 113

def subscription_for(subscriber)
  if subscriber.is_a?(Subscription)
    return unless subscriber.publisher.equal?(self)
    subscription = map[subscriber.key]
    subscription if subscription.equal?(subscriber)
  else
    map[subscriber.__id__]
  end
end

#unsubscribe(subscriber) Also known as: cancel, unmonitor

This method returns an undefined value.

Unsubscribes the given subscriber from the publisher.

Parameters:

  • subscriber (BasicObject) —

    the subscriber to unsubscribe



129
 
# File 'lib/farce/abstract/publisher.rb', line 129

def unsubscribe(subscriber) = subscription_for(subscriber)&.unsubscribe