Class: Puppet::Util::Queue::Stomp

Inherits:
Object
  • Object
show all
Defined in:
lib/vendor/puppet/util/queue/stomp.rb

Overview

Implements the Ruby Stomp client as a queue type within the Puppet::Indirector::Queue::Client registry, for use with the :queue indirection terminus type.

Looks to Puppet[:queue_source] for the sole argument to the underlying Stomp::Client constructor; consequently, for this client to work, Puppet[:queue_source] must use the Stomp::Client URL-like syntax for identifying the Stomp message broker: login:[email protected]

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Stomp

Returns a new instance of Stomp.



16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
 
# File 'lib/vendor/puppet/util/queue/stomp.rb', line 16

def initialize
  begin
    uri = URI.parse(Puppet[:queue_source])
  rescue => detail
    raise ArgumentError, "Could not create Stomp client instance - queue source #{Puppet[:queue_source]} is invalid: #{detail}"
  end
  unless uri.scheme == "stomp"
    raise ArgumentError, "Could not create Stomp client instance - queue source #{Puppet[:queue_source]} is not a Stomp URL: #{detail}"
  end

  begin
    self.stomp_client = Stomp::Client.new(uri.user, uri.password, uri.host, uri.port, true)
  rescue => detail
    raise ArgumentError, "Could not create Stomp client instance with queue source #{Puppet[:queue_source]}: got internal Stomp client error #{detail}"
  end

  # Identify the supported method for sending messages.
  @method =
    case
    when stomp_client.respond_to?(:publish)
      :publish
    when stomp_client.respond_to?(:send)
      :send
    else
      raise ArgumentError, "STOMP client does not respond to either publish or send"
    end
end

Instance Attribute Details

#stomp_client ⇒ Object

Returns the value of attribute stomp_client.



14
15
16
 
# File 'lib/vendor/puppet/util/queue/stomp.rb', line 14

def stomp_client
  @stomp_client
end

Instance Method Details

#publish_message(target, msg) ⇒ Object 



44
45
46
 
# File 'lib/vendor/puppet/util/queue/stomp.rb', line 44

def publish_message(target, msg)
  stomp_client.__send__(@method, stompify_target(target), msg, :persistent => true)
end

#stompify_target(target) ⇒ Object 



55
56
57
 
# File 'lib/vendor/puppet/util/queue/stomp.rb', line 55

def stompify_target(target)
  '/queue/' + target.to_s
end

#subscribe(target) ⇒ Object 



48
49
50
51
52
53
 
# File 'lib/vendor/puppet/util/queue/stomp.rb', line 48

def subscribe(target)
  stomp_client.subscribe(stompify_target(target), :ack => :client) do |stomp_message|
    yield(stomp_message.body)
    stomp_client.acknowledge(stomp_message)
  end
end