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 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.
19 20 21 |
# File 'lib/pebblebed/river.rb', line 19 def initialize(env = ENV['RACK_ENV']) @environment = env || 'development' end |
Class Method Details
.route(options) ⇒ Object
9 10 11 12 13 14 15 16 |
# File 'lib/pebblebed/river.rb', line 9 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
27 28 29 30 31 |
# File 'lib/pebblebed/river.rb', line 27 def connect unless connected? bunny.start end end |
#connected? ⇒ Boolean
23 24 25 |
# File 'lib/pebblebed/river.rb', line 23 def connected? bunny.connected? end |
#disconnect ⇒ Object
33 34 35 |
# File 'lib/pebblebed/river.rb', line 33 def disconnect bunny.stop if connected? end |
#exchange_name ⇒ Object
57 58 59 60 61 62 63 64 |
# File 'lib/pebblebed/river.rb', line 57 def exchange_name unless @exchange_name name = 'pebblebed.river' name << ".#{environment}" unless production? @exchange_name = name end @exchange_name end |
#publish(options = {}) ⇒ Object
37 38 39 40 41 42 43 |
# File 'lib/pebblebed/river.rb', line 37 def publish( = {}) connect persistent = .fetch(:persistent) { true } key = self.class.route() exchange.publish(.to_json, :persistent => persistent, :key => key) end |
#queue(options = {}) ⇒ Object
45 46 47 48 49 50 51 52 53 54 55 |
# File 'lib/pebblebed/river.rb', line 45 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 |