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

Inherits:
Object
  • Object
show all
Defined in:
lib/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.



14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
 
# File 'lib/puppet/util/queue/stomp.rb', line 14

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
end

Instance Attribute Details

#stomp_client ⇒ Object

Returns the value of attribute stomp_client.



12
13
14
 
# File 'lib/puppet/util/queue/stomp.rb', line 12

def stomp_client
  @stomp_client
end

Instance Method Details

#publish_message(target, msg) ⇒ Object 



31
32
33
 
# File 'lib/puppet/util/queue/stomp.rb', line 31

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

#stompify_target(target) ⇒ Object 



42
43
44
 
# File 'lib/puppet/util/queue/stomp.rb', line 42

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

#subscribe(target) ⇒ Object 



35
36
37
38
39
40
 
# File 'lib/puppet/util/queue/stomp.rb', line 35

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