Merge lp:~rainct/zeitgeist/monitor-queue into lp:~zeitgeist/zeitgeist/bluebird

[Siegfried Gevatter]

[Michal Hruby (mhr3)]

22 + [Compact]
23 + private class InsertionData : NotificationData

I can't help it, but this looks like serious misusage of the type system, I'd rather go with single compact class that has an enum indicating the type plus a method that dispatches a call (passing in the proxy object - although as inner class it should be able to access private members of the parent class iirc)

121 + queued_notifications.append (

Appending SList is really slow, let's prepend and reverse when we're about to flush the queue.

review: Needs Fixing

[lp:~rainct/zeitgeist/monitor-queue]

378. By Siegfried Gevatter on 2012-01-31

Use a single class for the queued notifications

[Michal Hruby (mhr3)]

Beautiful :)

review: Approve

=== modified file 'src/notify.vala'
--- src/notify.vala	2011-10-20 14:20:17 +0000
+++ src/notify.vala	2012-01-31 10:17:12 +0000
@@ -32,7 +32,7 @@
         construct
         {
             monitors = new HashTable<string, Monitor> (str_hash, str_equal);
-            connections = new HashTable<string, GenericArray<string>> 
+            connections = new HashTable<string, GenericArray<string>>
                 (str_hash, str_equal);
 
             // FIXME: it'd be nice if this supported arg2
@@ -78,9 +78,47 @@
             private TimeRange time_range;
             private RemoteMonitor? proxy_object = null;
 
+            private enum NotificationType
+            {
+                INSERTION,
+                DELETION
+            }
+            [Compact]
+            private class QueuedNotification {
+                // (Compact classes don't support private fields)
+                public NotificationType type;
+                public Variant time_range;
+                public Variant events; // for insertions
+                public uint32[] event_ids; // for deletions
+
+                public QueuedNotification.insertion (Variant time_range, Variant events)
+                {
+                    type = NotificationType.INSERTION;
+                    this.time_range = time_range;
+                    this.events = events;
+                }
+
+                public QueuedNotification.deletion (Variant time_range, uint32[] event_ids)
+                {
+                    type = NotificationType.DELETION;
+                    this.time_range = time_range;
+                    this.event_ids = event_ids;
+                }
+
+                public void send (RemoteMonitor proxy_object)
+                {
+                    if (type == NotificationType.INSERTION)
+                        proxy_object.notify_insert (time_range, events);
+                    else
+                        proxy_object.notify_delete (time_range, event_ids);
+                }
+            }
+            private SList<QueuedNotification> queued_notifications;
+
             public Monitor (BusName peer, string object_path,
                 TimeRange tr, GenericArray<Event> templates)
             {
+                queued_notifications = new SList<QueuedNotification> ();
                 Bus.get_proxy<RemoteMonitor> (BusType.SESSION, peer,
                     object_path, DBusProxyFlags.DO_NOT_LOAD_PROPERTIES |
                     DBusProxyFlags.DO_NOT_CONNECT_SIGNALS,
@@ -94,6 +132,15 @@
                         {
                             warning ("%s", err.message);
                         }
+
+                        // Process queued notifications...
+                        queued_notifications.reverse ();
+                        foreach (unowned QueuedNotification notification
+                            in queued_notifications)
+                        {
+                            notification.send (proxy_object);
+                        }
+                        queued_notifications = null;
                     });
                 time_range = tr;
                 event_templates = templates;
@@ -113,15 +160,13 @@
                 return false;
             }
 
-            // FIXME: we need to queue the notification if proxy_object == null
             public void notify_insert (TimeRange time_range, GenericArray<Event> events)
-                requires (proxy_object != null)
             {
                 var intersect_tr = time_range.intersect (this.time_range);
                 if (intersect_tr != null)
                 {
                     var matching_events = new GenericArray<Event> ();
-                    for (int i=0; i<events.length; i++)
+                    for (int i = 0; i < events.length; i++)
                     {
                         if (events[i] != null && matches (events[i])
                             && events[i].timestamp >= intersect_tr.start
@@ -132,24 +177,46 @@
                     }
                     if (matching_events.length > 0)
                     {
-                        DBusProxy p = (DBusProxy) proxy_object;
-                        debug ("Notifying %s about %d insertions",
-                            p.get_name (), matching_events.length);
-
-                        proxy_object.notify_insert (intersect_tr.to_variant (),
-                            Events.to_variant (matching_events));
+                        Variant time_v = intersect_tr.to_variant ();
+                        // FIXME: do we want to "cache" this for sharing
+                        // between monitors?
+                        Variant events_v = Events.to_variant (matching_events);
+
+                        if (proxy_object != null)
+                        {
+                            DBusProxy p = (DBusProxy) proxy_object;
+                            debug ("Notifying %s about %d insertions",
+                                p.get_name (), matching_events.length);
+
+                            proxy_object.notify_insert (time_v, events_v);
+                        }
+                        else
+                        {
+                            debug ("Queueing notification about %d insertions",
+                                matching_events.length);
+                            queued_notifications.prepend (
+                                new QueuedNotification.insertion (time_v, events_v));
+                        }
                     }
                 }
             }
 
             public void notify_delete (TimeRange time_range, uint32[] event_ids)
-                requires (proxy_object != null)
             {
                 var intersect_tr = time_range.intersect (this.time_range);
                 if (intersect_tr != null)
                 {
-                    proxy_object.notify_delete (intersect_tr.to_variant (),
-                        event_ids);
+                    Variant time_v = intersect_tr.to_variant ();
+
+                    if (proxy_object != null)
+                    {
+                        proxy_object.notify_delete (time_v, event_ids);
+                    }
+                    else
+                    {
+                        queued_notifications.prepend (
+                            new QueuedNotification.deletion (time_v, event_ids));
+                    }
                 }
             }
         }
@@ -179,12 +246,12 @@
         {
             debug ("Removing monitor %s%s", peer, object_path);
             var hash = "%s#%s".printf (peer, object_path);
-            
+
             if (monitors.lookup (hash) != null)
                 monitors.remove (hash);
             else
                 warning ("There's no monitor installed for %s", hash);
-            
+
             if (connections.lookup (peer) != null)
             {
                 var paths = connections.lookup (peer);