summaryrefslogtreecommitdiff
path: root/src/TimedQueue.vala
diff options
context:
space:
mode:
Diffstat (limited to 'src/TimedQueue.vala')
-rw-r--r--src/TimedQueue.vala284
1 files changed, 284 insertions, 0 deletions
diff --git a/src/TimedQueue.vala b/src/TimedQueue.vala
new file mode 100644
index 0000000..7001421
--- /dev/null
+++ b/src/TimedQueue.vala
@@ -0,0 +1,284 @@
+/* Copyright 2010-2014 Yorba Foundation
+ *
+ * This software is licensed under the GNU Lesser General Public License
+ * (version 2.1 or later). See the COPYING file in this distribution.
+ */
+
+// TimedQueue is a specialized collection class. It holds items in order, but rather than being
+// manually dequeued, they are dequeued automatically after a specified amount of time has elapsed
+// for that item. As of today, it's possible the item will be dequeued a bit later than asked
+// for, but it will never be early. Future implementations might tighten up the lateness.
+//
+// The original design was to use a signal to notify when an item has been dequeued, but Vala has
+// a bug with passing an unnamed type as a signal parameter:
+// https://bugzilla.gnome.org/show_bug.cgi?id=628639
+//
+// The rate the items come off the queue can be spaced out. Note that this can cause items to back
+// up. As of today, TimedQueue makes no effort to combat this.
+
+public delegate void DequeuedCallback<G>(G item);
+
+public class TimedQueue<G> {
+ private class Element<G> {
+ public G item;
+ public ulong ready;
+
+ public Element(G item, ulong ready) {
+ this.item = item;
+ this.ready = ready;
+ }
+
+ public static int64 comparator(void *a, void *b) {
+ return (int64) ((Element *) a)->ready - (int64) ((Element *) b)->ready;
+ }
+ }
+
+ private uint hold_msec;
+ private unowned DequeuedCallback<G> callback;
+ private Gee.EqualDataFunc<G> equal_func;
+ private int priority;
+ private uint timer_id = 0;
+ private SortedList<Element<G>> queue;
+ private uint dequeue_spacing_msec = 0;
+ private ulong last_dequeue = 0;
+ private bool paused_state = false;
+
+ public virtual signal void paused(bool is_paused) {
+ }
+
+ // Initial design was to have a signal that passed the dequeued G, but bug in valac meant
+ // finding a workaround, namely using a delegate:
+ // https://bugzilla.gnome.org/show_bug.cgi?id=628639
+ public TimedQueue(uint hold_msec, DequeuedCallback<G> callback,
+ owned Gee.EqualDataFunc? equal_func = null, int priority = Priority.DEFAULT) {
+ this.hold_msec = hold_msec;
+ this.callback = callback;
+
+ if (equal_func != null)
+ this.equal_func = (owned) equal_func;
+ else
+ this.equal_func = (Gee.EqualDataFunc<G>) (Gee.Functions.get_equal_func_for(typeof(G)));
+
+ this.priority = priority;
+
+ queue = new SortedList<Element<G>>(Element.comparator);
+
+ timer_id = Timeout.add(get_heartbeat_timeout(), on_heartbeat, priority);
+ }
+
+ ~TimedQueue() {
+ if (timer_id != 0)
+ Source.remove(timer_id);
+ }
+
+ public uint get_dequeue_spacing_msec() {
+ return dequeue_spacing_msec;
+ }
+
+ public void set_dequeue_spacing_msec(uint msec) {
+ if (msec == dequeue_spacing_msec)
+ return;
+
+ if (timer_id != 0)
+ Source.remove(timer_id);
+
+ dequeue_spacing_msec = msec;
+
+ timer_id = Timeout.add(get_heartbeat_timeout(), on_heartbeat, priority);
+ }
+
+ private uint get_heartbeat_timeout() {
+ return ((dequeue_spacing_msec == 0)
+ ? (hold_msec / 10)
+ : (dequeue_spacing_msec / 2)).clamp(10, uint.MAX);
+ }
+
+ protected virtual void notify_dequeued(G item) {
+ callback(item);
+ }
+
+ public bool is_paused() {
+ return paused_state;
+ }
+
+ public void pause() {
+ if (paused_state)
+ return;
+
+ paused_state = true;
+
+ paused(true);
+ }
+
+ public void unpause() {
+ if (!paused_state)
+ return;
+
+ paused_state = false;
+
+ paused(false);
+ }
+
+ public virtual void clear() {
+ queue.clear();
+ }
+
+ public virtual bool contains(G item) {
+ foreach (Element<G> e in queue) {
+ if (equal_func(item, e.item))
+ return true;
+ }
+
+ return false;
+ }
+
+ public virtual bool enqueue(G item) {
+ return queue.add(new Element<G>(item, calc_ready_time()));
+ }
+
+ public virtual bool enqueue_many(Gee.Collection<G> items) {
+ ulong ready_time = calc_ready_time();
+
+ Gee.ArrayList<Element<G>> elements = new Gee.ArrayList<Element<G>>();
+ foreach (G item in items)
+ elements.add(new Element<G>(item, ready_time));
+
+ return queue.add_list(elements);
+ }
+
+ public virtual bool remove_first(G item) {
+ Gee.Iterator<Element<G>> iter = queue.iterator();
+ while (iter.next()) {
+ Element<G> e = iter.get();
+ if (equal_func(item, e.item)) {
+ iter.remove();
+
+ return true;
+ }
+ }
+
+ return false;
+ }
+
+ public virtual int size {
+ get {
+ return queue.size;
+ }
+ }
+
+ private ulong calc_ready_time() {
+ return now_ms() + (ulong) hold_msec;
+ }
+
+ private bool on_heartbeat() {
+ if (paused_state)
+ return true;
+
+ ulong now = 0;
+
+ for (;;) {
+ if (queue.size == 0)
+ break;
+
+ Element<G>? head = queue.get_at(0);
+ assert(head != null);
+
+ if (now == 0)
+ now = now_ms();
+
+ if (head.ready > now)
+ break;
+
+ // if a space of time is required between dequeues, check now
+ if ((dequeue_spacing_msec != 0) && ((now - last_dequeue) < dequeue_spacing_msec))
+ break;
+
+ Element<G>? h = queue.remove_at(0);
+ assert(head == h);
+
+ notify_dequeued(head.item);
+ last_dequeue = now;
+
+ // if a dequeue spacing is in place, it's a lock that only one item is dequeued per
+ // heartbeat
+ if (dequeue_spacing_msec != 0)
+ break;
+ }
+
+ return true;
+ }
+}
+
+// HashTimedQueue uses a HashMap for quick lookups of elements via contains().
+
+public class HashTimedQueue<G> : TimedQueue<G> {
+ private Gee.HashMap<G, int> item_count;
+
+ public HashTimedQueue(uint hold_msec, DequeuedCallback<G> callback,
+ owned Gee.HashDataFunc<G>? hash_func = null, owned Gee.EqualDataFunc<G>? equal_func = null,
+ int priority = Priority.DEFAULT) {
+ base (hold_msec, callback, (owned) equal_func, priority);
+
+ item_count = new Gee.HashMap<G, int>((owned) hash_func, (owned) equal_func);
+ }
+
+ protected override void notify_dequeued(G item) {
+ removed(item);
+
+ base.notify_dequeued(item);
+ }
+
+ public override void clear() {
+ item_count.clear();
+
+ base.clear();
+ }
+
+ public override bool contains(G item) {
+ return item_count.has_key(item);
+ }
+
+ public override bool enqueue(G item) {
+ if (!base.enqueue(item))
+ return false;
+
+ item_count.set(item, item_count.has_key(item) ? item_count.get(item) + 1 : 1);
+
+ return true;
+ }
+
+ public override bool enqueue_many(Gee.Collection<G> items) {
+ if (!base.enqueue_many(items))
+ return false;
+
+ foreach (G item in items)
+ item_count.set(item, item_count.has_key(item) ? item_count.get(item) + 1 : 1);
+
+ return true;
+ }
+
+ public override bool remove_first(G item) {
+ if (!base.remove_first(item))
+ return false;
+
+ removed(item);
+
+ return true;
+ }
+
+ private void removed(G item) {
+ // item in question is either already removed
+ // or was never added, safe to do nothing here
+ if (!item_count.has_key(item))
+ return;
+
+ int count = item_count.get(item);
+ assert(count > 0);
+
+ if (--count == 0)
+ item_count.unset(item);
+ else
+ item_count.set(item, count);
+ }
+}
+