gloox  1.0.1
connectionbosh.cpp
1 /*
2  * Copyright (c) 2007-2012 by Jakob Schroeter <js@camaya.net>
3  * This file is part of the gloox library. http://camaya.net/gloox
4  *
5  * This software is distributed under a license. The full license
6  * agreement can be found in the file LICENSE in this distribution.
7  * This software may not be copied, modified, sold or distributed
8  * other than expressed in the named license agreement.
9  *
10  * This software is distributed without any warranty.
11  */
12 
13 #include "config.h"
14 
15 #include "gloox.h"
16 
17 #include "connectionbosh.h"
18 #include "logsink.h"
19 #include "prep.h"
20 #include "tag.h"
21 #include "util.h"
22 
23 #include <string>
24 #include <cstdlib>
25 #include <cctype>
26 #include <algorithm>
27 
28 namespace gloox
29 {
30 
31  ConnectionBOSH::ConnectionBOSH( ConnectionBase* connection, const LogSink& logInstance,
32  const std::string& boshHost, const std::string& xmppServer,
33  int xmppPort )
34  : ConnectionBase( 0 ),
35  m_logInstance( logInstance ), m_parser( this ), m_boshHost( boshHost ), m_path( "/http-bind/" ),
36  m_rid( 0 ), m_initialStreamSent( false ), m_openRequests( 0 ),
37  m_maxOpenRequests( 2 ), m_wait( 30 ), m_hold( 2 ), m_streamRestart( false ),
38  m_lastRequestTime( std::time( 0 ) ), m_minTimePerRequest( 0 ), m_bufferContentLength( 0 ),
39  m_connMode( ModePipelining )
40  {
41  initInstance( connection, xmppServer, xmppPort );
42  }
43 
45  const LogSink& logInstance, const std::string& boshHost,
46  const std::string& xmppServer, int xmppPort )
47  : ConnectionBase( cdh ),
48  m_logInstance( logInstance ), m_parser( this ), m_boshHost( boshHost ), m_path( "/http-bind/" ),
49  m_rid( 0 ), m_initialStreamSent( false ), m_openRequests( 0 ),
50  m_maxOpenRequests( 2 ), m_wait( 30 ), m_hold( 2 ), m_streamRestart( false ),
51  m_lastRequestTime( std::time( 0 ) ), m_minTimePerRequest( 0 ), m_bufferContentLength( 0 ),
52  m_connMode( ModePipelining )
53  {
54  initInstance( connection, xmppServer, xmppPort );
55  }
56 
57  void ConnectionBOSH::initInstance( ConnectionBase* connection, const std::string& xmppServer,
58  const int xmppPort )
59  {
60 // FIXME: check return value
61  prep::idna( xmppServer, m_server );
62  m_port = xmppPort;
63  if( m_port != -1 )
64  {
65  m_boshedHost = m_boshHost + ":" + util::int2string( m_port );
66  }
67 
68  // drop this connection into our pool of available connections
69  if( connection )
70  {
71  connection->registerConnectionDataHandler( this );
72  m_connectionPool.push_back( connection );
73  }
74  }
75 
77  {
78  util::clearList( m_activeConnections );
79  util::clearList( m_connectionPool );
80  }
81 
82  ConnectionBase* ConnectionBOSH::newInstance() const
83  {
84  ConnectionBase* pBaseConn = 0;
85 
86  if( !m_connectionPool.empty() )
87  {
88  pBaseConn = m_connectionPool.front()->newInstance();
89  }
90  else if( !m_activeConnections.empty() )
91  {
92  pBaseConn = m_activeConnections.front()->newInstance();
93  }
94  else
95  {
96  return 0;
97  }
98 
99  return new ConnectionBOSH( m_handler, pBaseConn, m_logInstance,
100  m_boshHost, m_server, m_port );
101  }
102 
103  ConnectionError ConnectionBOSH::connect()
104  {
105  if( m_state >= StateConnecting )
106  return ConnNoError;
107 
108  if( !m_handler )
109  return ConnNotConnected;
110 
112  m_logInstance.dbg( LogAreaClassConnectionBOSH,
113  "bosh initiating connection to server: " +
114  ( ( m_connMode == ModePipelining ) ? std::string( "Pipelining" )
115  : ( ( m_connMode == ModeLegacyHTTP ) ? std::string( "LegacyHTTP" )
116  : std::string( "PersistentHTTP" ) ) ) );
117  getConnection();
118  return ConnNoError; // FIXME?
119  }
120 
121  void ConnectionBOSH::disconnect()
122  {
123  if( ( m_connMode == ModePipelining && m_activeConnections.empty() )
124  || ( m_connectionPool.empty() && m_activeConnections.empty() ) )
125  return;
126 
127  if( m_state != StateDisconnected )
128  {
129  ++m_rid;
130 
131  std::string requestBody = "<body rid='" + util::int2string( m_rid ) + "' ";
132  requestBody += "sid='" + m_sid + "' ";
133  requestBody += "type='terminal' ";
134  requestBody += "xml:lang='en' ";
135  requestBody += "xmlns='" + XMLNS_HTTPBIND + "'";
136  if( m_sendBuffer.empty() ) // Make sure that any data in the send buffer gets sent
137  requestBody += "/>";
138  else
139  {
140  requestBody += ">" + m_sendBuffer + "</body>";
141  m_sendBuffer = EmptyString;
142  }
143  sendRequest( requestBody );
144 
145  m_logInstance.dbg( LogAreaClassConnectionBOSH, "bosh disconnection request sent" );
146  }
147  else
148  {
149  m_logInstance.err( LogAreaClassConnectionBOSH,
150  "disconnecting from server in a non-graceful fashion" );
151  }
152 
153  util::ForEach( m_activeConnections, &ConnectionBase::disconnect );
154  util::ForEach( m_connectionPool, &ConnectionBase::disconnect );
155 
157  if( m_handler )
159  }
160 
161  ConnectionError ConnectionBOSH::recv( int timeout )
162  {
163  if( m_state == StateDisconnected )
164  return ConnNotConnected;
165 
166  if( !m_connectionPool.empty() )
167  m_connectionPool.front()->recv( 0 );
168  if( !m_activeConnections.empty() )
169  m_activeConnections.front()->recv( timeout );
170 
171  // If there are no open requests then the spec allows us to send an empty request...
172  // (Some CMs do not obey this, it seems)
173  if( ( m_openRequests == 0 || m_sendBuffer.size() > 0 ) && m_state == StateConnected )
174  {
175  m_logInstance.dbg( LogAreaClassConnectionBOSH,
176  "Sending empty request (or there is data in the send buffer)" );
177  sendXML();
178  }
179 
180  return ConnNoError; // FIXME?
181  }
182 
183  bool ConnectionBOSH::send( const std::string& data )
184  {
185 
186  if( m_state == StateDisconnected )
187  return false;
188 
189  if( data.substr( 0, 2 ) == "<?" )
190  {
191 // if( m_initialStreamSent )
192  {
193  m_streamRestart = true;
194  sendXML();
195  return true;
196  }
197 // else
198 // {
199 // m_initialStreamSent = true;
200 // m_logInstance.dbg( LogAreaClassConnectionBOSH, "initial <stream:stream> dropped" );
201 // return true;
202 // }
203  }
204  else if( data == "</stream:stream>" )
205  return true;
206 
207  m_sendBuffer += data;
208  sendXML();
209 
210  return true;
211  }
212 
213  /* Sends XML. Wraps data in a <body/> tag, and then passes to sendRequest(). */
214  bool ConnectionBOSH::sendXML()
215  {
216  if( m_state != StateConnected )
217  {
218  m_logInstance.warn( LogAreaClassConnectionBOSH,
219  "Data sent before connection established (will be buffered)" );
220  return false;
221  }
222 
223  if( m_sendBuffer.empty() )
224  {
225  time_t now = time( 0 );
226  unsigned int delta = (int)(now - m_lastRequestTime);
227  if( delta < m_minTimePerRequest && m_openRequests > 0 )
228  {
229  m_logInstance.dbg( LogAreaClassConnectionBOSH, "Too little time between requests: " + util::int2string( delta ) + " seconds" );
230  return false;
231  }
232  m_logInstance.dbg( LogAreaClassConnectionBOSH, "Send buffer is empty, sending empty request" );
233  }
234 
235  ++m_rid;
236 
237  std::string requestBody = "<body rid='" + util::int2string( m_rid ) + "' ";
238  requestBody += "sid='" + m_sid + "' ";
239  requestBody += "xmlns='" + XMLNS_HTTPBIND + "'";
240 
241  if( m_streamRestart )
242  {
243  requestBody += " xmpp:restart='true' to='" + m_server + "' xml:lang='en' xmlns:xmpp='"
244  + XMLNS_XMPP_BOSH + "' />";
245  m_logInstance.dbg( LogAreaClassConnectionBOSH, "Restarting stream" );
246  }
247  else
248  {
249  requestBody += ">" + m_sendBuffer + "</body>";
250  }
251  // Send a request. Force if we are not sending an empty request, or if there are no connections open
252  if( sendRequest( requestBody ) )
253  {
254  m_logInstance.dbg( LogAreaClassConnectionBOSH, "Successfully sent m_sendBuffer" );
255  m_sendBuffer = EmptyString;
256  m_streamRestart = false;
257  }
258  else
259  {
260  --m_rid; // I think... (may need to rethink when acks are implemented)
261  m_logInstance.warn( LogAreaClassConnectionBOSH,
262  "Unable to send. Connection not complete, or too many open requests,"
263  " so added to buffer.\n" );
264  }
265 
266  return true;
267  }
268 
269  /* Chooses the appropriate connection, or opens a new one if necessary. Wraps xml in HTTP and sends. */
270  bool ConnectionBOSH::sendRequest( const std::string& xml )
271  {
272  ConnectionBase* conn = getConnection();
273  if( !conn )
274  return false;
275 
276  std::string request = "POST " + m_path;
277  if( m_connMode == ModeLegacyHTTP )
278  {
279  request += " HTTP/1.0\r\n";
280  request += "Connection: close\r\n";
281  }
282  else
283  request += " HTTP/1.1\r\n";
284 
285  request += "Host: " + m_boshedHost + "\r\n";
286  request += "Content-Type: text/xml; charset=utf-8\r\n";
287  request += "Content-Length: " + util::int2string( xml.length() ) + "\r\n";
288  request += "User-Agent: gloox/" + GLOOX_VERSION + "\r\n\r\n";
289  request += xml;
290 
291 
292  if( conn->send( request ) )
293  {
294  m_lastRequestTime = time( 0 );
295  ++m_openRequests;
296  return true;
297  }
298 // else // FIXME What to do in this case?
299 // printf( "Error while trying to send on socket (state: %d)\n", conn->state() );
300 
301  return false;
302  }
303 
304  bool ci_equal( char ch1, char ch2 )
305  {
306  return std::toupper( (unsigned char)ch1 ) == std::toupper( (unsigned char)ch2 );
307  }
308 
309  std::string::size_type ci_find( const std::string& str1, const std::string& str2 )
310  {
311  std::string::const_iterator pos = std::search( str1.begin(), str1.end(),
312  str2.begin(), str2.end(), ci_equal );
313  if( pos == str1.end() )
314  return std::string::npos;
315  else
316  return std::distance( str1.begin(), pos );
317  }
318 
319  const std::string ConnectionBOSH::getHTTPField( const std::string& field )
320  {
321  std::string::size_type fp = ci_find( m_bufferHeader, "\r\n" + field + ": " );
322 
323  if( fp == std::string::npos )
324  return EmptyString;
325 
326  fp += field.length() + 4;
327 
328  const std::string::size_type fp2 = m_bufferHeader.find( "\r\n", fp );
329  if( fp2 == std::string::npos )
330  return EmptyString;
331 
332  return m_bufferHeader.substr( fp, fp2 - fp );
333  }
334 
335  ConnectionError ConnectionBOSH::receive()
336  {
338  while( m_state != StateDisconnected && ( err = recv( 10 ) ) == ConnNoError )
339  ;
340  return err == ConnNoError ? ConnNotConnected : err;
341  }
342 
343  void ConnectionBOSH::cleanup()
344  {
346 
347  util::ForEach( m_activeConnections, &ConnectionBase::cleanup );
348  util::ForEach( m_connectionPool, &ConnectionBase::cleanup );
349  }
350 
351  void ConnectionBOSH::getStatistics( long int& totalIn, long int& totalOut )
352  {
353  util::ForEach( m_activeConnections, &ConnectionBase::getStatistics, totalIn, totalOut );
354  util::ForEach( m_connectionPool, &ConnectionBase::getStatistics, totalIn, totalOut );
355  }
356 
357  void ConnectionBOSH::handleReceivedData( const ConnectionBase* /*connection*/,
358  const std::string& data )
359  {
360  m_buffer += data;
361  std::string::size_type headerLength = 0;
362  while( ( headerLength = m_buffer.find( "\r\n\r\n" ) ) != std::string::npos )
363  {
364  m_bufferHeader = m_buffer.substr( 0, headerLength+2 );
365 
366  const std::string& statusCode = m_bufferHeader.substr( 9, 3 );
367  if( statusCode != "200" )
368  {
369  m_logInstance.warn( LogAreaClassConnectionBOSH,
370  "Received error via legacy HTTP status code: " + statusCode
371  + ". Disconnecting." );
372  m_state = StateDisconnected; // As per XEP, consider connection broken
373  disconnect();
374  }
375 
376  m_bufferContentLength = atol( getHTTPField( "Content-Length" ).c_str() );
377  if( !m_bufferContentLength )
378  return;
379 
380  if( m_connMode != ModeLegacyHTTP && ( getHTTPField( "Connection" ) == "close"
381  || m_bufferHeader.substr( 0, 8 ) == "HTTP/1.0" ) )
382  {
383  m_logInstance.dbg( LogAreaClassConnectionBOSH,
384  "Server indicated lack of support for HTTP/1.1 - falling back to HTTP/1.0" );
385  m_connMode = ModeLegacyHTTP;
386  }
387 
388  if( m_buffer.length() >= ( headerLength + 4 + m_bufferContentLength ) )
389  {
390  putConnection();
391  --m_openRequests;
392  std::string xml = m_buffer.substr( headerLength + 4, m_bufferContentLength );
393  m_parser.feed( xml );
394  m_buffer.erase( 0, headerLength + 4 + m_bufferContentLength );
395  m_bufferContentLength = 0;
396  m_bufferHeader = EmptyString;
397  }
398  else
399  {
400  m_logInstance.warn( LogAreaClassConnectionBOSH, "buffer length mismatch" );
401  break;
402  }
403  }
404  }
405 
406  void ConnectionBOSH::handleConnect( const ConnectionBase* /*connection*/ )
407  {
408  if( m_state == StateConnecting )
409  {
410  m_rid = rand() % 100000 + 1728679472;
411 
412  Tag requestBody( "body" );
413  requestBody.setXmlns( XMLNS_HTTPBIND );
414  requestBody.setXmlns( XMLNS_XMPP_BOSH, "xmpp" );
415 
416  requestBody.addAttribute( "content", "text/xml; charset=utf-8" );
417  requestBody.addAttribute( "hold", (long)m_hold );
418  requestBody.addAttribute( "rid", (long)m_rid );
419  requestBody.addAttribute( "ver", "1.6" );
420  requestBody.addAttribute( "wait", (long)m_wait );
421  requestBody.addAttribute( "ack", 0 );
422  requestBody.addAttribute( "secure", "false" );
423  requestBody.addAttribute( "route", "xmpp:" + m_server + ":5222" );
424  requestBody.addAttribute( "xml:lang", "en" );
425  requestBody.addAttribute( "xmpp:version", "1.0" );
426  requestBody.addAttribute( "to", m_server );
427 
428  m_logInstance.dbg( LogAreaClassConnectionBOSH, "sending bosh connection request" );
429  sendRequest( requestBody.xml() );
430  }
431  }
432 
433  void ConnectionBOSH::handleDisconnect( const ConnectionBase* /*connection*/,
434  ConnectionError reason )
435  {
436  if( m_handler && m_state == StateConnecting )
437  {
439  m_handler->handleDisconnect( this, reason );
440  return;
441  }
442 
443  switch( m_connMode ) // FIXME avoid that if we're disconnecting on purpose
444  {
445  case ModePipelining:
446  m_connMode = ModeLegacyHTTP; // Server seems not to support pipelining
447  m_logInstance.dbg( LogAreaClassConnectionBOSH,
448  "connection closed - falling back to HTTP/1.0 connection method" );
449  break;
450  case ModeLegacyHTTP:
451  case ModePersistentHTTP:
452  // FIXME do we need to do anything here?
453 // printf( "A TCP connection %p was disconnected (reason: %d).\n", connection, reason );
454  break;
455  }
456  }
457 
458  void ConnectionBOSH::handleTag( Tag* tag )
459  {
460  if( !m_handler || tag->name() != "body" )
461  return;
462 
463  if( m_streamRestart )
464  {
465  m_streamRestart = false;
466  m_logInstance.dbg( LogAreaClassConnectionBOSH, "sending spoofed <stream:stream>" );
467  m_handler->handleReceivedData( this, "<?xml version='1.0' ?>"
468  "<stream:stream xmlns:stream='http://etherx.jabber.org/streams'"
469  " xmlns='" + XMLNS_CLIENT + "' version='" + XMPP_STREAM_VERSION_MAJOR
470  + "." + XMPP_STREAM_VERSION_MINOR + "' from='" + m_server + "' id ='"
471  + m_sid + "' xml:lang='en'>" );
472  }
473 
474  if( tag->hasAttribute( "sid" ) )
475  {
477  m_sid = tag->findAttribute( "sid" );
478 
479  if( tag->hasAttribute( "requests" ) )
480  {
481  const int serverRequests = atoi( tag->findAttribute( "requests" ).c_str() );
482  if( serverRequests < m_maxOpenRequests )
483  {
484  m_maxOpenRequests = serverRequests;
485  m_logInstance.dbg( LogAreaClassConnectionBOSH,
486  "bosh parameter 'requests' now set to " + tag->findAttribute( "requests" ) );
487  }
488  }
489  if( tag->hasAttribute( "hold" ) )
490  {
491  const int maxHold = atoi( tag->findAttribute( "hold" ).c_str() );
492  if( maxHold < m_hold )
493  {
494  m_hold = maxHold;
495  m_logInstance.dbg( LogAreaClassConnectionBOSH,
496  "bosh parameter 'hold' now set to " + tag->findAttribute( "hold" ) );
497  }
498  }
499  if( tag->hasAttribute( "wait" ) )
500  {
501  const int maxWait = atoi( tag->findAttribute( "wait" ).c_str() );
502  if( maxWait < m_wait )
503  {
504  m_wait = maxWait;
505  m_logInstance.dbg( LogAreaClassConnectionBOSH,
506  "bosh parameter 'wait' now set to " + tag->findAttribute( "wait" )
507  + " seconds" );
508  }
509  }
510  if( tag->hasAttribute( "polling" ) )
511  {
512  const int minTime = atoi( tag->findAttribute( "polling" ).c_str() );
513  m_minTimePerRequest = minTime;
514  m_logInstance.dbg( LogAreaClassConnectionBOSH,
515  "bosh parameter 'polling' now set to " + tag->findAttribute( "polling" )
516  + " seconds" );
517  }
518 
519  if( m_state < StateConnected )
520  m_handler->handleConnect( this );
521 
522  m_handler->handleReceivedData( this, "<?xml version='1.0' ?>" // FIXME move to send() so that
523  // it is more clearly a response
524  // to the initial stream opener?
525  "<stream:stream xmlns:stream='http://etherx.jabber.org/streams' "
526  "xmlns='" + XMLNS_CLIENT
527  + "' version='" + XMPP_STREAM_VERSION_MAJOR + "." + XMPP_STREAM_VERSION_MINOR
528  + "' from='" + m_server + "' id ='" + m_sid + "' xml:lang='en'>" );
529  }
530 
531  if( tag->findAttribute( "type" ) == "terminate" )
532  {
533  m_logInstance.dbg( LogAreaClassConnectionBOSH,
534  "bosh connection closed by server: " + tag->findAttribute( "condition" ) );
537  return;
538  }
539 
540  const TagList& stanzas = tag->children();
541  TagList::const_iterator it = stanzas.begin();
542  for( ; it != stanzas.end(); ++it )
543  m_handler->handleReceivedData( this, (*it)->xml() );
544  }
545 
546  ConnectionBase* ConnectionBOSH::getConnection()
547  {
548  if( m_openRequests > 0 && m_openRequests >= m_maxOpenRequests )
549  {
550  m_logInstance.warn( LogAreaClassConnectionBOSH,
551  "Too many requests already open. Cannot send." );
552  return 0;
553  }
554 
555  ConnectionBase* conn = 0;
556  switch( m_connMode )
557  {
558  case ModePipelining:
559  if( !m_activeConnections.empty() )
560  {
561  m_logInstance.dbg( LogAreaClassConnectionBOSH, "Using default connection for Pipelining." );
562  return m_activeConnections.front();
563  }
564  else if( !m_connectionPool.empty() )
565  {
566  m_logInstance.warn( LogAreaClassConnectionBOSH,
567  "Pipelining selected, but no connection open. Opening one." );
568  return activateConnection();
569  }
570  else
571  m_logInstance.warn( LogAreaClassConnectionBOSH,
572  "No available connections to pipeline on." );
573  break;
574  case ModeLegacyHTTP:
575  case ModePersistentHTTP:
576  {
577  if( !m_connectionPool.empty() )
578  {
579  m_logInstance.dbg( LogAreaClassConnectionBOSH, "LegacyHTTP/PersistentHTTP selected, "
580  "using connection from pool." );
581  return activateConnection();
582  }
583  else if( !m_activeConnections.empty() )
584  {
585  m_logInstance.dbg( LogAreaClassConnectionBOSH, "No connections in pool, creating a new one." );
586  conn = m_activeConnections.front()->newInstance();
587  conn->registerConnectionDataHandler( this );
588  m_connectionPool.push_back( conn );
589  conn->connect();
590  }
591  else
592  m_logInstance.warn( LogAreaClassConnectionBOSH,
593  "No available connections to send on." );
594  break;
595  }
596  }
597  return 0;
598  }
599 
600  ConnectionBase* ConnectionBOSH::activateConnection()
601  {
602  ConnectionBase* conn = m_connectionPool.front();
603  m_connectionPool.pop_front();
604  if( conn->state() == StateConnected )
605  {
606  m_activeConnections.push_back( conn );
607  return conn;
608  }
609 
610  m_logInstance.dbg( LogAreaClassConnectionBOSH, "Connecting pooled connection." );
611  m_connectionPool.push_back( conn );
612  conn->connect();
613  return 0;
614  }
615 
616  void ConnectionBOSH::putConnection()
617  {
618  ConnectionBase* conn = m_activeConnections.front();
619 
620  switch( m_connMode )
621  {
622  case ModeLegacyHTTP:
623  m_logInstance.dbg( LogAreaClassConnectionBOSH, "Disconnecting LegacyHTTP connection" );
624  conn->disconnect();
625  conn->cleanup(); // This is necessary
626  m_activeConnections.pop_front();
627  m_connectionPool.push_back( conn );
628  break;
629  case ModePersistentHTTP:
630  m_logInstance.dbg( LogAreaClassConnectionBOSH, "Deactivating PersistentHTTP connection" );
631  m_activeConnections.pop_front();
632  m_connectionPool.push_back( conn );
633  break;
634  case ModePipelining:
635  m_logInstance.dbg( LogAreaClassConnectionBOSH, "Keeping Pipelining connection" );
636  default:
637  break;
638  }
639  }
640 
641 }