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
26 27 28 29 30 31 32 33 34 35 36 37 |
# File 'lib/pcp/client.rb', line 26 def initialize(params = {}) @server = params[:server] || 'wss://localhost:8142/pcp' @ssl_key = params[:ssl_key] @ssl_cert = params[:ssl_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
13 14 15 |
# File 'lib/pcp/client.rb', line 13 def identity @identity end |
#on_message ⇒ Proc
Set a proc that will be used to handle messages
19 20 21 |
# File 'lib/pcp/client.rb', line 19 def @on_message end |
Instance Method Details
#associated? ⇒ true, false
Is the client associated with the server
137 138 139 |
# File 'lib/pcp/client.rb', line 137 def associated? @associated end |
#close ⇒ Object
Disconnect the client
157 158 159 160 161 162 |
# File 'lib/pcp/client.rb', line 157 def close EM.next_tick do @logger.debug { [:close] } @connection.close end end |
#connect(seconds = 0) ⇒ true, ...
Connect to the server
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 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 |
# File 'lib/pcp/client.rb', line 44 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] } @connection = Faye::WebSocket::Client.new(@server, nil, {:tls => {:private_key_file => @ssl_key, :cert_chain_file => @ssl_cert, :ssl_version => ["TLSv1", "TLSv1_1", "TLSv1_2"]}}) @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
146 147 148 149 150 151 152 |
# File 'lib/pcp/client.rb', line 146 def send() EM.next_tick do @logger.debug { [:send, ] } [:sender] = identity @connection.send(.encode) end end |