Class: Stomper::SubscriptionManager

Inherits:
Object
  • Object
show all
Defined in:
lib/stomper/subscription_manager.rb

Overview

Manages subscriptions for a Connection

Defined Under Namespace

Classes: Subscription

Instance Method Summary collapse

Constructor Details

#initialize(connection) ⇒ SubscriptionManager

Creates a new subscription handler for the supplied connection

Parameters:



7
8
9
10
11
12
 
# File 'lib/stomper/subscription_manager.rb', line 7

def initialize(connection)
  @mon = ::Monitor.new
  @subscriptions = {}
  connection.on_message { |m, con| dispatch(m) }
  connection.on_unsubscribe { |u, con| remove(u[:id]) }
end

Instance Method Details

#add(subscribe, callback) ⇒ self

Adds a callback handler for a MESSAGE frame that is sent via the subscription associated with the supplied SUBSCRIBE frame.

Parameters:

  • subscribe (Stomper::Frame) —

    SUBSCRIBE frame for the subscription

  • callback (Proc) —

    Proc to invoke when a matching MESSAGE frame is received from the broker.

Returns:

  • (self) 


20
21
22
23
24
25
26
 
# File 'lib/stomper/subscription_manager.rb', line 20

def add(subscribe, callback)
  s_id = subscribe[:id]
  dest = subscribe[:destination]
  @mon.synchronize do
    @subscriptions[s_id] = Subscription.new(subscribe, callback)
  end
end

#clear ⇒ Object

Remove all subscriptions. This method does not send UNSUBSCRIBE frames to the broker.



56
57
58
 
# File 'lib/stomper/subscription_manager.rb', line 56

def clear
  @mon.synchronize { @subscriptions.clear }
end

#remove(sub_id) ⇒ Array<String>

Removes a subscription by ID or destination.

Parameters:

  • sub_id (String) —

    ID or destination of the subscription

Returns:

  • (Array<String>) —

    array of subscription IDs matching sub_id



31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
 
# File 'lib/stomper/subscription_manager.rb', line 31

def remove(sub_id)
  @mon.synchronize do
    if @subscriptions.key? sub_id
      @subscriptions.delete sub_id
      [sub_id]
    else
      @subscriptions.values.inject([]) do |ids, sub|
        if sub.destination == sub_id
          @subscriptions.delete sub.id
          ids << sub.id
        end
        ids
      end
    end
  end
end

#subscriptions ⇒ Array<Stomper::SubscriptionManager::Subscription>

Returns all current subscriptions.

Returns:



50
51
52
 
# File 'lib/stomper/subscription_manager.rb', line 50

def subscriptions
  @mon.synchronize { @subscriptions.values }
end