Class: Pebblebed::River
- Inherits:
-
Object
- Object
- Pebblebed::River
- Defined in:
- lib/pebblebed/river.rb,
lib/pebblebed/river/subscription.rb
Defined Under Namespace
Classes: Subscription
Class Attribute Summary collapse
-
.rabbitmq_options ⇒ Object
Returns the value of attribute rabbitmq_options.
Class Method Summary collapse
Instance Method Summary collapse
- #connect ⇒ Object
- #connected? ⇒ Boolean
- #disconnect ⇒ Object
- #exchange_name ⇒ Object
-
#initialize(env = ) ⇒ River
constructor
A new instance of River.
- #publish(options = {}) ⇒ Object
- #queue(options = {}) ⇒ Object
Constructor Details
#initialize(env = ) ⇒ River
Returns a new instance of River.
21 22 23 |
# File 'lib/pebblebed/river.rb', line 21 def initialize(env = ENV['RACK_ENV']) @environment = env || 'development' end |
Class Attribute Details
.rabbitmq_options ⇒ Object
Returns the value of attribute rabbitmq_options.
9 10 11 |
# File 'lib/pebblebed/river.rb', line 9 def @rabbitmq_options end |
Class Method Details
.route(options) ⇒ Object
11 12 13 14 15 16 17 18 |
# File 'lib/pebblebed/river.rb', line 11 def route() raise ArgumentError.new(':event is required') unless [:event] raise ArgumentError.new(':uid is required') unless [:uid] uid = Pebblebed::Uid.new([:uid]) key = [[:event], uid.klass, uid.path].compact key.join('._.') end |
Instance Method Details
#connect ⇒ Object
29 30 31 32 33 |
# File 'lib/pebblebed/river.rb', line 29 def connect unless connected? bunny.start end end |
#connected? ⇒ Boolean
25 26 27 |
# File 'lib/pebblebed/river.rb', line 25 def connected? bunny.connected? end |
#disconnect ⇒ Object
35 36 37 |
# File 'lib/pebblebed/river.rb', line 35 def disconnect bunny.stop if connected? end |
#exchange_name ⇒ Object
59 60 61 62 63 64 65 66 |
# File 'lib/pebblebed/river.rb', line 59 def exchange_name unless @exchange_name name = 'pebblebed.river' name << ".#{environment}" unless production? @exchange_name = name end @exchange_name end |
#publish(options = {}) ⇒ Object
39 40 41 42 43 44 45 |
# File 'lib/pebblebed/river.rb', line 39 def publish( = {}) connect persistent = .fetch(:persistent) { true } key = self.class.route() exchange.publish(.to_json, :persistent => persistent, :key => key) end |
#queue(options = {}) ⇒ Object
47 48 49 50 51 52 53 54 55 56 57 |
# File 'lib/pebblebed/river.rb', line 47 def queue( = {}) connect raise ArgumentError.new 'Queue must be named' unless [:name] queue = channel.queue([:name], :durable => true) Subscription.new().queries.each do |key| queue.bind(exchange.name, :key => key) end queue end |