Class: Babs::Client

Inherits:
Object
  • Object
show all
Defined in:
lib/babs/client.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(options = {}) ⇒ Client

Returns a new instance of Client.



13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
 
# File 'lib/babs/client.rb', line 13

def initialize(options = {})
  # make this connection part a singleton and a LOT of time is saved, as well as reusing the same connection
  @conn = Bunny.new(:automatically_recover => false)
  @conn.start
  @ch   = @conn.create_channel

  @defaults = Hashie::Mash.new({
      server_queue: nil,
      exchange: @ch.default_exchange
    }.merge(options)
  )

  @lock      = Mutex.new
  @condition = ConditionVariable.new

end

Instance Attribute Details

#call_id ⇒ Object

Returns the value of attribute call_id.



9
10
11
 
# File 'lib/babs/client.rb', line 9

def call_id
  @call_id
end

#condition ⇒ Object (readonly)

Returns the value of attribute condition.



10
11
12
 
# File 'lib/babs/client.rb', line 10

def condition
  @condition
end

#lock ⇒ Object (readonly)

Returns the value of attribute lock.



10
11
12
 
# File 'lib/babs/client.rb', line 10

def lock
  @lock
end

#reply_queue ⇒ Object (readonly)

Returns the value of attribute reply_queue.



11
12
13
 
# File 'lib/babs/client.rb', line 11

def reply_queue
  @reply_queue
end

#response ⇒ Object

Returns the value of attribute response.



9
10
11
 
# File 'lib/babs/client.rb', line 9

def response
  @response
end

#route ⇒ Object

Returns the value of attribute route.



8
9
10
 
# File 'lib/babs/client.rb', line 8

def route
  @route
end

Instance Method Details

#close_connection ⇒ Object 



30
31
32
33
 
# File 'lib/babs/client.rb', line 30

def close_connection
  @ch.close
  @conn.close
end

#listen_for_response ⇒ Object 



35
36
37
38
39
40
41
42
43
44
45
46
 
# File 'lib/babs/client.rb', line 35

def listen_for_response
  # listen on a new queue for this response
  @reply_queue = @ch.queue("", :exclusive => true)
  @reply_queue.subscribe do |delivery_info, properties, payload|
    puts "response_id #{properties[:correlation_id]}"
    puts properties[:correlation_id] == self.call_id ? "correct id" : "BAD id"
    if properties[:correlation_id] == self.call_id
      self.response = payload
      self.lock.synchronize{self.condition.signal}
    end
  end
end

#request(options = {}) ⇒ Object 



65
66
67
68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/babs/client.rb', line 65

def request(options = {})
  options = Hashie::Mash.new(options)
  # grab out the expected data
  method = options.delete(:method)
  params = options.delete(:params)

  # merge the connection options with the defaults
  routing_options = @defaults.merge(options)

  response = listen_for_response
  response = send_request(routing_options, method, params)
  # parse and return response
  Hashie::Mash.new(JSON.parse(response))
end

#send_request(routing_options, method, params) ⇒ Object 



48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
 
# File 'lib/babs/client.rb', line 48

def send_request(routing_options, method, params)
  self.call_id = SecureRandom.uuid

  data_string = {method: method, params: params}.to_json

  routing_options.exchange.publish(
    data_string,
    routing_key:    routing_options.server_queue,
    correlation_id: call_id,
    reply_to:       @reply_queue.name)
  puts "call id #{call_id}"
  self.response = nil
  # params to synchronize are mutex, timeout_in_seconds
  lock.synchronize{condition.wait(lock, 5)}
  response
end