Class: Jabber::Stream

Inherits:
Object
  • Object
show all
Defined in:
lib/xmpp4r/stream.rb

Overview

The stream class manages a connection stream (a file descriptor using which XML messages are read and sent)

You may register callbacks for the three Jabber stanzas (message, presence and iq) and use the send and send_with_id methods.

To ensure the order of received stanzas, callback blocks are launched in the parser thread. If further blocking operations are intended in those callbacks, run your own thread there.

Direct Known Subclasses

Connection

Defined Under Namespace

Classes: ThreadBlock

Constant Summary collapse

DISCONNECTED = 
1
CONNECTED = 
2

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Stream

Initialize a new stream



43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
 
# File 'lib/xmpp4r/stream.rb', line 43

def initialize
  @fd = nil
  @status = DISCONNECTED
  @xmlcbs = CallbackList.new
  @stanzacbs = CallbackList.new
  @messagecbs = CallbackList.new
  @iqcbs = CallbackList.new
  @presencecbs = CallbackList.new
  @send_lock = Mutex.new
  @last_send = Time.now
  @exception_block = nil
  @tbcbmutex = Mutex.new
  @threadblocks = []
  @wakeup_thread = nil
  @streamid = nil
  @streamns = 'jabber:client'
  @features_sem = Semaphore.new
  @parser_thread = nil
  @processing = 0
end

Instance Attribute Details

#fd ⇒ Object (readonly)

file descriptor used



33
34
35
 
# File 'lib/xmpp4r/stream.rb', line 33

def fd
  @fd
end

#processing ⇒ Object (readonly)

number of stanzas currently being processed



39
40
41
 
# File 'lib/xmpp4r/stream.rb', line 39

def processing
  @processing
end

#status ⇒ Object (readonly)

connection status



36
37
38
 
# File 'lib/xmpp4r/stream.rb', line 36

def status
  @status
end

Instance Method Details

#add_iq_callback(priority = 0, ref = nil, &block) ⇒ Object

Adds a callback block to process received Iqs

priority
Integer

The callback’s priority, the higher, the sooner

ref
String

The callback’s reference

&block
Block

The optional block



564
565
566
567
568
 
# File 'lib/xmpp4r/stream.rb', line 564

def add_iq_callback(priority = 0, ref = nil, &block)
  @tbcbmutex.synchronize do
    @iqcbs.add(priority, ref, block)
  end
end

#add_message_callback(priority = 0, ref = nil, &block) ⇒ Object

Adds a callback block to process received Messages

priority
Integer

The callback’s priority, the higher, the sooner

ref
String

The callback’s reference

&block
Block

The optional block



498
499
500
501
502
 
# File 'lib/xmpp4r/stream.rb', line 498

def add_message_callback(priority = 0, ref = nil, &block)
  @tbcbmutex.synchronize do
    @messagecbs.add(priority, ref, block)
  end
end

#add_presence_callback(priority = 0, ref = nil, &block) ⇒ Object

Adds a callback block to process received Presences

priority
Integer

The callback’s priority, the higher, the sooner

ref
String

The callback’s reference

&block
Block

The optional block



542
543
544
545
546
 
# File 'lib/xmpp4r/stream.rb', line 542

def add_presence_callback(priority = 0, ref = nil, &block)
  @tbcbmutex.synchronize do
    @presencecbs.add(priority, ref, block)
  end
end

#add_stanza_callback(priority = 0, ref = nil, &block) ⇒ Object

Adds a callback block to process received Stanzas

priority
Integer

The callback’s priority, the higher, the sooner

ref
String

The callback’s reference

&block
Block

The optional block



520
521
522
523
524
 
# File 'lib/xmpp4r/stream.rb', line 520

def add_stanza_callback(priority = 0, ref = nil, &block)
  @tbcbmutex.synchronize do
    @stanzacbs.add(priority, ref, block)
  end
end

#add_xml_callback(priority = 0, ref = nil, &block) ⇒ Object

Adds a callback block to process received XML messages, these will be handled before any blocks given to Stream#send or other callbacks.

priority
Integer

The callback’s priority, the higher, the sooner

ref
String

The callback’s reference

&block
Block

The optional block



476
477
478
479
480
 
# File 'lib/xmpp4r/stream.rb', line 476

def add_xml_callback(priority = 0, ref = nil, &block)
  @tbcbmutex.synchronize do
    @xmlcbs.add(priority, ref, block)
  end
end

#close ⇒ Object

Closes the connection to the Jabber service



582
583
584
 
# File 'lib/xmpp4r/stream.rb', line 582

def close
  close!
end

#close! ⇒ Object 



586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
 
# File 'lib/xmpp4r/stream.rb', line 586

def close!
  pr = 1
  n = 0
  # In some cases, we might lost count of some stanzas
  # (for example, if the handler raises an exception)
  # so we can't block forever.
  while pr > 0 and n <= 20
    @tbcbmutex.synchronize { pr = @processing }
    if pr > 0
      n += 1
      Jabber::debuglog("TRYING TO CLOSE, STILL PROCESSING #{pr} STANZAS")
      #puts("TRYING TO CLOSE, STILL PROCESSING #{pr} STANZAS")
      sleep 0.1
    end
  end

  # Order Matters here! If this method is called from within 
  # @parser_thread then killing @parser_thread first would 
  # mean the other parts of the method fail to execute. 
  # That would be bad. So kill parser_thread last
  @fd.close if @fd and !@fd.closed?
  @status = DISCONNECTED
  @parser_thread.kill if @parser_thread
end

#delete_iq_callback(ref) ⇒ Object

Delete an Iq callback

ref
String

The reference of the callback to delete



575
576
577
578
579
 
# File 'lib/xmpp4r/stream.rb', line 575

def delete_iq_callback(ref)
  @tbcbmutex.synchronize do
    @iqcbs.delete(ref)
  end
end

#delete_message_callback(ref) ⇒ Object

Delete an Message callback

ref
String

The reference of the callback to delete



508
509
510
511
512
 
# File 'lib/xmpp4r/stream.rb', line 508

def delete_message_callback(ref)
  @tbcbmutex.synchronize do
    @messagecbs.delete(ref)
  end
end

#delete_presence_callback(ref) ⇒ Object

Delete a Presence callback

ref
String

The reference of the callback to delete



552
553
554
555
556
 
# File 'lib/xmpp4r/stream.rb', line 552

def delete_presence_callback(ref)
  @tbcbmutex.synchronize do
    @presencecbs.delete(ref)
  end
end

#delete_stanza_callback(ref) ⇒ Object

Delete a Stanza callback

ref
String

The reference of the callback to delete



530
531
532
533
534
 
# File 'lib/xmpp4r/stream.rb', line 530

def delete_stanza_callback(ref)
  @tbcbmutex.synchronize do
    @stanzacbs.delete(ref)
  end
end

#delete_xml_callback(ref) ⇒ Object

Delete an XML-messages callback

ref
String

The reference of the callback to delete



486
487
488
489
490
 
# File 'lib/xmpp4r/stream.rb', line 486

def delete_xml_callback(ref)
  @tbcbmutex.synchronize do
    @xmlcbs.delete(ref)
  end
end

#iq_callbacks ⇒ Object

Get the list of iq callbacks.



296
297
298
 
# File 'lib/xmpp4r/stream.rb', line 296

def iq_callbacks
  @iqcbs
end

#is_connected? ⇒ Boolean

Returns if this connection is connected to a Jabber service

return
Boolean

Connection status

Returns:

  • (Boolean) 


167
168
169
 
# File 'lib/xmpp4r/stream.rb', line 167

def is_connected?
  return @status == CONNECTED
end

#is_disconnected? ⇒ Boolean

Returns if this connection is NOT connected to a Jabber service

return
Boolean

Connection status

Returns:

  • (Boolean) 


175
176
177
 
# File 'lib/xmpp4r/stream.rb', line 175

def is_disconnected?
  return @status == DISCONNECTED
end

#message_callbacks ⇒ Object

Get the list of message callbacks.



302
303
304
 
# File 'lib/xmpp4r/stream.rb', line 302

def message_callbacks
  @messagecbs
end

#on_exception(&block) ⇒ Object

Mounts a block to handle exceptions if they occur during the poll send. This will likely be the first indication that the socket dropped in a Jabber Session.

The block has to take three arguments:

  • the Exception

  • the Jabber::Stream object (self)

  • a symbol where it happened, namely :start, :parser, :sending and :end



124
125
126
 
# File 'lib/xmpp4r/stream.rb', line 124

def on_exception(&block)
  @exception_block = block
end

#parse_failure(e) ⇒ Object

This method is called by the parser when a failure occurs



130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
 
# File 'lib/xmpp4r/stream.rb', line 130

def parse_failure(e)
  Jabber::warnlog("EXCEPTION:\n#{e.class}\n#{e.message}\n#{e.backtrace.join("\n")}")

  # A new thread has to be created because close will cause the thread
  # to commit suicide(???)
  if @exception_block
    # New thread, because close will kill the current thread
    Thread.new do
      Thread.current.abort_on_exception = true
      close
      @exception_block.call(e, self, :parser)
    end
  else
    Jabber::warnlog "Stream#parse_failure was called by XML parser. Dumping " +
      "backtrace...\n" + e.exception + "\n#{e.backtrace.join("\n")}"
    close
    raise
  end
end

#parser_end ⇒ Object

This method is called by the parser upon receiving </stream:stream>



152
153
154
155
156
157
158
159
160
161
162
 
# File 'lib/xmpp4r/stream.rb', line 152

def parser_end
  if @exception_block
    Thread.new do
      Thread.current.abort_on_exception = true
      close
      @exception_block.call(nil, self, :close)
    end
  else
    close
  end
end

#presence_callbacks ⇒ Object

Get the list of presence callbacks.



308
309
310
 
# File 'lib/xmpp4r/stream.rb', line 308

def presence_callbacks
  @presencecbs
end

#receive(element) ⇒ Object

Processes a received REXML::Element and executes registered thread blocks and filters against it.

element
REXML::Element

The received element



184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
 
# File 'lib/xmpp4r/stream.rb', line 184

def receive(element)
  @tbcbmutex.synchronize { @processing += 1 }
  Jabber::debuglog("RECEIVED:\n#{element.to_s}")

  if element.namespace('').to_s == '' # REXML namespaces are always strings
    element.add_namespace(@streamns)
  end

  case element.prefix
  when 'stream'
    case element.name
      when 'stream'
        stanza = element
        @streamid = element.attributes['id']
        @streamns = element.namespace('') if element.namespace('')

        # Hack: component streams are basically client streams.
        # Someday we may want to create special stanza classes
        # for components/s2s deriving from normal stanzas but
        # posessing these namespaces
        @streamns = 'jabber:client' if @streamns == 'jabber:component:accept'

        unless element.attributes['version']  # isn't XMPP compliant, so
          Jabber::debuglog("FEATURES: server not XMPP compliant, will not wait for features")
          @features_sem.run                   # don't wait for <stream:features/>
        end
      when 'features'
        stanza = element
        element.each { |e|
          if e.name == 'mechanisms' and e.namespace == 'urn:ietf:params:xml:ns:xmpp-sasl'
            e.each_element('mechanism') { |mech|
              @stream_mechanisms.push(mech.text)
            }
          else
            @stream_features[e.name] = e.namespace
          end
        }
        Jabber::debuglog("FEATURES: received")
        @features_sem.run
      else
        stanza = element
    end
  else
    # Any stanza, classes are registered by XMPPElement::name_xmlns
    begin
      stanza = XMPPStanza::import(element)
    rescue NoNameXmlnsRegistered
      stanza = element
    end
  end

  if @xmlcbs.process(stanza)
    @tbcbmutex.synchronize { @processing -= 1 }
    return true
  end

  # Iterate through blocked threads (= waiting for an answer)
  #
  # We're dup'ping the @threadblocks here, so that we won't end up in an
  # endless loop if Stream#send is being nested. That means, the nested
  # threadblock won't receive the stanza currently processed, but the next
  # one.
  threadblocks = nil
  @tbcbmutex.synchronize do
    threadblocks = @threadblocks.dup
  end
  threadblocks.each { |threadblock|
    exception = nil
    r = false
    begin
      r = threadblock.call(stanza)
    rescue Exception => e
      exception = e
    end

    if r == true
      @tbcbmutex.synchronize do
        @threadblocks.delete(threadblock)
      end
      threadblock.wakeup
      @tbcbmutex.synchronize { @processing -= 1 }
      return true
    elsif exception
      @tbcbmutex.synchronize do
        @threadblocks.delete(threadblock)
      end
      threadblock.raise(exception)
    end
  }

  Jabber::debuglog("PROCESSING:\n#{stanza.to_s} (#{stanza.class})")
  Jabber::debuglog("TRYING stanzacbs...")
  if @stanzacbs.process(stanza)
      @tbcbmutex.synchronize { @processing -= 1 }
      return true
  end
  r = false
  Jabber::debuglog("TRYING message/iq/presence/cbs...")
  case stanza
  when Message
    r = @messagecbs.process(stanza)
  when Iq
    r = @iqcbs.process(stanza)
  when Presence
    r = @presencecbs.process(stanza)
  end
  @tbcbmutex.synchronize { @processing -= 1 }
  return r
end

#send(xml, callback_timeout = nil, &block) ⇒ Object

Sends XML data to the socket and (optionally) waits to process received data.

Do not invoke this in a callback but in a seperate thread because we may not suspend the parser-thread (in whose context callbacks are executed).

Pass in a callback_timeout to force a Timeout::Error if the callback is not completed within that time (in seconds). By default there is not timeout on the callback. TODO Would like to set a default here, but will require lots of testing

xml
String

The xml data to send

&block
Block

The optional block



373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
 
# File 'lib/xmpp4r/stream.rb', line 373

def send(xml, callback_timeout=nil, &block)
  Jabber::debuglog("SENDING:\n#{xml}")
  if block
    threadblock = ThreadBlock.new(block)
    @tbcbmutex.synchronize do
      @threadblocks.unshift(threadblock)
    end
  end
  begin
    # Temporarily remove stanza's namespace to
    # reduce bandwidth consumption
    if xml.kind_of? XMPPStanza and xml.namespace == 'jabber:client' and
        xml.prefix != 'stream' and xml.name != 'stream'
      xml.delete_namespace
      send_data(xml.to_s)
      xml.add_namespace(@streamns)
    else
      send_data(xml.to_s)
    end
  rescue Exception => e
    Jabber::warnlog("EXCEPTION:\n#{e.class}\n#{e.message}\n#{e.backtrace.join("\n")}")

    if @exception_block
      Thread.new do
        Thread.current.abort_on_exception = true
        close!
        @exception_block.call(e, self, :sending)
      end
    else
      Jabber::warnlog "Exception caught while sending! (#{e.class})\n#{e.backtrace.join("\n")}"
      close!
      raise
    end
  end
  # The parser thread might be running this (think of a callback running send())
  # If this is the case, we mustn't stop (or we would cause a deadlock)
  if block and Thread.current != @parser_thread
    Timeout.timeout(callback_timeout) {threadblock.wait}
  elsif block
    Jabber::warnlog("WARNING:\nCannot stop current thread in Jabber::Stream#send because it is the parser thread!")
  end
end

#send_data(data) ⇒ Object 



350
351
352
353
354
355
356
 
# File 'lib/xmpp4r/stream.rb', line 350

def send_data(data)
  @send_lock.synchronize do
    @last_send = Time.now
    @fd << data
    @fd.flush
  end
end

#send_with_id(xml, &block) ⇒ Object

Send an XMMP stanza with an Jabber::XMPPStanza#id. The id will be generated by Jabber::IdGenerator if not already set.

The block will be called once: when receiving a stanza with the same Jabber::XMPPStanza#id. There is no need to return true to complete this! Instead the return value of the block will be returned. This is a direct result of unique request/response stanza identification via the id attribute.

The block may be omitted. Then, the result will be the response stanza.

Be aware that if a stanza with type='error' is received the function does not yield but raises an ServerError with the corresponding error element.

Please see Stream#send for some implementational details.

Please read the note about nesting at Stream#send

xml
XMPPStanza


437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
 
# File 'lib/xmpp4r/stream.rb', line 437

def send_with_id(xml, &block)
  if xml.id.nil?
    xml.id = Jabber::IdGenerator.instance.generate_id
  end

  res = nil
  error = nil
  send(xml) do |received|
    if received.kind_of? XMPPStanza and received.id == xml.id
      if received.type == :error
        error = (received.error ? received.error : ErrorResponse.new)
        true
      elsif block_given?
        res = yield(received)
        true
      else
        res = received
        true
      end
    else
      false
    end
  end

  unless error.nil?
    raise ServerError.new(error)
  end

  res
end

#stanza_callbacks ⇒ Object

Get the list of stanza callbacks.



314
315
316
 
# File 'lib/xmpp4r/stream.rb', line 314

def stanza_callbacks
  @stanzacbs
end

#start(fd) ⇒ Object

Start the XML parser on the fd



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
 
# File 'lib/xmpp4r/stream.rb', line 66

def start(fd)
  @stream_mechanisms = []
  @stream_features = {}

  @fd = fd
  @doc = StreamParser.new(self)
  @parser = Nokogiri::XML::SAX::PushParser.new(@doc)
          
  @parser_thread = Thread.new do
    Thread.current.abort_on_exception = true
    
    begin
      until @fd.eof?
        @parser << @fd.read_nonblock(500)
    end
      @parser.finish
    
      Jabber::debuglog("DISCONNECTED\n")

      if @exception_block
        Thread.new { close!; @exception_block.call(nil, self, :disconnected) }
      else
        close!
      end
    rescue Exception => e
      Jabber::warnlog("EXCEPTION:\n#{e.class}\n#{e.message}\n#{e.backtrace.join("\n")}")

      if @exception_block
        Thread.new do
          Thread.current.abort_on_exception = true
          close
          @exception_block.call(e, self, :start)
        end
      else
        Jabber::warnlog "Exception caught in Parser thread! (#{e.class})\n#{e.backtrace.join("\n")}"
        close!
        raise
      end
    end
  end

  @status = CONNECTED
end

#stop ⇒ Object 



110
111
112
113
 
# File 'lib/xmpp4r/stream.rb', line 110

def stop
  @parser_thread.kill
  @parser = nil
end

#xml_callbacks ⇒ Object

Get the list of xml callbacks.



320
321
322
 
# File 'lib/xmpp4r/stream.rb', line 320

def xml_callbacks
  @xmlcbs
end