Merge pull request #431 from FlashSystems/master

Asynchronous event API

src/main/java/com/nutomic/syncthingandroid/syncthing/EventProcessor.java

@@ -0,0 +1,190 @@
+package com.nutomic.syncthingandroid.syncthing;
+
+import android.content.Context;
+import android.content.Intent;
+import android.content.SharedPreferences;
+import android.os.Bundle;
+import android.os.Handler;
+import android.os.Looper;
+import android.preference.PreferenceManager;
+import android.support.v4.content.LocalBroadcastManager;
+import android.util.Log;
+
+import java.io.File;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Run by the syncthing service to convert syncthing events into local broadcasts.
+ * It uses SyncthingService.GetEvents to read the pending events and wait for new events.
+ */
+public class EventProcessor implements SyncthingService.OnWebGuiAvailableListener, Runnable,
+        RestApi.OnReceiveEventListener {
+
+    private static final String TAG = "EventProcessor";
+    private static final String PREF_LAST_SYNC_ID = "last_sync_id";
+
+    private static final String EVENT_BASE_ACTION = "com.nutomic.syncthingandroid.event";
+
+    /**
+     * Minimum interval in seconds at which the events are polled from syncthing and processed.
+     * This intervall will not wake up the device to save battery power.
+     */
+    public static final long EVENT_UPDATE_INTERVAL = TimeUnit.SECONDS.toMillis(15);
+
+    // Use the MainThread for all callbacks and message handling
+    // or we have to track down nasty threading problems.
+    private final Handler mMainThreadHandler = new Handler(Looper.getMainLooper());
+
+    private volatile long mLastEventId = 0;
+    private volatile boolean mShutdown = true;
+
+    private final Context mContext;
+    private final RestApi mApi;
+    private final LocalBroadcastManager mLocalBM;
+    private final Map<String, String> mFolderToPath = new HashMap<>();
+
+    /**
+     * Returns the action used by notification Intents fired for the given Syncthing event.
+     * @param eventName Name of the Syncthing event.
+     * @return Returns the full intent action used for local broadcasts.
+     */
+    public static String getEventIntentAction(String eventName) {
+        return EVENT_BASE_ACTION + "." + eventName.toUpperCase();
+    }
+
+    /**
+     * C'tor
+     * @param context Context of the service using this event processor.
+     * @param api Reference to the RestApi-Instance used for all API calls by this instance of the
+     *            Event processor.
+     */
+    public EventProcessor(Context context, RestApi api) {
+        mContext = context;
+        mApi = api;
+        mLocalBM = LocalBroadcastManager.getInstance(mContext);
+    }
+
+    private void updateFolderMap()
+    {
+        synchronized(mFolderToPath) {
+            mFolderToPath.clear();
+            for (RestApi.Folder folder: mApi.getFolders()) {
+                mFolderToPath.put(folder.id, folder.path);
+            }
+        }
+    }
+
+    /**
+     * @see Runnable
+     */
+    @Override
+    public void run() {
+        // Restore the last event id if the event processor may have been restartet.
+        if (mLastEventId == 0) {
+            mLastEventId = PreferenceManager.getDefaultSharedPreferences(mContext)
+                    .getLong(PREF_LAST_SYNC_ID, 0);
+        }
+
+        // First check if the event number ran backwards.
+        // If that's the case we've to start at zero because syncthing was restartet.
+        mApi.getEvents(0, 1, new RestApi.OnReceiveEventListener() {
+            @Override
+            public void onEvent(long id, String eventType, Bundle eventData) {
+            }
+
+            @Override
+            public void onDone(long lastId) {
+                if (lastId < mLastEventId) mLastEventId = 0;
+
+                Log.d(TAG, "Reading events starting with id " + mLastEventId);
+
+                mApi.getEvents(mLastEventId, 0, EventProcessor.this);
+            }
+        });
+    }
+
+    /**
+     * @see RestApi.OnReceiveEventListener
+     */
+    @Override
+    public void onEvent(final long id, final String eventType, final Bundle eventData) {
+        // If a folder item is contained within the event. Resolve the local path.
+        if (eventData.containsKey("folder")) {
+            String folderPath = null;
+            synchronized (mFolderToPath) {
+                if (mFolderToPath.size() == 0) updateFolderMap();
+                folderPath = mFolderToPath.get(eventData.getString("folder"));
+            }
+
+            if (folderPath != null) {
+                eventData.putString("_localFolderPath",folderPath);
+
+                if (eventData.containsKey("item")) {
+                    final File file = new File(new File(folderPath), eventData.getString("item"));
+
+                    eventData.putString("_localItemPath", file.getPath());
+                }
+            }
+        }
+
+        final Intent eventBroadcastItent = new Intent(EVENT_BASE_ACTION + "." + eventType.toUpperCase());
+        eventBroadcastItent.putExtras(eventData);
+        mLocalBM.sendBroadcast(eventBroadcastItent);
+
+        Log.d(TAG,
+                "Sent local event broadcast " + eventBroadcastItent.getAction() +
+                " including " + eventType.length() + " extra data items."
+        );
+    }
+
+    /**
+     * @see RestApi.OnReceiveEventListener
+     */
+    @Override
+    public void onDone(long id) {
+        if (mLastEventId < id) {
+            mLastEventId = id;
+
+            // Store the last EventId in case we get killed
+            final SharedPreferences sp = PreferenceManager.getDefaultSharedPreferences(mContext);
+
+            //noinspection CommitPrefEdits
+            sp.edit().putLong(PREF_LAST_SYNC_ID, mLastEventId).commit();
+        }
+
+        synchronized (mMainThreadHandler) {
+            if (!mShutdown) {
+                mMainThreadHandler.removeCallbacks(this);
+                mMainThreadHandler.postDelayed(this, EVENT_UPDATE_INTERVAL);
+            }
+        }
+    }
+
+    /**
+     * @see SyncthingService.OnWebGuiAvailableListener
+     */
+    @Override
+    public void onWebGuiAvailable() {
+        Log.d(TAG, "WebGUI available. Starting event processor.");
+
+        updateFolderMap();
+
+        // Remove all pending callbacks and add a new one. This makes sure that only one
+        // event poller is running at any given time.
+        synchronized (mMainThreadHandler) {
+            mShutdown = false;
+            mMainThreadHandler.removeCallbacks(this);
+            mMainThreadHandler.postDelayed(this, EVENT_UPDATE_INTERVAL);
+        }
+    }
+
+    public void shutdown() {
+        Log.d(TAG, "Shutdown event processor.");
+        synchronized (mMainThreadHandler) {
+            mShutdown = true;
+            mMainThreadHandler.removeCallbacks(this);
+        }
+    }
+}

src/main/java/com/nutomic/syncthingandroid/syncthing/GetTask.java

@@ -36,6 +36,7 @@ public class GetTask extends AsyncTask<String, Void, String> {
     public static final String URI_MODEL       = "/rest/db/status";
     public static final String URI_DEVICEID    = "/rest/svc/deviceid";
     public static final String URI_REPORT      = "/rest/svc/report";
+    public static final String URI_EVENTS      = "/rest/events";
 
     private String mHttpsCertPath;
 
@@ -55,9 +56,11 @@ public class GetTask extends AsyncTask<String, Void, String> {
         String fullUri = params[0] + params[1];
         Log.v(TAG, "Calling Rest API at " + fullUri);
 
-        if (params.length == 5) {
+        if (params.length >= 5) {
             LinkedList<NameValuePair> urlParams = new LinkedList<>();
-            urlParams.add(new BasicNameValuePair(params[3], params[4]));
+            for (int paramCounter = 3; paramCounter + 1 < params.length; ) {
+                urlParams.add(new BasicNameValuePair(params[paramCounter++], params[paramCounter++]));
+            }
             fullUri += "?" + URLEncodedUtils.format(urlParams, HTTP.UTF_8);
         }
 

src/main/java/com/nutomic/syncthingandroid/syncthing/RestApi.java

@@ -9,6 +9,8 @@ import android.content.ClipboardManager;
 import android.content.Context;
 import android.content.DialogInterface;
 import android.content.Intent;
+import android.os.AsyncTask;
+import android.os.Bundle;
 import android.util.Log;
 import android.widget.Toast;
 
@@ -174,6 +176,11 @@ public class RestApi implements SyncthingService.OnWebGuiAvailableListener,
      */
     private HashMap<String, Model> mCachedModelInfo = new HashMap<>();
 
+    /**
+     * Stores a hash map to resolve folders to paths for events.
+     */
+    private final Map<String, String> mCacheFolderPathLookup = new HashMap<String, String>();
+
     public RestApi(Context context, String url, String apiKey, String guiUser, String guiPassword,
                    OnApiAvailableListener listener) {
         mContext = context;
@@ -618,7 +625,6 @@ public class RestApi implements SyncthingService.OnWebGuiAvailableListener,
                 : 100;
     }
 
-
     /**
      * Listener for {@link #getModel}.
      */
@@ -626,6 +632,25 @@ public class RestApi implements SyncthingService.OnWebGuiAvailableListener,
         public void onReceiveModel(String folderId, Model model);
     }
 
+    /**
+     * Listener for {@link #getEvents}.
+     */
+    public interface OnReceiveEventListener {
+        /**
+         * Called for each event.
+         * @param id ID of the event. Monotonously increasing.
+         * @param eventType Name of the event. (See Syncthing documentation)
+         * @param eventData Bundle containing the data fields of the event as data elements.
+         */
+        void onEvent(final long id, final String eventType, final Bundle eventData);
+
+        /**
+         * Called after all available events have been processed.
+         * @param lastId The id of the last event processed. Should be used as a starting point for
+         *               the next round of event processing.
+         */
+        void onDone(long lastId);
+    }
     /**
      * Returns status information about the folder with the given id.
      */
@@ -660,6 +685,105 @@ public class RestApi implements SyncthingService.OnWebGuiAvailableListener,
         }.execute(mUrl, GetTask.URI_MODEL, mApiKey, "folder", folderId);
     }
 
+    /**
+     * Refreshes the lookup table to convert folder names to paths for events.
+     */
+    private String getPathForFolder(String folderName) {
+        synchronized(mCacheFolderPathLookup) {
+            if (!mCacheFolderPathLookup.containsKey(folderName)) {
+                mCacheFolderPathLookup.clear();
+                for (Folder folder : getFolders()) {
+                    mCacheFolderPathLookup.put(folder.id, folder.path);
+                }
+            }
+
+            return mCacheFolderPathLookup.get(folderName);
+        }
+    }
+
+    private void clearFolderCache() {
+        synchronized(mCacheFolderPathLookup) {
+            mCacheFolderPathLookup.clear();
+        }
+    }
+
+    /**
+     * Retrieves the events that have accumulated since the given event id.
+     * The OnReceiveEventListeners onEvent method is called for each event.
+     */
+    public final void getEvents(final long sinceId, final long limit, final OnReceiveEventListener listener) {
+
+        GetTask eventGetTask = new GetTask(mHttpsCertPath) {
+            @Override
+            protected void onPostExecute(String s) {
+                if (s == null)
+                    return;
+
+                try {
+                    JSONArray jsonEvents = new JSONArray(s);
+
+                    long lastId = 0;
+
+                    for (int i = 0; i < jsonEvents.length(); i++) {
+
+                        final JSONObject json = jsonEvents.getJSONObject(i);
+                        final String eventType = json.getString("type").toLowerCase();
+                        final long id = json.getLong("id");
+
+                        Bundle dataBundle = null;
+
+                        if (lastId < id) lastId = id;
+
+                        switch (eventType) {
+                            // This special shortcut can be used if data only contains strings.
+                            // It just copies everything into a bundle.
+                            case "itemfinished":
+                            case "foldercompletion":
+                            case "deviceconnected":
+                            case "devicediscovered":
+                            case "statechanged":
+                                dataBundle = new Bundle();
+                                JSONObject data = json.getJSONObject("data");
+
+                                for (Iterator<String> keyIterator = data.keys(); keyIterator.hasNext();) {
+                                    String key = keyIterator.next();
+                                    dataBundle.putString(key, data.getString(key));
+                                }
+
+                                // If the event contains a folder keyword but no path synthesise
+                                // a path keyword to ease processing of the event.
+                                String folder = data.optString("folder", null);
+                                if ((folder != null) && (data.optString("path", null) == null)) {
+                                    String folderPath = getPathForFolder(folder);
+                                    if (folderPath != null) {
+                                        dataBundle.putString("path", folderPath);
+                                    }
+                                }
+
+                                break;
+
+                            // Ignored events.
+                            case "ping":
+                                break;
+
+                            default:
+                                Log.d(TAG, "Unhandled event " + json.getString("type"));
+                        }
+
+                        if (dataBundle != null) listener.onEvent(id, eventType, dataBundle);
+                    }
+
+                    listener.onDone(lastId);
+                }
+                catch (JSONException e) {
+                    Log.w(TAG, "Failed to read events", e);
+                }
+            }
+        };
+
+        eventGetTask.executeOnExecutor(AsyncTask.THREAD_POOL_EXECUTOR, mUrl, GetTask.URI_EVENTS, mApiKey, "since", String.valueOf(sinceId), "limit", String.valueOf(limit));
+    }
+
     /**
      * Returns the folder's state as a localized string.
      *
@@ -802,6 +926,8 @@ public class RestApi implements SyncthingService.OnWebGuiAvailableListener,
             Log.w(TAG, "Failed to edit folder " + folder.id + " at " + folder.path, e);
             return false;
         }
+
+        clearFolderCache();
         return true;
     }
 
@@ -825,6 +951,8 @@ public class RestApi implements SyncthingService.OnWebGuiAvailableListener,
             Log.w(TAG, "Failed to edit folder", e);
             return false;
         }
+
+        clearFolderCache();
         return true;
     }
 

src/main/java/com/nutomic/syncthingandroid/syncthing/SyncthingService.java

@@ -5,17 +5,20 @@ import android.app.AlertDialog;
 import android.app.NotificationManager;
 import android.app.PendingIntent;
 import android.app.Service;
+import android.content.BroadcastReceiver;
 import android.content.Context;
 import android.content.DialogInterface;
 import android.content.Intent;
 import android.content.IntentFilter;
 import android.content.SharedPreferences;
+import android.net.Uri;
 import android.os.AsyncTask;
 import android.os.Build;
 import android.os.Environment;
 import android.os.IBinder;
 import android.preference.PreferenceManager;
 import android.support.v4.app.NotificationCompat;
+import android.support.v4.content.LocalBroadcastManager;
 import android.util.Log;
 import android.util.Pair;
 import android.widget.Toast;
@@ -99,10 +102,28 @@ public class SyncthingService extends Service implements
 
     private RestApi mApi;
 
+    private EventProcessor mEventProcessor;
+
     private LinkedList<FolderObserver> mObservers = new LinkedList<>();
 
     private final SyncthingServiceBinder mBinder = new SyncthingServiceBinder(this);
 
+    /**
+     * Processes the local broadcast message if an item was finished.
+     * Launches the media scanner to update the media library.
+     */
+    private final BroadcastReceiver mItemFinishedBroadcastReceiver = new BroadcastReceiver() {
+
+        @Override
+        public void onReceive(Context context, Intent intent) {
+            final File updatedFile = new File(intent.getStringExtra("_localItemPath"));
+
+            Log.d(TAG, "Notified media scanner about " + updatedFile.toString());
+
+            context.sendBroadcast(new Intent(Intent.ACTION_MEDIA_SCANNER_SCAN_FILE, Uri.fromFile(updatedFile)));
+        }
+    };
+
     /**
      * Callback for when the Syncthing web interface becomes first available after service start.
      */
@@ -219,6 +240,7 @@ public class SyncthingService extends Service implements
             if (mConfig != null) {
                 mCurrentState = State.STARTING;
                 registerOnWebGuiAvailableListener(mApi);
+                registerOnWebGuiAvailableListener(mEventProcessor);
                 new PollWebGuiAvailableTaskImpl(getFilesDir() + "/" + HTTPS_CERT_FILE)
                         .execute(mConfig.getWebGuiUrl());
                 mRunnable = new SyncthingRunnable(this, SyncthingRunnable.Command.main);
@@ -301,6 +323,7 @@ public class SyncthingService extends Service implements
         registerReceiver(mDeviceStateHolder, new IntentFilter(Intent.ACTION_BATTERY_CHANGED));
         new StartupTask(sp.getString("gui_user",""), sp.getString("gui_password","")).execute();
         sp.registerOnSharedPreferenceChangeListener(this);
+        LocalBroadcastManager.getInstance(this).registerReceiver(this.mItemFinishedBroadcastReceiver, new IntentFilter(EventProcessor.getEventIntentAction("itemfinished")));
     }
 
     /**
@@ -364,7 +387,11 @@ public class SyncthingService extends Service implements
                     }).start();
                 }
             });
+
+            mEventProcessor = new EventProcessor(SyncthingService.this, mApi);
+
             registerOnWebGuiAvailableListener(mApi);
+            registerOnWebGuiAvailableListener(mEventProcessor);
             Log.i(TAG, "Web GUI will be available at " + mConfig.getWebGuiUrl());
             updateState();
         }
@@ -385,9 +412,13 @@ public class SyncthingService extends Service implements
         shutdown();
         SharedPreferences sp = PreferenceManager.getDefaultSharedPreferences(this);
         sp.unregisterOnSharedPreferenceChangeListener(this);
+        LocalBroadcastManager.getInstance(this).unregisterReceiver(this.mItemFinishedBroadcastReceiver);
     }
 
     private void shutdown() {
+        if (mEventProcessor != null)
+            mEventProcessor.shutdown();
+
         if (mRunnable != null)
             mRunnable.killSyncthing();
 
@@ -586,14 +617,12 @@ public class SyncthingService extends Service implements
         } finally {
             try {
                 if (is != null)
-                  is.close();
+                    is.close();
                 if (os != null)
-                  os.close();
+                    os.close();
             } catch (IOException e) {
                 Log.w(TAG, "Failed to close stream", e);
             }
         }
     }
-
-
 }