#!/usr/bin/env python # -*- coding: utf-8 -*- #### optz = dict( qlen = 10, tbf_size = 4, tbf_tick = 15, tbf_max_delay = 60, tbf_inc = 2, tbf_dec = 2 ) fs_poll_interval = 60 dbus_name = 'org.freedesktop.Notifications' dbus_iface = 'org.freedesktop.Notifications' ## Also there are these non-proxied interfaces: # org.freedesktop.DBus.Introspectable # org.freedesktop.DBus.Properties dbus_path = '/org/freedesktop/Notifications' dbus_dst = 'org.freedesktop.NotificationCore' urgency_levels = ['low', 'normal', 'critical'] #### from optparse import OptionParser parser = OptionParser(usage='%prog [options]', description='Start dbus notification proxy') parser.add_option('-f', '--no-fs-check', action='store_false', dest='fs_check', default=True, help='Dont queue messages if active window is fullscreen') parser.add_option('-u', '--no-urgency-check', action='store_false', dest='urgency_check', default=True, help='Queue messages even if urgency is critical') parser.add_option('-s', '--tbf-size', action='store', dest='tbf_size', type='int', metavar='NUM', default=optz['tbf_size'], help='Token-bucket message-flow filter (tbf)' ' bucket size (default: %default)') parser.add_option('-t', '--tbf-tick', action='store', dest='tbf_tick', type='int', metavar='NUM', default=optz['tbf_tick'], help='tbf update interval (new token), so token_inflow' ' = token / tbf_tick (default: %defaults)') parser.add_option('-m', '--tbf-max-delay', action='store', dest='tbf_max_delay', type='int', metavar='NUM', default=optz['tbf_max_delay'], help='Maxmum amount of seconds, between' ' message queue flush (default: %defaults)') parser.add_option('-i', '--tbf-inc', action='store', dest='tbf_inc', type='int', metavar='NUM', default=optz['tbf_inc'], help='tbf_tick multiplier on consequent tbf' ' overflow (default: %default)') parser.add_option('-d', '--tbf-dec', action='store', dest='tbf_dec', type='int', metavar='NUM', default=optz['tbf_dec'], help='tbf_tick divider on successful grab from non-empty bucket,' ' wont lower multiplier below 1 (default: %default)') parser.add_option('-q', '--queue-len', action='store', dest='qlen', type='int', metavar='NUM', default=optz['qlen'], help='How many messages should be' ' queued on tbf overflow (default: %default)') # parser.add_option('-l', '--log', # action='store', dest='log', type='str', metavar='PATH', # help='Write log to a given path instead of stderr') parser.add_option('--debug', action='store_true', dest='debug', help='Enable debugging output (to stderr)') optz, argz = parser.parse_args() if argz: parser.error('This command takes no arguments') from fgc import log log.cfg(level=(log.DEBUG if optz.debug else log.WARNING)) import itertools as it, operator as op, functools as ft from dbus.mainloop.glib import DBusGMainLoop from fgc.wm import Window from fgc.fc import FC_TokenBucket, RRQ from threading import Lock import dbus, dbus.service, gobject, signal, cgi DBusGMainLoop(set_as_default=True) bus = dbus.SessionBus() bus_name = dbus.service.BusName(dbus_name, bus) ncore = bus.get_object(dbus_dst, dbus_path) class NotifyProxy(dbus.service.Object): _async_lock = Lock() _last_note = None def __init__(self, *argz, **kwz): tick_strangle_max = op.truediv(optz.tbf_max_delay, optz.tbf_tick) super(NotifyProxy, self).__init__(*argz, **kwz) self._notify_limit = FC_TokenBucket( tick=optz.tbf_tick, burst=optz.tbf_size, tick_strangle=lambda x: min(x*optz.tbf_inc, tick_strangle_max), tick_free=lambda x: max(op.truediv(x, optz.tbf_dec), 1) ) self._notify_buffer = RRQ(optz.qlen) @dbus.service.method( dbus_iface, in_signature='', out_signature='ssss' ) def GetServerInformation(self): return ncore.GetServerInformation() @dbus.service.method( dbus_iface, in_signature='', out_signature='as' ) def GetCapabilities(self): return ncore.GetCapabilities() @dbus.service.method( dbus_iface, in_signature='u', out_signature='' ) def CloseNotification(self, id): return ncore.CloseNotification(id) def _escape_tags(self, text, encoding='utf-8'): if not isinstance(text, unicode): text = text.decode(encoding, 'replace') return cgi.escape(text).encode('ascii', 'xmlcharrefreplace') def forward(self, app_name, id, icon, summary, body, actions, hints, timeout): summary, body = self._escape_tags(summary), self._escape_tags(body) return ncore.Notify(app_name, id, icon, summary, body, actions, hints, timeout) @dbus.service.method( dbus_iface, in_signature='susssasa{sv}i', out_signature='u' ) def Notify(self, app_name, id, icon, summary, body, actions, hints, timeout): try: urgency = int(hints[u'urgency']) except (KeyError, ValueError): urgency = None with self._async_lock as lock: self._last_note = app_name, id, icon, \ summary, body, actions, hints, timeout fs = optz.fs_check and Window.get_active().fullscreen urgent = optz.urgency_check and urgency == urgency_levels.index('critical') if urgent: log.debug( 'Urgent message immediate passthru' ', tokens left: {0}'.format(self._notify_limit.tokens) ) self._notify_limit.consume(block=False) return self.forward(app_name, id, icon, summary, body, actions, hints, timeout) if fs or not self._notify_limit.consume(block=False): # Delay notification to = self._notify_limit.get_eta() if not fs else fs_poll_interval if to > 1: # no need to bother otherwise, note that it'll be an extra token ;) self._notify_buffer.append((summary, body)) to = int(to) + 1 # +1 is to ensure token arrival by that time log.debug( 'Queueing notification. Reason: {0}. Flush attempt in {1}s'\ .format('fullscreen window detected' if fs else 'notification rate limit', to) ) signal.alarm(to) return 0 signal.alarm(0) # no async calls past this point: either way it's a flush if self._notify_buffer: self._notify_buffer.append((summary, body)) log.debug('Token-flush of notification queue') return self.flush_buffer(scheduled=False) else: log.debug('Token-pass, {0} token(s) left'.format(self._notify_limit.tokens)) return self.forward(app_name, id, icon, summary, body, actions, hints, timeout) def _flush_buffer(self, signum=signal.SIGALRM, frame=None, scheduled=True): signal.alarm(0) # reset alarm, if present if not self._async_lock.acquire(False): return # queue-op in progress already self._async_lock.release() log.debug( 'Flushing notification queue ({0} msgs, {1} dropped)'\ .format(len(self._notify_buffer), self._notify_buffer.dropped) ) if scheduled: self._notify_limit.consume() if signum != signal.SIGALRM: self._notify_buffer.append(( 'System', 'Received death-signal ({0}), shutting down...'.format(signum) )) elif optz.fs_check and scheduled and Window.get_active().fullscreen: log.debug( 'Fullscreen window detected,' ' delaying buffer flush by {0}s'.format(fs_poll_interval) ) signal.alarm(fs_poll_interval) return 0 status = self.forward(*self._last_note) \ if len(self._notify_buffer) == 1 else self.forward( 'notification-feed', dbus.UInt32(0), 'FBReader', 'Feed' if not self._notify_buffer.dropped else 'Feed ({0} dropped)'.format(self._notify_buffer.dropped), '\n\n'.join(it.starmap('--- {0}\n {1}'.format, self._notify_buffer)), dbus.Array(signature="s"), dbus.Dictionary(signature='sv'), -1 ) self._notify_buffer.flush() log.debug('Notification buffer flushed') if signum != signal.SIGALRM: log.debug('Got termination signal ({0}), shutting down'.format(signum)) loop.quit() return status @ft.wraps(_flush_buffer) def flush_buffer(self, *argz, **kwz): # Needed to get exceptions outta gobject loop try: return self._flush_buffer(*argz, **kwz) except Exception as err: from fgc.err import ext_traceback log.error(ext_traceback()) raise interceptor = NotifyProxy(bus, dbus_path) signal.signal(signal.SIGALRM, interceptor.flush_buffer) loop = gobject.MainLoop() log.debug('Starting gobject loop') loop.run()