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
72 73 74 75 76 77 78 79 80 81 82 83 84 |
# File 'lib/pcp/client.rb', line 72 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] @associated = false end |
Instance Attribute Details
#identity ⇒ String
Read the @identity property
59 60 61 |
# File 'lib/pcp/client.rb', line 59 def identity @identity end |
#on_message ⇒ Proc
Set a proc that will be used to handle messages
65 66 67 |
# File 'lib/pcp/client.rb', line 65 def @on_message end |
Instance Method Details
#associated? ⇒ true, false
Is the client associated with the server
196 197 198 |
# File 'lib/pcp/client.rb', line 196 def associated? @associated end |
#close ⇒ Object
Disconnect the client
216 217 218 219 220 221 |
# File 'lib/pcp/client.rb', line 216 def close EM.next_tick do @logger.debug { [:close] } @connection.close end end |
#connect(seconds = 0) ⇒ true, ...
Connect to the server
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 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 |
# File 'lib/pcp/client.rb', line 91 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, :verify_peer => true, :fail_if_no_peer_cert => true, # side-channeled properties we want around during ssl # verification are prefixed with xxx_. :xxx_logger => @logger, :xxx_ssl_ca_cert => @ssl_ca_cert, :xxx_hostname => URI.parse(@server).host, } @connection = Faye::WebSocket::Client.new(@server, nil, {:tls => }) @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
205 206 207 208 209 210 211 |
# File 'lib/pcp/client.rb', line 205 def send() EM.next_tick do @logger.debug { [:send, ] } [:sender] = identity @connection.send(.encode) end end |