android_scripting/facade/EventFacade.java

/*
 * Copyright (C) 2017 The Android Open Source Project
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.googlecode.android_scripting.facade;

import android.content.Context;
import android.content.Intent;
import android.content.IntentFilter;
import android.os.Bundle;

import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.Multimap;
import com.google.common.collect.Multimaps;
import com.googlecode.android_scripting.Log;
import com.googlecode.android_scripting.event.Event;
import com.googlecode.android_scripting.event.EventObserver;
import com.googlecode.android_scripting.future.FutureResult;
import com.googlecode.android_scripting.jsonrpc.JsonBuilder;
import com.googlecode.android_scripting.jsonrpc.RpcReceiver;
import com.googlecode.android_scripting.rpc.Rpc;
import com.googlecode.android_scripting.rpc.RpcDefault;
import com.googlecode.android_scripting.rpc.RpcDeprecated;
import com.googlecode.android_scripting.rpc.RpcName;
import com.googlecode.android_scripting.rpc.RpcOptional;
import com.googlecode.android_scripting.rpc.RpcParameter;

import org.json.JSONException;

import java.util.HashMap;
import java.util.List;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;

/**
 * Manage the event queue. <br>
 * <br>
 * <b>Usage Notes:</b><br>
 * EventFacade APIs interact with the Event Queue (a data buffer containing up to 1024 event
 * entries).<br>
 * Events are automatically entered into the Event Queue following API calls such as startSensing()
 * and startLocating().<br>
 * The Event Facade provides control over how events are entered into (and removed from) the Event
 * Queue.<br>
 * The Event Queue provides a useful means of recording background events (such as sensor data) when
 * the phone is busy with foreground activities.
 */
public class EventFacade extends RpcReceiver {
    /**
     * The maximum length of the event queue. Old events will be discarded when this limit is
     * exceeded.
     */
    private static final int MAX_QUEUE_SIZE = 1024;
    private final Queue<Event> mEventQueue = new ConcurrentLinkedQueue<Event>();
    private final CopyOnWriteArrayList<EventObserver> mGlobalEventObservers =
            new CopyOnWriteArrayList<EventObserver>();
    private final Multimap<String, EventObserver> mNamedEventObservers = Multimaps
            .synchronizedListMultimap(ArrayListMultimap.<String, EventObserver>create());
    private final HashMap<String, BroadcastListener> mBroadcastListeners =
            new HashMap<String, BroadcastListener>();
    private final Context mContext;

    public EventFacade(FacadeManager manager) {
        super(manager);
        mContext = manager.getService().getApplicationContext();
        Log.v("Creating new EventFacade Instance()");
    }

    /**
     * Example (python): droid.eventClearBuffer()
     */
    @Rpc(description = "Clears all events from the event buffer.")
    public void eventClearBuffer() {
        mEventQueue.clear();
    }

    /**
     * Registers a listener for a new broadcast signal
     */
    @Rpc(description = "Registers a listener for a new broadcast signal")
    public boolean eventRegisterForBroadcast(
            @RpcParameter(name = "category") String category,
            @RpcParameter(name = "enqueue",
                    description = "Should this events be added to the event queue or only dispatched") @RpcDefault(value = "true") Boolean enqueue) {
        if (mBroadcastListeners.containsKey(category)) {
            return false;
        }

        BroadcastListener b = new BroadcastListener(this, enqueue.booleanValue());
        IntentFilter c = new IntentFilter(category);
        mContext.registerReceiver(b, c);
        mBroadcastListeners.put(category, b);

        return true;
    }

    @Rpc(description = "Stop listening for a broadcast signal")
    public void eventUnregisterForBroadcast(
            @RpcParameter(name = "category") String category) {
        if (!mBroadcastListeners.containsKey(category)) {
            return;
        }

        mContext.unregisterReceiver(mBroadcastListeners.get(category));
        mBroadcastListeners.remove(category);
    }

    @Rpc(description = "Lists all the broadcast signals we are listening for")
    public Set<String> eventGetBrodcastCategories() {
        return mBroadcastListeners.keySet();
    }

    /**
     * Actual data returned in the map will depend on the type of event.
     * <p>
     * <pre>
     * Example (python):
     *     import android, time
     *     droid = android.Android()
     *     droid.startSensing()
     *     time.sleep(1)
     *     droid.eventClearBuffer()
     *     time.sleep(1)
     *     e = eventPoll(1).result
     *     event_entry_number = 0
     *     x = e[event_entry_ number]['data']['xforce']
     * </pre>
     * <p>
     * e has the format:<br>
     * [{u'data': {u'accuracy': 0, u'pitch': -0.48766891956329345, u'xmag': -5.6875, u'azimuth':
     * 0.3312483489513397, u'zforce': 8.3492730000000002, u'yforce': 4.5628165999999997, u'time':
     * 1297072704.813, u'ymag': -11.125, u'zmag': -42.375, u'roll': -0.059393649548292161,
     * u'xforce': 0.42223078000000003}, u'name': u'sensors', u'time': 1297072704813000L}]<br>
     * x has the string value of the x force data (0.42223078000000003) at the time of the event
     * entry. </pre>
     */

    @Rpc(description = "Returns and removes the oldest n events (i.e. location or sensor update, etc.) from the event buffer.",
            returns = "A List of Maps of event properties.")
    public List<Event> eventPoll(
            @RpcParameter(name = "number_of_events") @RpcDefault("1") Integer number_of_events) {
        List<Event> events = Lists.newArrayList();
        for (int i = 0; i < number_of_events; i++) {
            Event event = mEventQueue.poll();
            if (event == null) {
                break;
            }
            events.add(event);
        }
        return events;
    }

    @Rpc(description = "Blocks until an event with the supplied name occurs. Event is removed from the buffer if removeEvent is True.",
            returns = "Map of event properties.")
    public Event eventWaitFor(
            @RpcParameter(name = "eventName") final String eventName,
            @RpcParameter(name = "removeEvent") final Boolean removeEvent,
            @RpcParameter(name = "timeout", description = "the maximum time to wait (in ms)") @RpcOptional Integer timeout)
            throws InterruptedException {
        Event result = null;
        final FutureResult<Event> futureEvent;
        synchronized (mEventQueue) { // First check to make sure it isn't already there
            for (Event event : mEventQueue) {
                if (event.getName().equals(eventName)) {
                    result = event;
                    if (removeEvent)
                        mEventQueue.remove(event);
                    return result;
                }
            }
            futureEvent = new FutureResult<Event>();
            addNamedEventObserver(eventName, new EventObserver() {
                @Override
                public void onEventReceived(Event event) {
                    if (event.getName().equals(eventName)) {
                        synchronized (futureEvent) {
                            if (!futureEvent.isDone()) {
                                futureEvent.set(event);
                                // TODO: Remove log.
                                Log.v(String.format("Removing observer (%s) got event  (%s)",
                                        this,
                                        event));
                                removeNamedEventObserver(eventName, this);
                            }
                            if (removeEvent)
                                mEventQueue.remove(event);
                        }
                    }
                }
            });
        }
        if (futureEvent != null) {
            if (timeout != null) {
                result = futureEvent.get(timeout, TimeUnit.MILLISECONDS);
            } else {
                result = futureEvent.get();
            }
        }
        return result;
    }

    @Rpc(description = "Blocks until an event occurs. The returned event is removed from the buffer.",
            returns = "Map of event properties.")
    public Event eventWait(
            @RpcParameter(name = "timeout", description = "the maximum time to wait") @RpcOptional Integer timeout)
            throws InterruptedException {
        Event result = null;
        final FutureResult<Event> futureEvent = new FutureResult<Event>();
        EventObserver observer;
        synchronized (mEventQueue) { // Anything in queue?
            if (mEventQueue.size() > 0) {
                return mEventQueue.poll(); // return it.
            }
            observer = new EventObserver() {
                @Override
                public void onEventReceived(Event event) { // set up observer for any events.
                    synchronized (futureEvent) {
                        if (!futureEvent.isDone()) {
                            futureEvent.set(event);
                            // TODO: Remove log.
                            Log.v(String.format("onEventReceived for event (%s)", event));
                        }
                    }
                }
            };
            addGlobalEventObserver(observer);
        }
        if (timeout != null) {
            result = futureEvent.get(timeout, TimeUnit.MILLISECONDS);
        } else {
            result = futureEvent.get();
        }
        if (result != null) {
            mEventQueue.remove(result);
        }
        // TODO: Remove log.
        Log.v(String.format("Removing observer (%s) got event  (%s)", observer, result));
        if (observer != null) {
            removeEventObserver(observer); // Make quite sure this goes away.
        }
        return result;
    }

    /**
     * <pre>
     * Example:
     *   import android
     *   from datetime import datetime
     *   droid = android.Android()
     *   t = datetime.now()
     *   droid.eventPost('Some Event', t)
     * </pre>
     */
    @Rpc(description = "Post an event to the event queue.")
    public void eventPost(
            @RpcParameter(name = "name", description = "Name of event") String name,
            @RpcParameter(name = "data", description = "Data contained in event.") String data,
            @RpcParameter(name = "enqueue",
                    description = "Set to False if you don't want your events to be added to the event queue, just dispatched.") @RpcOptional @RpcDefault("false") Boolean enqueue) {
        postEvent(name, data, enqueue.booleanValue());
    }

    /**
     * Post an event and queue it
     */
    public void postEvent(String name, Object data) {
        postEvent(name, data, true);
    }

    /**
     * Posts an event with to the event queue.
     */
    public void postEvent(String name, Object data, boolean enqueue) {
        Event event = new Event(name, data);
        if (enqueue) {
            Log.v(String.format("postEvent(%s)", name));
            synchronized (mEventQueue) {
                while (mEventQueue.size() >= MAX_QUEUE_SIZE) {
                    mEventQueue.remove();
                }
                mEventQueue.add(event);
                // b/77306870: Posting to the EventObservers when enqueuing an event must be
                // done when mEventQueue is locked. Otherwise, we can run into the following
                // race condition:
                // 1) postEvent() adds the event to the event queue, and releases mEventQueue.
                //                Here, the thread is put to sleep.
                // 2) eventWait() is called when an event is queued, and exits immediately.
                // 3) eventWait() is called a second time, finds no event and creates a
                //                GlobalEventObserver.
                // 4) postEvent() wakes back up, and continues to post the event to the observers.
                //                The same event sent to the first eventWait call is sent to the
                //                second eventWait call's observer, causing a duplicated received
                //                event.
                postEventToNamedObservers(event);
                postEventToGlobalObservers(event);
            }
        } else {
            postEventToNamedObservers(event);
            postEventToGlobalObservers(event);
        }
    }

    /**
     * Posts the event to all applicable Named Observers.
     */
    private void postEventToNamedObservers(Event event) {
        synchronized (mNamedEventObservers) {
            for (EventObserver observer : mNamedEventObservers.get(event.getName())) {
                Log.d(String.format("namedEventObserver %s received event %s",
                        observer,
                        event.getName()));
                observer.onEventReceived(event);
            }
        }
    }

    /**
     * Posts the event to the Global Observers list.
     */
    private void postEventToGlobalObservers(Event event) {
        synchronized (mGlobalEventObservers) {
            for (EventObserver observer : mGlobalEventObservers) {
                Log.d(String.format("globalEventObserver %s received event %s",
                        observer,
                        event.getName()));
                observer.onEventReceived(event);
            }
        }
    }

    @RpcDeprecated(value = "eventPost", release = "r4")
    @Rpc(description = "Post an event to the event queue.")
    @RpcName(name = "postEvent")
    public void rpcPostEvent(
            @RpcParameter(name = "name") String name,
            @RpcParameter(name = "data") String data) {
        postEvent(name, data);
    }

    @RpcDeprecated(value = "eventPoll", release = "r4")
    @Rpc(description = "Returns and removes the oldest event (i.e. location or sensor update, etc.) from the event buffer.",
            returns = "Map of event properties.")
    public Event receiveEvent() {
        return mEventQueue.poll();
    }

    @RpcDeprecated(value = "eventWaitFor", release = "r4")
    @Rpc(description = "Blocks until an event with the supplied name occurs. Event is removed from the buffer if removeEvent is True.",
            returns = "Map of event properties.")
    public Event waitForEvent(
            @RpcParameter(name = "eventName") final String eventName,
            @RpcOptional final Boolean removeEvent,
            @RpcParameter(name = "timeout", description = "the maximum time to wait") @RpcOptional Integer timeout)
            throws InterruptedException {
        return eventWaitFor(eventName, removeEvent, timeout);
    }

    /**
     * Closes this SL4A session, and sends a terminating signal to the event observers.
     */
    @Rpc(description = "sl4a session is shutting down, send terminate event to client.")
    public void closeSl4aSession() {
        eventClearBuffer();
        postEvent("EventDispatcherShutdown", null);
    }

    /**
     * Shuts down the RPC server.
     */
    @Override
    public void shutdown() {
        mGlobalEventObservers.clear();
        mEventQueue.clear();
    }

    /**
     * Adds a named observer to the event listening queue.
     * @param eventName the name of the event to listen to
     * @param observer  the observer object
     */
    public void addNamedEventObserver(String eventName, EventObserver observer) {
        mNamedEventObservers.put(eventName, observer);
    }

    /**
     * Adds a global event listener ot the listening queue.
     * @param observer the observer object
     */
    public void addGlobalEventObserver(EventObserver observer) {
        mGlobalEventObservers.add(observer);
    }

    /**
     * Removes an observer from the event listening queue.
     * @param observer the observer to remove
     */
    public void removeEventObserver(EventObserver observer) {
        mGlobalEventObservers.remove(observer);
    }

    /**
     * Removes a named observer from the event listening queue.
     * @param eventName the name of the event being listened to.
     * @param observer the observer to remove
     */
    public void removeNamedEventObserver(String eventName, EventObserver observer) {
        mNamedEventObservers.removeAll(eventName);
        removeEventObserver(observer);
    }

    public class BroadcastListener extends android.content.BroadcastReceiver {
        private EventFacade mParent;
        private boolean mEnQueue;

        public BroadcastListener(EventFacade parent, boolean enqueue) {
            mParent = parent;
            mEnQueue = enqueue;
        }

        @Override
        public void onReceive(Context context, Intent intent) {
            Bundle data;
            if (intent.getExtras() != null) {
                data = (Bundle) intent.getExtras().clone();
            } else {
                data = new Bundle();
            }
            data.putString("action", intent.getAction());
            try {
                mParent.eventPost("sl4a", JsonBuilder.build(data).toString(), mEnQueue);
            } catch (JSONException e) {
                e.printStackTrace();
            }
        }

    }
}