Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
notification_center.py
from objc_util import *
import console
from functools import partial
NSNotificationCenter=ObjCClass('NSNotificationCenter')
#logging.basicConfig(filename='log.txt',format='%(levelname)s:%(message)s', level=logging.DEBUG)
class NotificationObserver(object):
''' Create a Notification oobserver for notification named name, and callback the callable callback.
if name is none, respond to all notifications, though exclude notifications starting with excludes
callback should be of form:def cb(name,obj, userInfo)
name can be a specific notification name, or None. None captures all notifications unless filtered by excludes/namefilter
excludes can be a tuple or list of strings to exclude using startswith
namefilter is a string, or None, and looks for names startung with some prefix
'''
def __init__(self, callback=None, name=None, excludes=('UI','NS','_UI'),namefilter=None) :
#self.captured=[]
self.captured={}
self.center=NSNotificationCenter.defaultCenter()
self.name=name
self.namefilter=namefilter
if not name:
self.excludes=excludes
else:
self.excludes=[]
if callable(callback):
self.callback=callback
else:
self.callback=None
self._observer=None
self._blk=None
def _block(self,_cmd,notification):
try:
name=str(ObjCInstance(notification).name())
if self.excludes and name.startswith(self.excludes):
return
if self.namefilter and not name.startswith(self.namefilter):
return
obj=str(ObjCInstance(notification).object())
userInfo=str(ObjCInstance(notification).userInfo())
if self.callback:p and callable(self.callback):
self.callback(name, obj, userInfo)
except Exception as e:
raise
def start(self):
if self._observer:
raise Exception('observer already started')
self._blk=ObjCBlock(self._block,
restype=None,
argtypes=[c_void_p,c_void_p])
self._observer = \
self.center.addObserverForName_object_queue_usingBlock_(self.name,None,None,self._blk)
#retain_global(self)
def stop(self):
import objc_util
if self._observer:
self.center.removeObserver_(self._observer)
self._observer=None
release_global(self)
def reset(self):
self.stop()
class Canary:
def __init__(self, g):
@on_main_thread
def die():
print('canary died')
g.center.removeObserver_(g._observer)
g._blk=None
g.callback=None
import weakref
self._ref=weakref.finalize(self,die)
if __name__=='__main__':
def cb(name,obj,userInfo):
print( 'notification',name,'\nobject: ',obj,'\nuserinfo: ',userInfo)
#g.captured=True
g=NotificationObserver(callback=cb,name='UIApplicationDidEnterBackgroundNotification')
g.start()
print('started watching for bg notifications')
'''there may be a better way to do this, but i couldnt get it to work, because there are some stubborn references to the notificationObserver that prevent it from actually getting gc'd. so, i create a "canary" that will be killed first when globals get cleared. the canary weakref finalize takes care of stopping the observer, so we are not getting duplicate notifications, and trying to clear some references that might keep it from getting gc'd later'''
canary=Canary(g)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.