From 765ee5db6fc39d77e55dcf4fe97fb96da2f46d30 Mon Sep 17 00:00:00 2001 From: Devendra Date: Wed, 2 Apr 2014 02:44:29 +0530 Subject: multiplexing support --- common/PubnubCoreAsync.py | 171 +++++++++++++++++++++++++++++++--------------- 1 file changed, 115 insertions(+), 56 deletions(-) (limited to 'common/PubnubCoreAsync.py') diff --git a/common/PubnubCoreAsync.py b/common/PubnubCoreAsync.py index a7fbb7d..0038243 100644 --- a/common/PubnubCoreAsync.py +++ b/common/PubnubCoreAsync.py @@ -10,8 +10,6 @@ class PubnubCoreAsync(PubnubBase): def start(self): pass def stop(self): pass - def timeout( self, delay, callback ): - pass def __init__( self, @@ -54,8 +52,23 @@ class PubnubCoreAsync(PubnubBase): self.timetoken = 0 self.version = '3.3.4' self.accept_encoding = 'gzip' - - def subscribe( self, args ) : + self.SUB_RECEIVER = None + self._connect = None + + def get_channel_list(self, channels): + channel = '' + first = True + for ch in channels: + if not channels[ch]['subscribed']: + continue + if not first: + channel += ',' + else: + first = False + channel += ch + return channel + + def subscribe( self, args=None, sync=False ) : """ #** #* Subscribe @@ -87,94 +100,140 @@ class PubnubCoreAsync(PubnubBase): }) """ - ## Fail if missing channel - if not 'channel' in args : - return 'Missing Channel.' - ## Fail if missing callback - if not 'callback' in args : - return 'Missing Callback.' + if sync is True and self.susbcribe_sync is not None: + self.susbcribe_sync(args) + return + + def _invoke(func,msg=None): + if func is not None: + if msg is not None: + func(msg) + else: + func() + + def _invoke_connect(): + for ch in self.subscriptions: + chobj = self.subscriptions[ch] + if chobj['connected'] is False: + chobj['connected'] = True + _invoke(chobj['connect']) + + def _invoke_error(err=None): + for ch in self.subscriptions: + chobj = self.subscriptions[ch] + _invoke(chobj.error,err) + - ## Capture User Input - channel = str(args['channel']) - callback = args['callback'] - connectcb = args['connect'] + if callback is None: + _invoke(error, "Callback Missing") + return + + if channel is None: + _invoke(error, "Channel Missing") + return + + def _get_channel(): + for ch in self.subscriptions: + chobj = self.subscriptions[ch] + if chobj['subscribed'] is True: + return chobj - if 'errorback' in args: - errorback = args['errorback'] - else: - errorback = lambda x: x ## New Channel? - if not (channel in self.subscriptions) : + if not channel in self.subscriptions: self.subscriptions[channel] = { - 'first' : False, - 'connected' : False, + 'name' : channel, + 'first' : False, + 'connected' : False, + 'subscribed' : True, + 'callback' : callback, + 'connect' : connect, + 'disconnect' : disconnect, + 'reconnect' : reconnect } - ## Ensure Single Connection + ## return if already connected to channel if self.subscriptions[channel]['connected'] : - return "Already Connected" + _invoke(error, "Already Connected") + return + - self.subscriptions[channel]['connected'] = 1 ## SUBSCRIPTION RECURSION - def _subscribe(): - ## STOP CONNECTION? - if not self.subscriptions[channel]['connected']: - return + def _connect(): + self._reset_offline() + def sub_callback(response): - if not self.subscriptions[channel]['first'] : - self.subscriptions[channel]['first'] = True - connectcb() + print response + ## ERROR ? + if not response or error in response: + _invoke_error() - ## STOP CONNECTION? - if not self.subscriptions[channel]['connected']: - return + _invoke_connect() + self.timetoken = response[1] - ## PROBLEM? - if not response: - def time_callback(_time): - if not _time: - self.timeout( 1, _subscribe ) - return errorback("Lost Network Connection") - else: - self.timeout( 1, _subscribe) + if len(response) > 2: + channel_list = response[2].split(',') + response_list = response[0] + for ch in enumerate(channel_list): + if ch[1] in self.subscriptions: + chobj = self.subscriptions[ch[1]] + _invoke(chobj['callback'],self.decrypt(response_list[ch[0]])) + else: + response_list = response[0] + chobj = _get_channel() + for r in response_list: + if chobj: + _invoke(chobj['callback'], self.decrypt(r)) - ## ENSURE CONNECTED (Call Time Function) - return self.time({ 'callback' : time_callback }) - self.timetoken = response[1] - _subscribe() + _connect() + - pc = PubnubCrypto() - out = [] - for message in response[0]: - callback(self.decrypt(message)) + channel_list = self.get_channel_list(self.subscriptions) + print channel_list ## CONNECT TO PUBNUB SUBSCRIBE SERVERS try: - self._request( { "urlcomponents" : [ + self.SUB_RECEIVER = self._request( { "urlcomponents" : [ 'subscribe', self.subscribe_key, - channel, + channel_list, '0', str(self.timetoken) - ], "urlparams" : {"uuid":self.uuid} }, sub_callback ) - except : - self.timeout( 1, _subscribe) + ], "urlparams" : {"uuid":self.uuid} }, sub_callback, single=True ) + except Exception as e: + self.timeout( 1, _connect) return + self._connect = _connect + + ## BEGIN SUBSCRIPTION (LISTEN FOR MESSAGES) - _subscribe() + _connect() + + def _reset_offline(self): + if self.SUB_RECEIVER is not None: + self.SUB_RECEIVER() + self.SUB_RECEIVER = None + + def CONNECT(self): + self._reset_offline() + self._connect() + + def unsubscribe( self, args ): + #print(args['channel']) channel = str(args['channel']) if not (channel in self.subscriptions): return False ## DISCONNECT self.subscriptions[channel]['connected'] = 0 + self.subscriptions[channel]['subscribed'] = False self.subscriptions[channel]['timetoken'] = 0 self.subscriptions[channel]['first'] = False + self.CONNECT() -- cgit v1.2.3 From 99096b8c11b9a541f6350639e8735495cf90091c Mon Sep 17 00:00:00 2001 From: Devendra Date: Fri, 11 Apr 2014 14:49:43 +0530 Subject: v1 MX and async code for python, twisted, tornado --- common/PubnubCoreAsync.py | 149 ++++++++++++++++++++++++++++++++-------------- 1 file changed, 103 insertions(+), 46 deletions(-) (limited to 'common/PubnubCoreAsync.py') diff --git a/common/PubnubCoreAsync.py b/common/PubnubCoreAsync.py index 0038243..4251d47 100644 --- a/common/PubnubCoreAsync.py +++ b/common/PubnubCoreAsync.py @@ -5,6 +5,9 @@ except ImportError: import Crypto.Hash.SHA256 as digestmod sha256 = digestmod.new import hmac +import threading +from threading import current_thread +import threading class PubnubCoreAsync(PubnubBase): @@ -17,6 +20,7 @@ class PubnubCoreAsync(PubnubBase): subscribe_key, secret_key = False, cipher_key = False, + auth_key = None, ssl_on = False, origin = 'pubsub.pubnub.com', uuid = None @@ -43,6 +47,7 @@ class PubnubCoreAsync(PubnubBase): subscribe_key=subscribe_key, secret_key=secret_key, cipher_key=cipher_key, + auth_key=auth_key, ssl_on=ssl_on, origin=origin, UUID=uuid @@ -50,22 +55,36 @@ class PubnubCoreAsync(PubnubBase): self.subscriptions = {} self.timetoken = 0 + self.last_timetoken = 0 self.version = '3.3.4' self.accept_encoding = 'gzip' self.SUB_RECEIVER = None self._connect = None + self._tt_lock = threading.RLock() def get_channel_list(self, channels): channel = '' first = True - for ch in channels: - if not channels[ch]['subscribed']: - continue - if not first: - channel += ',' - else: - first = False - channel += ch + if self._channel_list_lock: + with self._channel_list_lock: + for ch in channels: + if not channels[ch]['subscribed']: + continue + if not first: + channel += ',' + else: + first = False + channel += ch + else: + for ch in channels: + if not channels[ch]['subscribed']: + continue + if not first: + channel += ',' + else: + first = False + channel += ch + return channel def subscribe( self, args=None, sync=False ) : @@ -100,6 +119,26 @@ class PubnubCoreAsync(PubnubBase): }) """ + if args is None: + _invoke(error, "Arguments Missing") + return + channel = args['channel'] if 'channel' in args else None + callback = args['callback'] if 'callback' in args else None + connect = args['connect'] if 'connect' in args else None + disconnect = args['disconnect'] if 'disconnect' in args else None + reconnect = args['reconnect'] if 'reconnect' in args else None + error = args['error'] if 'error' in args else None + + with self._tt_lock: + self.last_timetoken = self.timetoken if self.timetoken != 0 else self.last_timetoken + self.timetoken = 0 + + if channel is None: + _invoke(error, "Channel Missing") + return + if callback is None: + _invoke(error, "Callback Missing") + return if sync is True and self.susbcribe_sync is not None: self.susbcribe_sync(args) @@ -113,18 +152,20 @@ class PubnubCoreAsync(PubnubBase): func() def _invoke_connect(): - for ch in self.subscriptions: - chobj = self.subscriptions[ch] - if chobj['connected'] is False: - chobj['connected'] = True - _invoke(chobj['connect']) + if self._channel_list_lock: + with self._channel_list_lock: + for ch in self.subscriptions: + chobj = self.subscriptions[ch] + if chobj['connected'] is False: + chobj['connected'] = True + _invoke(chobj['connect'],chobj['name']) def _invoke_error(err=None): for ch in self.subscriptions: chobj = self.subscriptions[ch] _invoke(chobj.error,err) - + ''' if callback is None: _invoke(error, "Callback Missing") return @@ -132,6 +173,7 @@ class PubnubCoreAsync(PubnubBase): if channel is None: _invoke(error, "Channel Missing") return + ''' def _get_channel(): for ch in self.subscriptions: @@ -142,22 +184,36 @@ class PubnubCoreAsync(PubnubBase): ## New Channel? if not channel in self.subscriptions: - self.subscriptions[channel] = { - 'name' : channel, - 'first' : False, - 'connected' : False, - 'subscribed' : True, - 'callback' : callback, - 'connect' : connect, - 'disconnect' : disconnect, - 'reconnect' : reconnect - } + if self._channel_list_lock: + with self._channel_list_lock: + self.subscriptions[channel] = { + 'name' : channel, + 'first' : False, + 'connected' : False, + 'subscribed' : True, + 'callback' : callback, + 'connect' : connect, + 'disconnect' : disconnect, + 'reconnect' : reconnect + } + else: + self.subscriptions[channel] = { + 'name' : channel, + 'first' : False, + 'connected' : False, + 'subscribed' : True, + 'callback' : callback, + 'connect' : connect, + 'disconnect' : disconnect, + 'reconnect' : reconnect + } ## return if already connected to channel - if self.subscriptions[channel]['connected'] : + if channel in self.subscriptions and 'connected' in self.subscriptions[channel] and self.subscriptions[channel]['connected'] is True: _invoke(error, "Already Connected") return + ## SUBSCRIPTION RECURSION def _connect(): @@ -165,37 +221,37 @@ class PubnubCoreAsync(PubnubBase): self._reset_offline() def sub_callback(response): - print response ## ERROR ? if not response or error in response: _invoke_error() _invoke_connect() - - self.timetoken = response[1] - - if len(response) > 2: - channel_list = response[2].split(',') - response_list = response[0] - for ch in enumerate(channel_list): - if ch[1] in self.subscriptions: - chobj = self.subscriptions[ch[1]] - _invoke(chobj['callback'],self.decrypt(response_list[ch[0]])) - else: - response_list = response[0] - chobj = _get_channel() - for r in response_list: - if chobj: - _invoke(chobj['callback'], self.decrypt(r)) - - - _connect() + with self._tt_lock: + #print 'A tt : ', self.timetoken , ' last tt : ' , self.last_timetoken + self.timetoken = self.last_timetoken if self.timetoken == 0 and self.last_timetoken != 0 else response[1] + #print 'B tt : ', self.timetoken , ' last tt : ' , self.last_timetoken + if len(response) > 2: + channel_list = response[2].split(',') + response_list = response[0] + for ch in enumerate(channel_list): + if ch[1] in self.subscriptions: + chobj = self.subscriptions[ch[1]] + _invoke(chobj['callback'],self.decrypt(response_list[ch[0]])) + else: + response_list = response[0] + chobj = _get_channel() + for r in response_list: + if chobj: + _invoke(chobj['callback'], self.decrypt(r)) + + #with self._tt_lock: + # self.timetoken = self.last_timetoken if self.timetoken == 0 and self.last_timetoken != 0 else response[1] + _connect() channel_list = self.get_channel_list(self.subscriptions) - print channel_list ## CONNECT TO PUBNUB SUBSCRIBE SERVERS try: self.SUB_RECEIVER = self._request( { "urlcomponents" : [ @@ -206,6 +262,7 @@ class PubnubCoreAsync(PubnubBase): str(self.timetoken) ], "urlparams" : {"uuid":self.uuid} }, sub_callback, single=True ) except Exception as e: + print e self.timeout( 1, _connect) return -- cgit v1.2.3 From 150ae1566d813acbb773839e919db2c0f467931c Mon Sep 17 00:00:00 2001 From: Devendra Date: Wed, 16 Apr 2014 00:00:40 +0530 Subject: adding code to support async and pam client capabilities with python v2 and v3 --- common/PubnubCoreAsync.py | 125 ++++++++++++++++++++++++---------------------- 1 file changed, 65 insertions(+), 60 deletions(-) (limited to 'common/PubnubCoreAsync.py') diff --git a/common/PubnubCoreAsync.py b/common/PubnubCoreAsync.py index 4251d47..f7b57cc 100644 --- a/common/PubnubCoreAsync.py +++ b/common/PubnubCoreAsync.py @@ -5,9 +5,14 @@ except ImportError: import Crypto.Hash.SHA256 as digestmod sha256 = digestmod.new import hmac -import threading -from threading import current_thread -import threading + +class EmptyLock(): + def __enter__(self): + pass + def __exit__(self,a,b,c): + pass + +empty_lock = EmptyLock() class PubnubCoreAsync(PubnubBase): @@ -23,7 +28,9 @@ class PubnubCoreAsync(PubnubBase): auth_key = None, ssl_on = False, origin = 'pubsub.pubnub.com', - uuid = None + uuid = None, + _tt_lock=empty_lock, + _channel_list_lock=empty_lock ) : """ #** @@ -53,29 +60,20 @@ class PubnubCoreAsync(PubnubBase): UUID=uuid ) - self.subscriptions = {} - self.timetoken = 0 - self.last_timetoken = 0 - self.version = '3.3.4' - self.accept_encoding = 'gzip' - self.SUB_RECEIVER = None - self._connect = None - self._tt_lock = threading.RLock() + self.subscriptions = {} + self.timetoken = 0 + self.last_timetoken = 0 + self.version = '3.3.4' + self.accept_encoding = 'gzip' + self.SUB_RECEIVER = None + self._connect = None + self._tt_lock = _tt_lock + self._channel_list_lock = _channel_list_lock def get_channel_list(self, channels): channel = '' first = True - if self._channel_list_lock: - with self._channel_list_lock: - for ch in channels: - if not channels[ch]['subscribed']: - continue - if not first: - channel += ',' - else: - first = False - channel += ch - else: + with self._channel_list_lock: for ch in channels: if not channels[ch]['subscribed']: continue @@ -84,9 +82,15 @@ class PubnubCoreAsync(PubnubBase): else: first = False channel += ch - return channel + + def each(l, func): + if func is None: + return + for i in l: + func(i) + def subscribe( self, args=None, sync=False ) : """ #** @@ -122,12 +126,12 @@ class PubnubCoreAsync(PubnubBase): if args is None: _invoke(error, "Arguments Missing") return - channel = args['channel'] if 'channel' in args else None - callback = args['callback'] if 'callback' in args else None - connect = args['connect'] if 'connect' in args else None - disconnect = args['disconnect'] if 'disconnect' in args else None - reconnect = args['reconnect'] if 'reconnect' in args else None - error = args['error'] if 'error' in args else None + channel = args['channel'] if 'channel' in args else None + callback = args['callback'] if 'callback' in args else None + connect = args['connect'] if 'connect' in args else None + disconnect = args['disconnect'] if 'disconnect' in args else None + reconnect = args['reconnect'] if 'reconnect' in args else None + error = args['error'] if 'error' in args else None with self._tt_lock: self.last_timetoken = self.timetoken if self.timetoken != 0 else self.last_timetoken @@ -160,10 +164,15 @@ class PubnubCoreAsync(PubnubBase): chobj['connected'] = True _invoke(chobj['connect'],chobj['name']) - def _invoke_error(err=None): - for ch in self.subscriptions: - chobj = self.subscriptions[ch] - _invoke(chobj.error,err) + def _invoke_error(channel_list=None, err=None): + if channel_list is None: + for ch in self.subscriptions: + chobj = self.subscriptions[ch] + _invoke(chobj['error'],err) + else: + for ch in channel_list: + chobj = self.subscriptions[ch] + _invoke(chobj['error'],err) ''' if callback is None: @@ -184,19 +193,7 @@ class PubnubCoreAsync(PubnubBase): ## New Channel? if not channel in self.subscriptions: - if self._channel_list_lock: - with self._channel_list_lock: - self.subscriptions[channel] = { - 'name' : channel, - 'first' : False, - 'connected' : False, - 'subscribed' : True, - 'callback' : callback, - 'connect' : connect, - 'disconnect' : disconnect, - 'reconnect' : reconnect - } - else: + with self._channel_list_lock: self.subscriptions[channel] = { 'name' : channel, 'first' : False, @@ -205,9 +202,11 @@ class PubnubCoreAsync(PubnubBase): 'callback' : callback, 'connect' : connect, 'disconnect' : disconnect, - 'reconnect' : reconnect + 'reconnect' : reconnect, + 'error' : error } + ## return if already connected to channel if channel in self.subscriptions and 'connected' in self.subscriptions[channel] and self.subscriptions[channel]['connected'] is True: _invoke(error, "Already Connected") @@ -222,8 +221,11 @@ class PubnubCoreAsync(PubnubBase): def sub_callback(response): ## ERROR ? - if not response or error in response: - _invoke_error() + #print response + if not response or ('message' in response and response['message'] == 'Forbidden'): + _invoke_error(response['payload']['channels'], response['message']) + _connect() + return _invoke_connect() @@ -250,7 +252,6 @@ class PubnubCoreAsync(PubnubBase): _connect() - channel_list = self.get_channel_list(self.subscriptions) ## CONNECT TO PUBNUB SUBSCRIBE SERVERS try: @@ -260,9 +261,9 @@ class PubnubCoreAsync(PubnubBase): channel_list, '0', str(self.timetoken) - ], "urlparams" : {"uuid":self.uuid} }, sub_callback, single=True ) + ], "urlparams" : {"uuid":self.uuid, "auth" : self.auth_key} }, sub_callback, sub_callback, single=True ) except Exception as e: - print e + print(e) self.timeout( 1, _connect) return @@ -283,14 +284,18 @@ class PubnubCoreAsync(PubnubBase): def unsubscribe( self, args ): - #print(args['channel']) - channel = str(args['channel']) - if not (channel in self.subscriptions): + + if 'channel' in self.subscriptions is False: return False + channel = str(args['channel']) + + ## DISCONNECT - self.subscriptions[channel]['connected'] = 0 - self.subscriptions[channel]['subscribed'] = False - self.subscriptions[channel]['timetoken'] = 0 - self.subscriptions[channel]['first'] = False + with self._channel_list_lock: + if channel in self.subscriptions: + self.subscriptions[channel]['connected'] = 0 + self.subscriptions[channel]['subscribed'] = False + self.subscriptions[channel]['timetoken'] = 0 + self.subscriptions[channel]['first'] = False self.CONNECT() -- cgit v1.2.3 From 4926061ebebbc060d14feac9c9d6b13880205724 Mon Sep 17 00:00:00 2001 From: Devendra Date: Tue, 22 Apr 2014 23:12:05 +0530 Subject: improvements to dev console --- common/PubnubCoreAsync.py | 20 +++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) (limited to 'common/PubnubCoreAsync.py') diff --git a/common/PubnubCoreAsync.py b/common/PubnubCoreAsync.py index f7b57cc..30d337c 100644 --- a/common/PubnubCoreAsync.py +++ b/common/PubnubCoreAsync.py @@ -84,6 +84,15 @@ class PubnubCoreAsync(PubnubBase): channel += ch return channel + def get_channel_array(self): + channels = self.subscriptions + channel = [] + with self._channel_list_lock: + for ch in channels: + if not channels[ch]['subscribed']: + continue + channel.append(ch) + return channel def each(l, func): if func is None: @@ -192,7 +201,7 @@ class PubnubCoreAsync(PubnubBase): ## New Channel? - if not channel in self.subscriptions: + if not channel in self.subscriptions or self.subscriptions[channel]['subscribed'] is False: with self._channel_list_lock: self.subscriptions[channel] = { 'name' : channel, @@ -221,7 +230,6 @@ class PubnubCoreAsync(PubnubBase): def sub_callback(response): ## ERROR ? - #print response if not response or ('message' in response and response['message'] == 'Forbidden'): _invoke_error(response['payload']['channels'], response['message']) _connect() @@ -230,9 +238,7 @@ class PubnubCoreAsync(PubnubBase): _invoke_connect() with self._tt_lock: - #print 'A tt : ', self.timetoken , ' last tt : ' , self.last_timetoken self.timetoken = self.last_timetoken if self.timetoken == 0 and self.last_timetoken != 0 else response[1] - #print 'B tt : ', self.timetoken , ' last tt : ' , self.last_timetoken if len(response) > 2: channel_list = response[2].split(',') response_list = response[0] @@ -247,12 +253,13 @@ class PubnubCoreAsync(PubnubBase): if chobj: _invoke(chobj['callback'], self.decrypt(r)) - #with self._tt_lock: - # self.timetoken = self.last_timetoken if self.timetoken == 0 and self.last_timetoken != 0 else response[1] _connect() channel_list = self.get_channel_list(self.subscriptions) + if len(channel_list) <= 0: + return + ## CONNECT TO PUBNUB SUBSCRIBE SERVERS try: self.SUB_RECEIVER = self._request( { "urlcomponents" : [ @@ -269,7 +276,6 @@ class PubnubCoreAsync(PubnubBase): self._connect = _connect - ## BEGIN SUBSCRIPTION (LISTEN FOR MESSAGES) _connect() -- cgit v1.2.3 From b2f1e0ed2337b61073a595eaf36afd718a21a3fe Mon Sep 17 00:00:00 2001 From: Devendra Date: Wed, 23 Apr 2014 02:10:55 +0530 Subject: fixing issues reported by Dara in dev console --- common/PubnubCoreAsync.py | 1 + 1 file changed, 1 insertion(+) (limited to 'common/PubnubCoreAsync.py') diff --git a/common/PubnubCoreAsync.py b/common/PubnubCoreAsync.py index 30d337c..db10575 100644 --- a/common/PubnubCoreAsync.py +++ b/common/PubnubCoreAsync.py @@ -69,6 +69,7 @@ class PubnubCoreAsync(PubnubBase): self._connect = None self._tt_lock = _tt_lock self._channel_list_lock = _channel_list_lock + self._connect = lambda: None def get_channel_list(self, channels): channel = '' -- cgit v1.2.3 From fdb46e56fa6794940f9fbe51a2863d58e927e655 Mon Sep 17 00:00:00 2001 From: Devendra Date: Wed, 23 Apr 2014 12:54:16 +0530 Subject: switching to positional arguments --- common/PubnubCoreAsync.py | 35 +++-------------------------------- 1 file changed, 3 insertions(+), 32 deletions(-) (limited to 'common/PubnubCoreAsync.py') diff --git a/common/PubnubCoreAsync.py b/common/PubnubCoreAsync.py index db10575..deb6038 100644 --- a/common/PubnubCoreAsync.py +++ b/common/PubnubCoreAsync.py @@ -101,7 +101,7 @@ class PubnubCoreAsync(PubnubBase): for i in l: func(i) - def subscribe( self, args=None, sync=False ) : + def subscribe( self, channel, callback, error=None, connect=None, disconnect=None, reconnect=None, sync=False ) : """ #** #* Subscribe @@ -133,27 +133,11 @@ class PubnubCoreAsync(PubnubBase): }) """ - if args is None: - _invoke(error, "Arguments Missing") - return - channel = args['channel'] if 'channel' in args else None - callback = args['callback'] if 'callback' in args else None - connect = args['connect'] if 'connect' in args else None - disconnect = args['disconnect'] if 'disconnect' in args else None - reconnect = args['reconnect'] if 'reconnect' in args else None - error = args['error'] if 'error' in args else None with self._tt_lock: self.last_timetoken = self.timetoken if self.timetoken != 0 else self.last_timetoken self.timetoken = 0 - if channel is None: - _invoke(error, "Channel Missing") - return - if callback is None: - _invoke(error, "Callback Missing") - return - if sync is True and self.susbcribe_sync is not None: self.susbcribe_sync(args) return @@ -184,16 +168,6 @@ class PubnubCoreAsync(PubnubBase): chobj = self.subscriptions[ch] _invoke(chobj['error'],err) - ''' - if callback is None: - _invoke(error, "Callback Missing") - return - - if channel is None: - _invoke(error, "Channel Missing") - return - ''' - def _get_channel(): for ch in self.subscriptions: chobj = self.subscriptions[ch] @@ -290,14 +264,11 @@ class PubnubCoreAsync(PubnubBase): self._connect() - def unsubscribe( self, args ): + def unsubscribe( self, channel ): - if 'channel' in self.subscriptions is False: + if channel in self.subscriptions is False: return False - channel = str(args['channel']) - - ## DISCONNECT with self._channel_list_lock: if channel in self.subscriptions: -- cgit v1.2.3 From 09cd0c015ae276aa849297a6a976065b2b3f247b Mon Sep 17 00:00:00 2001 From: Devendra Date: Wed, 23 Apr 2014 14:03:13 +0530 Subject: modifying code for pep 8 compliance --- common/PubnubCoreAsync.py | 146 +++++++++++++++++++++++++--------------------- 1 file changed, 80 insertions(+), 66 deletions(-) (limited to 'common/PubnubCoreAsync.py') diff --git a/common/PubnubCoreAsync.py b/common/PubnubCoreAsync.py index deb6038..de7627f 100644 --- a/common/PubnubCoreAsync.py +++ b/common/PubnubCoreAsync.py @@ -6,32 +6,38 @@ except ImportError: sha256 = digestmod.new import hmac + class EmptyLock(): def __enter__(self): pass - def __exit__(self,a,b,c): + + def __exit__(self, a, b, c): pass empty_lock = EmptyLock() + class PubnubCoreAsync(PubnubBase): - def start(self): pass - def stop(self): pass + def start(self): + pass + + def stop(self): + pass def __init__( self, publish_key, subscribe_key, - secret_key = False, - cipher_key = False, - auth_key = None, - ssl_on = False, - origin = 'pubsub.pubnub.com', - uuid = None, + secret_key=False, + cipher_key=False, + auth_key=None, + ssl_on=False, + origin='pubsub.pubnub.com', + uuid=None, _tt_lock=empty_lock, _channel_list_lock=empty_lock - ) : + ): """ #** #* Pubnub @@ -58,18 +64,18 @@ class PubnubCoreAsync(PubnubBase): ssl_on=ssl_on, origin=origin, UUID=uuid - ) - - self.subscriptions = {} - self.timetoken = 0 - self.last_timetoken = 0 - self.version = '3.3.4' - self.accept_encoding = 'gzip' - self.SUB_RECEIVER = None - self._connect = None - self._tt_lock = _tt_lock - self._channel_list_lock = _channel_list_lock - self._connect = lambda: None + ) + + self.subscriptions = {} + self.timetoken = 0 + self.last_timetoken = 0 + self.version = '3.3.4' + self.accept_encoding = 'gzip' + self.SUB_RECEIVER = None + self._connect = None + self._tt_lock = _tt_lock + self._channel_list_lock = _channel_list_lock + self._connect = lambda: None def get_channel_list(self, channels): channel = '' @@ -101,7 +107,8 @@ class PubnubCoreAsync(PubnubBase): for i in l: func(i) - def subscribe( self, channel, callback, error=None, connect=None, disconnect=None, reconnect=None, sync=False ) : + def subscribe(self, channel, callback, error=None, + connect=None, disconnect=None, reconnect=None, sync=False): """ #** #* Subscribe @@ -135,14 +142,15 @@ class PubnubCoreAsync(PubnubBase): """ with self._tt_lock: - self.last_timetoken = self.timetoken if self.timetoken != 0 else self.last_timetoken + self.last_timetoken = self.timetoken if self.timetoken != 0 \ + else self.last_timetoken self.timetoken = 0 if sync is True and self.susbcribe_sync is not None: self.susbcribe_sync(args) return - def _invoke(func,msg=None): + def _invoke(func, msg=None): if func is not None: if msg is not None: func(msg) @@ -156,17 +164,17 @@ class PubnubCoreAsync(PubnubBase): chobj = self.subscriptions[ch] if chobj['connected'] is False: chobj['connected'] = True - _invoke(chobj['connect'],chobj['name']) + _invoke(chobj['connect'], chobj['name']) def _invoke_error(channel_list=None, err=None): if channel_list is None: for ch in self.subscriptions: chobj = self.subscriptions[ch] - _invoke(chobj['error'],err) + _invoke(chobj['error'], err) else: for ch in channel_list: chobj = self.subscriptions[ch] - _invoke(chobj['error'],err) + _invoke(chobj['error'], err) def _get_channel(): for ch in self.subscriptions: @@ -174,53 +182,58 @@ class PubnubCoreAsync(PubnubBase): if chobj['subscribed'] is True: return chobj - ## New Channel? - if not channel in self.subscriptions or self.subscriptions[channel]['subscribed'] is False: - with self._channel_list_lock: - self.subscriptions[channel] = { - 'name' : channel, - 'first' : False, - 'connected' : False, - 'subscribed' : True, - 'callback' : callback, - 'connect' : connect, - 'disconnect' : disconnect, - 'reconnect' : reconnect, - 'error' : error - } - + if not channel in self.subscriptions or \ + self.subscriptions[channel]['subscribed'] is False: + with self._channel_list_lock: + self.subscriptions[channel] = { + 'name': channel, + 'first': False, + 'connected': False, + 'subscribed': True, + 'callback': callback, + 'connect': connect, + 'disconnect': disconnect, + 'reconnect': reconnect, + 'error': error + } ## return if already connected to channel - if channel in self.subscriptions and 'connected' in self.subscriptions[channel] and self.subscriptions[channel]['connected'] is True: - _invoke(error, "Already Connected") - return - - + if channel in self.subscriptions and \ + 'connected' in self.subscriptions[channel] and \ + self.subscriptions[channel]['connected'] is True: + _invoke(error, "Already Connected") + return - ## SUBSCRIPTION RECURSION + ## SUBSCRIPTION RECURSION def _connect(): - + self._reset_offline() def sub_callback(response): ## ERROR ? - if not response or ('message' in response and response['message'] == 'Forbidden'): - _invoke_error(response['payload']['channels'], response['message']) - _connect() - return + if not response or \ + ('message' in response and + response['message'] == 'Forbidden'): + _invoke_error(response['payload'][ + 'channels'], response['message']) + _connect() + return _invoke_connect() with self._tt_lock: - self.timetoken = self.last_timetoken if self.timetoken == 0 and self.last_timetoken != 0 else response[1] + self.timetoken = \ + self.last_timetoken if self.timetoken == 0 and \ + self.last_timetoken != 0 else response[1] if len(response) > 2: channel_list = response[2].split(',') response_list = response[0] for ch in enumerate(channel_list): if ch[1] in self.subscriptions: chobj = self.subscriptions[ch[1]] - _invoke(chobj['callback'],self.decrypt(response_list[ch[0]])) + _invoke(chobj['callback'], + self.decrypt(response_list[ch[0]])) else: response_list = response[0] chobj = _get_channel() @@ -230,23 +243,25 @@ class PubnubCoreAsync(PubnubBase): _connect() - channel_list = self.get_channel_list(self.subscriptions) if len(channel_list) <= 0: return ## CONNECT TO PUBNUB SUBSCRIBE SERVERS try: - self.SUB_RECEIVER = self._request( { "urlcomponents" : [ + self.SUB_RECEIVER = self._request({"urlcomponents": [ 'subscribe', self.subscribe_key, channel_list, '0', str(self.timetoken) - ], "urlparams" : {"uuid":self.uuid, "auth" : self.auth_key} }, sub_callback, sub_callback, single=True ) + ], "urlparams": {"uuid": self.uuid, "auth": self.auth_key}}, + sub_callback, + sub_callback, + single=True) except Exception as e: print(e) - self.timeout( 1, _connect) + self.timeout(1, _connect) return self._connect = _connect @@ -263,8 +278,7 @@ class PubnubCoreAsync(PubnubBase): self._reset_offline() self._connect() - - def unsubscribe( self, channel ): + def unsubscribe(self, channel): if channel in self.subscriptions is False: return False @@ -272,8 +286,8 @@ class PubnubCoreAsync(PubnubBase): ## DISCONNECT with self._channel_list_lock: if channel in self.subscriptions: - self.subscriptions[channel]['connected'] = 0 - self.subscriptions[channel]['subscribed'] = False - self.subscriptions[channel]['timetoken'] = 0 - self.subscriptions[channel]['first'] = False + self.subscriptions[channel]['connected'] = 0 + self.subscriptions[channel]['subscribed'] = False + self.subscriptions[channel]['timetoken'] = 0 + self.subscriptions[channel]['first'] = False self.CONNECT() -- cgit v1.2.3 From f7b89bfafae34fa22509c1d1c59d1284ec62c5df Mon Sep 17 00:00:00 2001 From: Devendra Date: Wed, 23 Apr 2014 21:35:06 +0530 Subject: exception handling changes --- common/PubnubCoreAsync.py | 24 +++++++++++++----------- 1 file changed, 13 insertions(+), 11 deletions(-) (limited to 'common/PubnubCoreAsync.py') diff --git a/common/PubnubCoreAsync.py b/common/PubnubCoreAsync.py index de7627f..f8e9e68 100644 --- a/common/PubnubCoreAsync.py +++ b/common/PubnubCoreAsync.py @@ -248,21 +248,23 @@ class PubnubCoreAsync(PubnubBase): return ## CONNECT TO PUBNUB SUBSCRIBE SERVERS - try: - self.SUB_RECEIVER = self._request({"urlcomponents": [ - 'subscribe', - self.subscribe_key, - channel_list, - '0', - str(self.timetoken) - ], "urlparams": {"uuid": self.uuid, "auth": self.auth_key}}, - sub_callback, - sub_callback, - single=True) + #try: + self.SUB_RECEIVER = self._request({"urlcomponents": [ + 'subscribe', + self.subscribe_key, + channel_list, + '0', + str(self.timetoken) + ], "urlparams": {"uuid": self.uuid, "auth": self.auth_key}}, + sub_callback, + sub_callback, + single=True) + ''' except Exception as e: print(e) self.timeout(1, _connect) return + ''' self._connect = _connect -- cgit v1.2.3 From 493e29a108255eb3ae3166dc920f40e2f4e5c4c4 Mon Sep 17 00:00:00 2001 From: Devendra Date: Thu, 24 Apr 2014 00:16:57 +0530 Subject: adding single file for all platforms --- common/PubnubCoreAsync.py | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) (limited to 'common/PubnubCoreAsync.py') diff --git a/common/PubnubCoreAsync.py b/common/PubnubCoreAsync.py index f8e9e68..c9a44c3 100644 --- a/common/PubnubCoreAsync.py +++ b/common/PubnubCoreAsync.py @@ -210,6 +210,19 @@ class PubnubCoreAsync(PubnubBase): self._reset_offline() + def error_callback(response): + ## ERROR ? + if not response or \ + ('message' in response and + response['message'] == 'Forbidden'): + _invoke_error(response['payload'][ + 'channels'], response['message']) + _connect() + return + if 'message' in response: + _invoke_error(err=response['message']) + + def sub_callback(response): ## ERROR ? if not response or \ @@ -257,7 +270,7 @@ class PubnubCoreAsync(PubnubBase): str(self.timetoken) ], "urlparams": {"uuid": self.uuid, "auth": self.auth_key}}, sub_callback, - sub_callback, + error_callback, single=True) ''' except Exception as e: -- cgit v1.2.3 From 98242257846e9276dd83adbea950f83e25a4f2b6 Mon Sep 17 00:00:00 2001 From: Devendra Date: Thu, 24 Apr 2014 02:07:56 +0530 Subject: pep8 compliance, removing redundant files --- common/PubnubCoreAsync.py | 308 ---------------------------------------------- 1 file changed, 308 deletions(-) delete mode 100644 common/PubnubCoreAsync.py (limited to 'common/PubnubCoreAsync.py') diff --git a/common/PubnubCoreAsync.py b/common/PubnubCoreAsync.py deleted file mode 100644 index c9a44c3..0000000 --- a/common/PubnubCoreAsync.py +++ /dev/null @@ -1,308 +0,0 @@ -try: - from hashlib import sha256 - digestmod = sha256 -except ImportError: - import Crypto.Hash.SHA256 as digestmod - sha256 = digestmod.new -import hmac - - -class EmptyLock(): - def __enter__(self): - pass - - def __exit__(self, a, b, c): - pass - -empty_lock = EmptyLock() - - -class PubnubCoreAsync(PubnubBase): - - def start(self): - pass - - def stop(self): - pass - - def __init__( - self, - publish_key, - subscribe_key, - secret_key=False, - cipher_key=False, - auth_key=None, - ssl_on=False, - origin='pubsub.pubnub.com', - uuid=None, - _tt_lock=empty_lock, - _channel_list_lock=empty_lock - ): - """ - #** - #* Pubnub - #* - #* Init the Pubnub Client API - #* - #* @param string publish_key required key to send messages. - #* @param string subscribe_key required key to receive messages. - #* @param string secret_key required key to sign messages. - #* @param boolean ssl required for 2048 bit encrypted messages. - #* @param string origin PUBNUB Server Origin. - #** - - ## Initiat Class - pubnub = Pubnub( 'PUBLISH-KEY', 'SUBSCRIBE-KEY', 'SECRET-KEY', False ) - - """ - super(PubnubCoreAsync, self).__init__( - publish_key=publish_key, - subscribe_key=subscribe_key, - secret_key=secret_key, - cipher_key=cipher_key, - auth_key=auth_key, - ssl_on=ssl_on, - origin=origin, - UUID=uuid - ) - - self.subscriptions = {} - self.timetoken = 0 - self.last_timetoken = 0 - self.version = '3.3.4' - self.accept_encoding = 'gzip' - self.SUB_RECEIVER = None - self._connect = None - self._tt_lock = _tt_lock - self._channel_list_lock = _channel_list_lock - self._connect = lambda: None - - def get_channel_list(self, channels): - channel = '' - first = True - with self._channel_list_lock: - for ch in channels: - if not channels[ch]['subscribed']: - continue - if not first: - channel += ',' - else: - first = False - channel += ch - return channel - - def get_channel_array(self): - channels = self.subscriptions - channel = [] - with self._channel_list_lock: - for ch in channels: - if not channels[ch]['subscribed']: - continue - channel.append(ch) - return channel - - def each(l, func): - if func is None: - return - for i in l: - func(i) - - def subscribe(self, channel, callback, error=None, - connect=None, disconnect=None, reconnect=None, sync=False): - """ - #** - #* Subscribe - #* - #* This is NON-BLOCKING. - #* Listen for a message on a channel. - #* - #* @param array args with channel and message. - #* @return false on fail, array on success. - #** - - ## Subscribe Example - def receive(message) : - print(message) - return True - - ## On Connect Callback - def connected() : - pubnub.publish({ - 'channel' : 'hello_world', - 'message' : { 'some_var' : 'text' } - }) - - ## Subscribe - pubnub.subscribe({ - 'channel' : 'hello_world', - 'connect' : connected, - 'callback' : receive - }) - - """ - - with self._tt_lock: - self.last_timetoken = self.timetoken if self.timetoken != 0 \ - else self.last_timetoken - self.timetoken = 0 - - if sync is True and self.susbcribe_sync is not None: - self.susbcribe_sync(args) - return - - def _invoke(func, msg=None): - if func is not None: - if msg is not None: - func(msg) - else: - func() - - def _invoke_connect(): - if self._channel_list_lock: - with self._channel_list_lock: - for ch in self.subscriptions: - chobj = self.subscriptions[ch] - if chobj['connected'] is False: - chobj['connected'] = True - _invoke(chobj['connect'], chobj['name']) - - def _invoke_error(channel_list=None, err=None): - if channel_list is None: - for ch in self.subscriptions: - chobj = self.subscriptions[ch] - _invoke(chobj['error'], err) - else: - for ch in channel_list: - chobj = self.subscriptions[ch] - _invoke(chobj['error'], err) - - def _get_channel(): - for ch in self.subscriptions: - chobj = self.subscriptions[ch] - if chobj['subscribed'] is True: - return chobj - - ## New Channel? - if not channel in self.subscriptions or \ - self.subscriptions[channel]['subscribed'] is False: - with self._channel_list_lock: - self.subscriptions[channel] = { - 'name': channel, - 'first': False, - 'connected': False, - 'subscribed': True, - 'callback': callback, - 'connect': connect, - 'disconnect': disconnect, - 'reconnect': reconnect, - 'error': error - } - - ## return if already connected to channel - if channel in self.subscriptions and \ - 'connected' in self.subscriptions[channel] and \ - self.subscriptions[channel]['connected'] is True: - _invoke(error, "Already Connected") - return - - ## SUBSCRIPTION RECURSION - def _connect(): - - self._reset_offline() - - def error_callback(response): - ## ERROR ? - if not response or \ - ('message' in response and - response['message'] == 'Forbidden'): - _invoke_error(response['payload'][ - 'channels'], response['message']) - _connect() - return - if 'message' in response: - _invoke_error(err=response['message']) - - - def sub_callback(response): - ## ERROR ? - if not response or \ - ('message' in response and - response['message'] == 'Forbidden'): - _invoke_error(response['payload'][ - 'channels'], response['message']) - _connect() - return - - _invoke_connect() - - with self._tt_lock: - self.timetoken = \ - self.last_timetoken if self.timetoken == 0 and \ - self.last_timetoken != 0 else response[1] - if len(response) > 2: - channel_list = response[2].split(',') - response_list = response[0] - for ch in enumerate(channel_list): - if ch[1] in self.subscriptions: - chobj = self.subscriptions[ch[1]] - _invoke(chobj['callback'], - self.decrypt(response_list[ch[0]])) - else: - response_list = response[0] - chobj = _get_channel() - for r in response_list: - if chobj: - _invoke(chobj['callback'], self.decrypt(r)) - - _connect() - - channel_list = self.get_channel_list(self.subscriptions) - if len(channel_list) <= 0: - return - - ## CONNECT TO PUBNUB SUBSCRIBE SERVERS - #try: - self.SUB_RECEIVER = self._request({"urlcomponents": [ - 'subscribe', - self.subscribe_key, - channel_list, - '0', - str(self.timetoken) - ], "urlparams": {"uuid": self.uuid, "auth": self.auth_key}}, - sub_callback, - error_callback, - single=True) - ''' - except Exception as e: - print(e) - self.timeout(1, _connect) - return - ''' - - self._connect = _connect - - ## BEGIN SUBSCRIPTION (LISTEN FOR MESSAGES) - _connect() - - def _reset_offline(self): - if self.SUB_RECEIVER is not None: - self.SUB_RECEIVER() - self.SUB_RECEIVER = None - - def CONNECT(self): - self._reset_offline() - self._connect() - - def unsubscribe(self, channel): - - if channel in self.subscriptions is False: - return False - - ## DISCONNECT - with self._channel_list_lock: - if channel in self.subscriptions: - self.subscriptions[channel]['connected'] = 0 - self.subscriptions[channel]['subscribed'] = False - self.subscriptions[channel]['timetoken'] = 0 - self.subscriptions[channel]['first'] = False - self.CONNECT() -- cgit v1.2.3