Class: OmfCommon::Comm::XMPP::Communicator
- Inherits:
-
OmfCommon::Comm
- Object
- OmfCommon::Comm
- OmfCommon::Comm::XMPP::Communicator
- Includes:
- Blather::DSL
- Defined in:
- lib/omf_common/comm/xmpp/communicator.rb
Constant Summary collapse
- HOST_PREFIX =
'pubsub'
- RETRY_INTERVAL =
180
- PING_INTERVAL =
1800
- PUBSUB_CONFIGURE =
Blather::Stanza::X.new({ :type => :submit, :fields => [ { :var => "FORM_TYPE", :type => 'hidden', :value => "http://jabber.org/protocol/pubsub#node_config" }, { :var => "pubsub#persist_items", :value => "0" }, { :var => "pubsub#purge_offline", :value => "1" }, { :var => "pubsub#send_last_published_item", :value => "never" }, { :var => "pubsub#notify_retract", :value => "0" }, { :var => "pubsub#publish_model", :value => "open" }] })
Instance Attribute Summary collapse
-
#normal_shutdown_mode ⇒ Object
Returns the value of attribute normal_shutdown_mode.
-
#published_messages ⇒ Object
Returns the value of attribute published_messages.
-
#retry_counter ⇒ Object
Returns the value of attribute retry_counter.
Instance Method Summary collapse
- #_create(topic, pubsub_host = default_host, &block) ⇒ Object
- #_subscribe(topic, pubsub_host = default_host, &block) ⇒ Object
-
#_unsubscribe_one(topic_id, pubsub_host = default_host) ⇒ Object
Un-subscribe one single topic by topic address.
- #affiliations(pubsub_host = default_host, &block) ⇒ Object
- #conn_info ⇒ Object
-
#connect(username, password, server) ⇒ Object
Set up XMPP options and start the Eventmachine, connect to XMPP server.
-
#create_topic(topic, opts = {}) ⇒ Object
Create a new pubsub topic with additional configuration.
-
#delete_topic(topic, pubsub_host = default_host, &block) ⇒ Object
Delete a pubsub topic.
-
#disconnect(opts = {}) ⇒ Object
Shut down XMPP connection.
-
#init(opts = {}) ⇒ Object
Set up XMPP options and start the Eventmachine, connect to XMPP server.
- #on_connected(&block) ⇒ Object
-
#on_interrupted(&block) ⇒ Object
Capture system :INT & :TERM signal.
-
#publish(topic, message, pubsub_host = default_host, &block) ⇒ Object
Publish to a pubsub topic.
- #string_to_topic_address(a_string) ⇒ Object
-
#subscribe(topic, opts = {}, &block) ⇒ Object
Subscribe to a pubsub topic.
-
#topic_event(additional_guard = nil, &block) ⇒ Object
Event callback for pubsub topic event(item published).
-
#unsubscribe(pubsub_host = default_host) ⇒ Object
Un-subscribe all existing subscriptions from all pubsub topics.
Methods inherited from OmfCommon::Comm
init, instance, #local_address, #local_topic, #options
Instance Attribute Details
#normal_shutdown_mode ⇒ Object
Returns the value of attribute normal_shutdown_mode.
52 53 54 |
# File 'lib/omf_common/comm/xmpp/communicator.rb', line 52 def normal_shutdown_mode @normal_shutdown_mode end |
#published_messages ⇒ Object
Returns the value of attribute published_messages.
52 53 54 |
# File 'lib/omf_common/comm/xmpp/communicator.rb', line 52 def @published_messages end |
#retry_counter ⇒ Object
Returns the value of attribute retry_counter.
52 53 54 |
# File 'lib/omf_common/comm/xmpp/communicator.rb', line 52 def retry_counter @retry_counter end |
Instance Method Details
#_create(topic, pubsub_host = default_host, &block) ⇒ Object
224 225 226 |
# File 'lib/omf_common/comm/xmpp/communicator.rb', line 224 def _create(topic, pubsub_host = default_host, &block) pubsub.create(topic, pubsub_host, PUBSUB_CONFIGURE, &callback_logging(__method__, topic, &block)) end |
#_subscribe(topic, pubsub_host = default_host, &block) ⇒ Object
220 221 222 |
# File 'lib/omf_common/comm/xmpp/communicator.rb', line 220 def _subscribe(topic, pubsub_host = default_host, &block) pubsub.subscribe(topic, nil, pubsub_host, &callback_logging(__method__, topic, &block)) end |
#_unsubscribe_one(topic_id, pubsub_host = default_host) ⇒ Object
Un-subscribe one single topic by topic address
241 242 243 244 245 246 247 248 249 |
# File 'lib/omf_common/comm/xmpp/communicator.rb', line 241 def _unsubscribe_one(topic_id, pubsub_host = default_host) pubsub.subscriptions(pubsub_host) do |m| m[:subscribed] && m[:subscribed].each do |s| if s[:node] == topic_id.to_s pubsub.unsubscribe(s[:node], nil, s[:subid], pubsub_host, &callback_logging(__method__, s[:node], s[:subid])) end end end end |
#affiliations(pubsub_host = default_host, &block) ⇒ Object
251 252 253 |
# File 'lib/omf_common/comm/xmpp/communicator.rb', line 251 def affiliations(pubsub_host = default_host, &block) pubsub.affiliations(pubsub_host, &callback_logging(__method__, &block)) end |
#conn_info ⇒ Object
69 70 71 |
# File 'lib/omf_common/comm/xmpp/communicator.rb', line 69 def conn_info { proto: :xmpp, user: jid.node, domain: jid.domain } end |
#connect(username, password, server) ⇒ Object
Set up XMPP options and start the Eventmachine, connect to XMPP server
172 173 174 175 176 177 178 179 180 181 182 |
# File 'lib/omf_common/comm/xmpp/communicator.rb', line 172 def connect(username, password, server) info "Connecting to '#{server}' ..." begin client.run rescue ::EventMachine::ConnectionError, Blather::Stream::ConnectionTimeout, Blather::Stream::NoConnection, Blather::Stream::ConnectionFailed => e warn "[#{e.class}] #{e}, try again..." OmfCommon.el.after(RETRY_INTERVAL) do connect(username, password, server) end end end |
#create_topic(topic, opts = {}) ⇒ Object
Create a new pubsub topic with additional configuration
198 199 200 |
# File 'lib/omf_common/comm/xmpp/communicator.rb', line 198 def create_topic(topic, opts = {}) OmfCommon::Comm::XMPP::Topic.create(topic) end |
#delete_topic(topic, pubsub_host = default_host, &block) ⇒ Object
Delete a pubsub topic
205 206 207 |
# File 'lib/omf_common/comm/xmpp/communicator.rb', line 205 def delete_topic(topic, pubsub_host = default_host, &block) pubsub.delete(topic, pubsub_host, &callback_logging(__method__, topic, &block)) end |
#disconnect(opts = {}) ⇒ Object
Shut down XMPP connection
185 186 187 188 189 190 191 192 193 |
# File 'lib/omf_common/comm/xmpp/communicator.rb', line 185 def disconnect(opts = {}) # NOTE Do not clean up @lock.synchronize do @normal_shutdown_mode = true end info "Disconnecting..." shutdown OmfCommon::DSL::Xmpp::MPConnection.inject(Time.now.to_f, jid, 'disconnect') if OmfCommon::Measure.enabled? end |
#init(opts = {}) ⇒ Object
Set up XMPP options and start the Eventmachine, connect to XMPP server
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 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 |
# File 'lib/omf_common/comm/xmpp/communicator.rb', line 88 def init(opts = {}) @lock = Monitor.new @pubsub_host = opts[:pubsub_domain] if opts[:url] url = URI(opts[:url]) username, password, server = url.user, url.password, url.host else username, password, server = opts[:username], opts[:password], opts[:server] end random_name = "#{Socket.gethostname}-#{Process.pid}" username ||= random_name password ||= random_name raise ArgumentError, "Username cannot be nil when connect to XMPP" if username.nil? raise ArgumentError, "Password cannot be nil when connect to XMPP" if password.nil? raise ArgumentError, "Server cannot be nil when connect to XMPP" if server.nil? @retry_counter = 0 @normal_shutdown_mode = false username.downcase! jid = "#{username}@#{server}" client.setup(jid, password) connect(username, password, server) when_ready do if @not_initial_connection info "Reconnected" else info "Connected" OmfCommon::DSL::Xmpp::MPConnection.inject(Time.now.to_f, jid, 'connected') if OmfCommon::Measure.enabled? @cbks[:connected].each { |cbk| cbk.call(self) } # It will be reconnection after this @lock.synchronize do @not_initial_connection = true end end @lock.synchronize do @pong = true @ping_alive_timer = OmfCommon.el.every(PING_INTERVAL) do if @pong @lock.synchronize do @pong = false # Reset @pong end ping_alive else warn "No PONG. No connection..." @lock.synchronize do @ping_alive_timer.cancel end connect(username, password, server) end end end end disconnected do @lock.synchronize do @pong = false # Reset @pong @ping_alive_timer && @ping_alive_timer.cancel end if normal_shutdown_mode shutdown else warn "Disconnected... Last known state: #{client.state}" retry_interval = client.state == :initializing ? 0 : RETRY_INTERVAL OmfCommon.el.after(retry_interval) do connect(username, password, server) end end end trap(:INT) { @cbks[:interpreted].empty? ? disconnect : @cbks[:interpreted].each { |cbk| cbk.call(self) } } trap(:TERM) { @cbks[:interpreted].empty? ? disconnect : @cbks[:interpreted].each { |cbk| cbk.call(self) } } super end |
#on_connected(&block) ⇒ Object
82 83 84 |
# File 'lib/omf_common/comm/xmpp/communicator.rb', line 82 def on_connected(&block) @cbks[:connected] << block end |
#on_interrupted(&block) ⇒ Object
Capture system :INT & :TERM signal
78 79 80 |
# File 'lib/omf_common/comm/xmpp/communicator.rb', line 78 def on_interrupted(&block) @cbks[:interpreted] << block end |
#publish(topic, message, pubsub_host = default_host, &block) ⇒ Object
Publish to a pubsub topic
259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 |
# File 'lib/omf_common/comm/xmpp/communicator.rb', line 259 def publish(topic, , pubsub_host = default_host, &block) raise StandardError, "Invalid message" unless .valid? = .marshall[1] unless .kind_of? String if .nil? debug "Cannot publish empty message, using authentication and not providing a proper cert?" return nil end new_block = proc do |stanza| << OpenSSL::Digest::SHA1.new(.to_s) block.call(stanza) if block end pubsub.publish(topic, , pubsub_host, &callback_logging(__method__, topic, &new_block)) OmfCommon::DSL::Xmpp::MPPublished.inject(Time.now.to_f, jid, topic, .to_s[/mid="(.{36})/, 1]) if OmfCommon::Measure.enabled? end |
#string_to_topic_address(a_string) ⇒ Object
73 74 75 |
# File 'lib/omf_common/comm/xmpp/communicator.rb', line 73 def string_to_topic_address(a_string) "xmpp://#{a_string}@#{jid.domain}" end |
#subscribe(topic, opts = {}, &block) ⇒ Object
Subscribe to a pubsub topic
214 215 216 217 218 |
# File 'lib/omf_common/comm/xmpp/communicator.rb', line 214 def subscribe(topic, opts = {}, &block) topic = topic.first if topic.is_a? Array OmfCommon::Comm::XMPP::Topic.create(topic, &block) OmfCommon::DSL::Xmpp::MPSubscription.inject(Time.now.to_f, jid, 'join', topic) if OmfCommon::Measure.enabled? end |
#topic_event(additional_guard = nil, &block) ⇒ Object
Event callback for pubsub topic event(item published)
279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 |
# File 'lib/omf_common/comm/xmpp/communicator.rb', line 279 def topic_event(additional_guard = nil, &block) guard_block = proc do |event| passed = !event.delayed? && event.items? && !event.items.first.payload.nil? #&& #!published_messages.include?(OpenSSL::Digest::SHA1.new(event.items.first.payload)) if additional_guard passed && additional_guard.call(event) else passed end end mblock = proc do |stanza| OmfCommon::DSL::Xmpp::MPReceived.inject(Time.now.to_f, jid, stanza.node, stanza.to_s[/mid="(.{36})/, 1]) if OmfCommon::Measure.enabled? block.call(stanza) if block end pubsub_event(guard_block, &callback_logging(__method__, &mblock)) end |
#unsubscribe(pubsub_host = default_host) ⇒ Object
Un-subscribe all existing subscriptions from all pubsub topics
230 231 232 233 234 235 236 237 |
# File 'lib/omf_common/comm/xmpp/communicator.rb', line 230 def unsubscribe(pubsub_host = default_host) pubsub.subscriptions(pubsub_host) do |m| m[:subscribed] && m[:subscribed].each do |s| pubsub.unsubscribe(s[:node], nil, s[:subid], pubsub_host, &callback_logging(__method__, s[:node], s[:subid])) OmfCommon::DSL::Xmpp::MPSubscription.inject(Time.now.to_f, jid, 'leave', s[:node]) if OmfCommon::Measure.enabled? end end end |