Class: PCP::Client
- Inherits:
-
Object
- Object
- PCP::Client
- Defined in:
- lib/pcp/client.rb
Overview
Manages a client connection to a pcp broker
Instance Attribute Summary collapse
-
#identity ⇒ String
Read the @identity property.
-
#on_message ⇒ Proc
Set a proc that will be used to handle messages.
Instance Method Summary collapse
-
#associated? ⇒ true, false
Is the client associated with the server.
-
#close ⇒ Object
Disconnect the client.
-
#connect(seconds = 0) ⇒ true, ...
Connect to the server.
-
#initialize(params = {}) ⇒ Object
constructor
Construct a new disconnected client.
-
#send(message) ⇒ Object
Send a message to the server.
Constructor Details
#initialize(params = {}) ⇒ Object
Construct a new disconnected client
27 28 29 30 31 32 33 34 35 36 37 38 39 40 |
# File 'lib/pcp/client.rb', line 27 def initialize(params = {}) @server = params[:server] || 'wss://localhost:8142/pcp' @ssl_key = params[:ssl_key] @ssl_cert = params[:ssl_cert] @ssl_ca_cert = params[:ssl_ca_cert] @logger = params[:logger] || Logger.new(STDOUT) @logger.level = params[:loglevel] || Logger::WARN @connection = nil type = params[:type] || "ruby-pcp-client-#{$$}" @identity = make_identity(@ssl_cert, type) @on_message = params[:on_message] @max_message_size = params[:max_message_size] || 64*1024*1024 @associated = false end |
Instance Attribute Details
#identity ⇒ String
Read the @identity property
14 15 16 |
# File 'lib/pcp/client.rb', line 14 def identity @identity end |
#on_message ⇒ Proc
Set a proc that will be used to handle messages
20 21 22 |
# File 'lib/pcp/client.rb', line 20 def @on_message end |
Instance Method Details
#associated? ⇒ true, false
Is the client associated with the server
150 151 152 |
# File 'lib/pcp/client.rb', line 150 def associated? @associated end |
#close ⇒ Object
Disconnect the client
173 174 175 176 177 178 |
# File 'lib/pcp/client.rb', line 173 def close EM.next_tick do @logger.debug { [:close] } @connection.close end end |
#connect(seconds = 0) ⇒ true, ...
Connect to the server
47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 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 |
# File 'lib/pcp/client.rb', line 47 def connect(seconds = 0) if @connection # We close over so much, we really just need to say no for now raise "Can only connect once per client" end mutex = Mutex.new associated_cv = ConditionVariable.new @logger.debug { [:connect, :scheduling] } EM.next_tick do @logger.debug { [:connect, @server] } = { :ssl_version => ["TLSv1", "TLSv1_1", "TLSv1_2"], :private_key_file => @ssl_key, :cert_chain_file => @ssl_cert, :root_cert_file => @ssl_ca_cert, :verify_peer => true, :fail_if_no_peer_cert => true, } @connection = Faye::WebSocket::Client.new(@server, nil, {:tls => , :ping => 30, :max_length => @max_message_size}) @connection.on :open do |event| begin @logger.info { [:open] } send(associate_request) rescue Exception => e @logger.error { [:open_exception, e] } end end @connection.on :message do |event| begin = ::PCP::Message.new(event.data) @logger.debug { [:message, :decoded, ] } if [:message_type] == 'http://puppetlabs.com/associate_response' mutex.synchronize do @associated = JSON.load(.data)["success"] associated_cv.signal end elsif @on_message @on_message.call() end rescue Exception => e @logger.error { [:message_exception, e] } end end @connection.on :close do |event| begin @logger.info { [:close, event.code, event.reason] } mutex.synchronize do @associated = false associated_cv.signal end rescue Exception => e @logger.error { [:close_exception, e] } end end @connection.on :error do |event| @logger.error { [:error, event] } @associated = false end end if !EM.reactor_running? @logger.debug { [:no_eventmachine_reactor, "Eventmachine reactor is not running" ] } return nil end if EM.reactor_thread? # Because we use a condition variable to signal this thread # from the reactor thread to provide an imperative interface, # they cannot be the same thread. We might associate later, # we just can't wait on ourselves from here. @logger.debug { [:connection_cannot_wait, "Cannot wait on a connection if we are in the same thread as the reactor" ] } return nil end begin Timeout::timeout(seconds) do mutex.synchronize do associated_cv.wait(mutex) return @associated end end rescue Timeout::Error return nil end end |
#send(message) ⇒ Object
Send a message to the server
159 160 161 162 163 164 165 166 167 168 |
# File 'lib/pcp/client.rb', line 159 def send() EM.next_tick do @logger.debug { [:send, ] } if [:expires].nil? .expires(3) end [:sender] = identity @connection.send(.encode) end end |