package com.openxc.sinks;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;

import com.openxc.remote.RawMeasurement;

import android.util.Log;

/**
 * Functionality to notify multiple clients asynchronously of new measurements.
 *
 * This class encapsulates the functionality to keep a thread-safe list of
 * listeners that want to be notified of updates asyncronously. Subclasses need
 * only to implement the {@link #propagateMeasurement(String, RawMeasurement)}
 * to add the actual logic for looping over the list of receivers and send them
 * new values.
 *
 * New measurments are queued up and propagated to receivers in a separate
 * thread, to avoid blocking the original sender of the data.
 */
public abstract class AbstractQueuedCallbackSink extends BaseVehicleDataSink {
    private final static String TAG = "AbstractQueuedCallbackSink";
    private final static int MAX_QUEUE_LENGTH = 500;

    private NotificationThread mNotificationThread;
    private BlockingQueue<String> mNotificationQueue;

    public AbstractQueuedCallbackSink() {
        mNotificationQueue = new ArrayBlockingQueue<String>(MAX_QUEUE_LENGTH);
        mNotificationThread = new NotificationThread();
        mNotificationThread.start();
    }

    abstract protected void propagateMeasurement(String measurementId,
            RawMeasurement measurement);

    public synchronized void stop() {
        if(mNotificationThread != null) {
            mNotificationThread.done();
        }
    }

    public void receive(RawMeasurement rawMeasurement)
            throws DataSinkException {
        super.receive(rawMeasurement);
        try  {
            mNotificationQueue.put(rawMeasurement.getName());
        } catch(InterruptedException e) {}
    }

    private class NotificationThread extends Thread {
        private boolean mRunning = true;

        private synchronized boolean isRunning() {
            return mRunning;
        }

        public synchronized void done() {
            Log.d(TAG, "Stopping notification thread");
            mRunning = false;
            // A context switch right can cause a race condition if we
            // used take() instead of poll(): when mRunning is set to
            // false and interrupt is called but we haven't called
            // take() yet, so nobody is waiting. By using poll we can not be
            // locked for more than 1s.
            interrupt();
        }

        public void run() {
            while(isRunning()) {
                String measurementId = null;
                try {
                    measurementId = mNotificationQueue.poll(1, TimeUnit.SECONDS);
                } catch(InterruptedException e) {
                    Log.d(TAG, "Interrupted while waiting for a new " +
                            "item for notification -- likely shutting down");
                    return;
                }

                if(measurementId != null) {
                    RawMeasurement rawMeasurement = get(measurementId);
                    if(rawMeasurement != null) {
                        propagateMeasurement(measurementId, rawMeasurement);
                    }
                }
            }
            Log.d(TAG, "Stopped measurement notifier");
        }
    }
}

@@ -48,29 +48,31 @@
 current.
-B
 lock
-ingQueue
+s.Condition
 ;%0Aimport
@@ -93,34 +93,26 @@
 current.
-ArrayBlockingQueue
+locks.Lock
 ;%0Aimport
@@ -133,23 +133,110 @@
 current.
-TimeUni
+locks.ReentrantLock;%0A%0Aimport java.util.HashSet;%0Aimport java.util.Iterator;%0Aimport java.util.Se
 t;%0A%0Aimpo
@@ -1006,16 +1006,17 @@
 kSink%22;%0A
+%0A
     priv
@@ -1023,49 +1023,47 @@
 ate 
-final static int MAX_QUEUE_LENGTH = 500;%0A
+NotificationThread mNotificationThread;
 %0A   
@@ -1063,32 +1063,38 @@
 ad;%0A    private 
+Lock m
 NotificationThre
@@ -1081,38 +1081,60 @@
 ck mNotification
-Thread
+sLock;%0A    private Condition
  mNotificationTh
@@ -1123,37 +1123,39 @@
 on mNotification
-Threa
+Receive
 d;%0A    private B
@@ -1153,29 +1153,19 @@
 private 
-BlockingQueue
+Set
 %3CString%3E
@@ -1178,21 +1178,17 @@
 fication
-Queue
+s
 ;%0A%0A    p
@@ -1249,63 +1249,151 @@
 tion
-Queue = new ArrayBlockingQueue%3CString%3E(MAX_QUEUE_LENGTH
+s = new HashSet%3CString%3E(32);%0A        mNotificationsLock = new ReentrantLock();%0A        mNotificationReceived = mNotificationsLock.newCondition(
 );%0A 
@@ -1886,19 +1886,142 @@
     
-try  %7B%0A    
+mNotificationsLock.lock();%0A        // TODO what we if we added the actual measurement here, that would save%0A        // a lookup later%0A
     
@@ -2041,17 +2041,13 @@
 tion
-Queue.put
+s.add
 (raw
@@ -2082,42 +2082,76 @@
     
-%7D catch(InterruptedException e) %7B%7D
+mNotificationReceived.signal();%0A        mNotificationsLock.unlock();
 %0A   
@@ -2911,141 +2911,167 @@
     
-String measurementId = null;%0A                try %7B%0A                    measurementId = mNotificationQueue.poll(1, TimeUnit.SECONDS);%0A
+mNotificationsLock.lock();%0A                if(mNotifications.isEmpty()) %7B%0A                    try %7B%0A                        mNotificationReceived.await();%0A    
     
@@ -3112,24 +3112,28 @@
 eption e) %7B%0A
+    
             
@@ -3220,16 +3220,20 @@
         
+    
 %22item fo
@@ -3298,54 +3298,241 @@
     
-return;%0A                %7D%0A%0A                if(
+    mNotificationsLock.unlock();%0A                        return;%0A                    %7D%0A                %7D%0A%0A                Iterator%3CString%3E it = mNotifications.iterator();%0A                if(it.hasNext()) %7B%0A                    String 
 meas
@@ -3541,26 +3541,61 @@
 ementId 
-!
 = 
-null) %7B
+it.next();%0A                    it.remove();
 %0A       
@@ -3817,32 +3817,77 @@
               %7D%0A
+                mNotificationsLock.unlock();%0A
             %7D%0A  

Use a sorted set instead of queue to speed up notifications.