Class: Protein::AMQPAdapter
- Inherits:
-
Object
- Object
- Protein::AMQPAdapter
- Defined in:
- lib/protein/amqp_adapter.rb
Class Attribute Summary collapse
-
.calls ⇒ Object
Returns the value of attribute calls.
-
.reply_queue ⇒ Object
readonly
Returns the value of attribute reply_queue.
Class Method Summary collapse
- .call(request_payload) ⇒ Object
- .from_hash(hash) ⇒ Object
- .push(message_payload) ⇒ Object
- .queue(queue = nil) ⇒ Object
- .serve(router) ⇒ Object
- .timeout(timeout = :not_set) ⇒ Object
- .url(url = nil) ⇒ Object
Class Attribute Details
.calls ⇒ Object
Returns the value of attribute calls.
38 39 40 |
# File 'lib/protein/amqp_adapter.rb', line 38 def calls @calls end |
.reply_queue ⇒ Object (readonly)
Returns the value of attribute reply_queue.
37 38 39 |
# File 'lib/protein/amqp_adapter.rb', line 37 def reply_queue @reply_queue end |
Class Method Details
.call(request_payload) ⇒ Object
40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 |
# File 'lib/protein/amqp_adapter.rb', line 40 def call(request_payload) prepare_client call_id = SecureRandom.uuid @x.publish(request_payload, correlation_id: call_id, routing_key: @server_queue, reply_to: @reply_queue.name, expiration: timeout) call = Concurrent::Hash.new mutex = Mutex.new condition = ConditionVariable.new call[:mutex] = mutex call[:condition] = condition calls[call_id] = call mutex.synchronize { condition.wait(mutex, timeout && timeout * 0.001) } response = call[:response] calls.delete(call_id) if response == nil raise(TransportError, "timeout after #{timeout}ms") elsif response == "ESRV" raise(TransportError, "failed to process the request") else response end end |
.from_hash(hash) ⇒ Object
8 9 10 11 12 13 14 15 16 17 18 19 20 |
# File 'lib/protein/amqp_adapter.rb', line 8 def from_hash(hash) if (new_url = hash[:url]) url(new_url) end if (new_queue = hash[:queue]) queue(new_queue) end if hash.has_key?(:timeout) timeout(hash[:timeout]) end end |
.push(message_payload) ⇒ Object
72 73 74 75 76 77 |
# File 'lib/protein/amqp_adapter.rb', line 72 def push() prepare_client @x.publish(, routing_key: @server_queue) end |
.queue(queue = nil) ⇒ Object
27 28 29 30 |
# File 'lib/protein/amqp_adapter.rb', line 27 def queue(queue = nil) @queue = queue if queue @queue || raise(DefinitionError, "queue is not defined") end |
.serve(router) ⇒ Object
79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 |
# File 'lib/protein/amqp_adapter.rb', line 79 def serve(router) @conn = Bunny.new(url) @terminating = false @processing = false begin @conn.start rescue Bunny::TCPConnectionFailed => e Protein.logger.error "RPC server connection error: #{e.inspect}" log_error(e) raise(e) end @ch = @conn.create_channel @ch.prefetch(1) @q = @ch.queue(queue) @x = @ch.default_exchange Signal.trap("TERM") do if @processing @terminating = true else exit end end Signal.trap("INT") do if @processing @terminating = true else exit end end Protein.logger.info "Connected to #{url}, serving RPC calls from #{queue}" loop do begin @q.subscribe(block: true, manual_ack: true) do |delivery_info, properties, payload| @processing = true begin @error = nil response = Processor.call(router, payload) rescue Exception => error @error = error response = "ESRV" end if response @x.publish(response, routing_key: properties.reply_to, correlation_id: properties.correlation_id) end @ch.ack(delivery_info.delivery_tag) @processing = false break if @terminating if @error log_error(@error) raise(@error) end end rescue StandardError => e @processing = false break if @terminating log_error(e) Protein.logger.error "RPC server error: #{e.inspect}, restarting the server in 5s..." sleep 5 end end end |
.timeout(timeout = :not_set) ⇒ Object
32 33 34 35 |
# File 'lib/protein/amqp_adapter.rb', line 32 def timeout(timeout = :not_set) @timeout = timeout if timeout != :not_set instance_variable_defined?("@timeout") ? @timeout : 15_000 end |
.url(url = nil) ⇒ Object
22 23 24 25 |
# File 'lib/protein/amqp_adapter.rb', line 22 def url(url = nil) @url = url if url @url || raise(DefinitionError, "url is not defined") end |