From 5efb170e1d7d210304d5bc31f343743770d82e13 Mon Sep 17 00:00:00 2001 From: Lance Stout Date: Mon, 4 Jul 2011 18:47:57 -0700 Subject: Added wait parameter to disconnect. If wait=True, then the disconnect call will block until the send queue has emptied. WARNING: Using wait=True when more stanzas are being added to the queue than can be processed such that the queue is never empty will cause the disconnect call to block indefinitely without actually disconnecting. --- sleekxmpp/xmlstream/xmlstream.py | 19 ++++++++++++++++--- 1 file changed, 16 insertions(+), 3 deletions(-) (limited to 'sleekxmpp/xmlstream/xmlstream.py') diff --git a/sleekxmpp/xmlstream/xmlstream.py b/sleekxmpp/xmlstream/xmlstream.py index c7d0d3a8..8b6c08b7 100644 --- a/sleekxmpp/xmlstream/xmlstream.py +++ b/sleekxmpp/xmlstream/xmlstream.py @@ -356,22 +356,34 @@ class XMLStream(object): self.reconnect_delay = delay return False - def disconnect(self, reconnect=False): + def disconnect(self, reconnect=False, wait=False): """ Terminate processing and close the XML streams. Optionally, the connection may be reconnected and resume processing afterwards. + If the disconnect should take place after all items + in the send queue have been sent, use wait=True. However, + take note: If you are constantly adding items to the queue + such that it is never empty, then the disconnect will + not occur and the call will continue to block. + Arguments: reconnect -- Flag indicating if the connection and processing should be restarted. Defaults to False. + wait -- Flag indicating if the send queue should + be emptied before disconnecting. """ self.state.transition('connected', 'disconnected', wait=0.0, - func=self._disconnect, args=(reconnect,)) + func=self._disconnect, args=(reconnect, wait)) + + def _disconnect(self, reconnect=False, wait=False): + # Wait for the send queue to empty. + if wait: + self.send_queue.join() - def _disconnect(self, reconnect=False): # Send the end of stream marker. self.send_raw(self.stream_footer, now=True) self.session_started_event.clear() @@ -1036,6 +1048,7 @@ class XMLStream(object): log.debug("SEND: %s" % data) try: self.socket.send(data.encode('utf-8')) + self.send_queue.task_done() except Socket.error as serr: self.event('socket_error', serr) log.warning("Failed to send %s" % data) -- cgit v1.2.3 From e022b2a36c342b6a83d7c7f0a39dfb851cdfd2de Mon Sep 17 00:00:00 2001 From: Lance Stout Date: Wed, 27 Jul 2011 19:35:03 -0700 Subject: Add support for HTTP Proxy connections. --- sleekxmpp/xmlstream/xmlstream.py | 75 ++++++++++++++++++++++++++++++++++++++-- 1 file changed, 73 insertions(+), 2 deletions(-) (limited to 'sleekxmpp/xmlstream/xmlstream.py') diff --git a/sleekxmpp/xmlstream/xmlstream.py b/sleekxmpp/xmlstream/xmlstream.py index 8b6c08b7..15bbe655 100644 --- a/sleekxmpp/xmlstream/xmlstream.py +++ b/sleekxmpp/xmlstream/xmlstream.py @@ -8,6 +8,7 @@ from __future__ import with_statement, unicode_literals +import base64 import copy import logging import signal @@ -23,6 +24,7 @@ try: except ImportError: import Queue as queue +import sleekxmpp from sleekxmpp.thirdparty.statemachine import StateMachine from sleekxmpp.xmlstream import Scheduler, tostring from sleekxmpp.xmlstream.stanzabase import StanzaBase, ET @@ -107,7 +109,13 @@ class XMLStream(object): stream_header -- The closing tag of the stream's root element. use_ssl -- Flag indicating if SSL should be used. use_tls -- Flag indicating if TLS should be used. + use_proxy -- Flag indicating that an HTTP Proxy should be used. stop -- threading Event used to stop all threads. + proxy_config -- An optional dictionary with the following entries: + host -- The host offering proxy services. + port -- The port for the proxy service. + username -- Optional username for the proxy. + password -- Optional password for the proxy. auto_reconnect -- Flag to determine whether we auto reconnect. reconnect_max_delay -- Maximum time to delay between connection @@ -180,6 +188,9 @@ class XMLStream(object): self.use_ssl = False self.use_tls = False + self.use_proxy = False + + self.proxy_config = {} self.default_ns = '' self.stream_header = "" @@ -322,6 +333,12 @@ class XMLStream(object): log.debug('Waiting %s seconds before connecting.' % delay) time.sleep(delay) + if self.use_proxy: + connected = self._connect_proxy() + if not connected: + self.reconnect_delay = delay + return False + if self.use_ssl and self.ssl_support: log.debug("Socket Wrapped for SSL") if self.ca_certs is None: @@ -341,8 +358,10 @@ class XMLStream(object): self.socket = ssl_socket try: - log.debug("Connecting to %s:%s" % self.address) - self.socket.connect(self.address) + if not self.use_proxy: + log.debug("Connecting to %s:%s" % self.address) + self.socket.connect(self.address) + self.set_socket(self.socket, ignore=True) #this event is where you should set your application state self.event("connected", direct=True) @@ -356,6 +375,58 @@ class XMLStream(object): self.reconnect_delay = delay return False + def _connect_proxy(self): + """Attempt to connect using an HTTP Proxy.""" + + # Extract the proxy address, and optional credentials + address = (self.proxy_config['host'], int(self.proxy_config['port'])) + cred = None + if self.proxy_config['username']: + username = self.proxy_config['username'] + password = self.proxy_config['password'] + + cred = '%s:%s' % (username, password) + if sys.version_info < (3, 0): + cred = bytes(cred) + else: + cred = bytes(cred, 'utf-8') + cred = base64.b64encode(cred).decode('utf-8') + + # Build the HTTP headers for connecting to the XMPP server + headers = ['CONNECT %s:%s HTTP/1.0' % self.address, + 'Host: %s:%s' % self.address, + 'Proxy-Connection: Keep-Alive', + 'Pragma: no-cache', + 'User-Agent: SleekXMPP/%s' % sleekxmpp.__version__] + if cred: + headers.append('Proxy-Authorization: Basic %s' % cred) + headers = '\r\n'.join(headers) + '\r\n\r\n' + + try: + log.debug("Connecting to proxy: %s:%s" % address) + self.socket.connect(address) + self.send_raw(headers, now=True) + resp = '' + while '\r\n\r\n' not in resp: + resp += self.socket.recv(1024).decode('utf-8') + log.debug('RECV: %s' % resp) + + lines = resp.split('\r\n') + if '200' not in lines[0]: + self.event('proxy_error', resp) + log.error('Proxy Error: %s' % lines[0]) + return False + + # Proxy connection established, continue connecting + # with the XMPP server. + return True + except Socket.error as serr: + error_msg = "Could not connect to %s:%s. Socket Error #%s: %s" + self.event('socket_error', serr) + log.error(error_msg % (self.address[0], self.address[1], + serr.errno, serr.strerror)) + return False + def disconnect(self, reconnect=False, wait=False): """ Terminate processing and close the XML streams. -- cgit v1.2.3