Change/simplify Observable interface

Previously, the Observable interface allowed registering callbacks to
receive each update to a value, in a thread-safe way.  However, this adds
unnecessary complexity and greatly complicates the implementation of
callbacks to track the camera ready-state.

Replacing Callback<T> with Runnables, which can simply call
Observable.get() to retrieve the latest value makes implementing
ready-state significantly simpler.

Bug: 18934542
Change-Id: I00d512f3c380148c0e9cb52352b2f304494e5e5a

10 files changed, 66 insertions, 247 deletions

diff --git a/src/com/android/camera/async/CallbackRunnable.java b/src/com/android/camera/async/CallbackRunnable.java
deleted file mode 100644
index f51504370..000000000
--- a/src/com/android/camera/async/CallbackRunnable.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * Copyright (C) 2014 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.android.camera.async;
-
-import com.android.camera.util.Callback;
-
-/**
- * A {@link Callback} which just invokes a {@link Runnable}.
- *
- * @param <T>
- */
-public class CallbackRunnable<T> implements Callback<T> {
-    private final Runnable mRunnable;
-
-    public CallbackRunnable(Runnable runnable) {
-        mRunnable = runnable;
-    }
-
-    @Override
-    public void onCallback(T result) {
-        mRunnable.run();
-    }
-}
diff --git a/src/com/android/camera/async/ConcurrentState.java b/src/com/android/camera/async/ConcurrentState.java
index f56aa381e..61cd342b9 100644
--- a/src/com/android/camera/async/ConcurrentState.java
+++ b/src/com/android/camera/async/ConcurrentState.java
@@ -16,10 +16,8 @@
 
 package com.android.camera.async;
 
-import com.android.camera.util.Callback;
-
-import java.util.HashSet;
 import java.util.Set;
+import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.Executor;
 
 import javax.annotation.CheckReturnValue;
@@ -32,36 +30,32 @@ import javax.annotation.ParametersAreNonnullByDefault;
  */
 @ParametersAreNonnullByDefault
 public class ConcurrentState<T> implements Updatable<T>, Observable<T> {
-
-    private static class ExecutorListenerPair<T> {
+    private static class ExecutorListenerPair implements Runnable {
         private final Executor mExecutor;
-        private final Callback<T> mListener;
+        private final Runnable mListener;
 
-        public ExecutorListenerPair(Executor executor, Callback<T> listener) {
+        public ExecutorListenerPair(Executor executor, Runnable listener) {
             mExecutor = executor;
             mListener = listener;
         }
 
         /**
-         * Runs the callback on the executor with the given value.
+         * Runs the callback on the executor.
          */
-        public void run(final T t) {
-            mExecutor.execute(new Runnable() {
-                @Override
-                public void run() {
-                    mListener.onCallback(t);
-                }
-            });
+        @Override
+        public void run() {
+            mExecutor.execute(mListener);
         }
     }
 
-    private final Object mLock;
-    private final Set<ExecutorListenerPair<? super T>> mListeners;
-    private T mValue;
+    private final Set<ExecutorListenerPair> mListeners;
+    private volatile T mValue;
 
     public ConcurrentState(T initialValue) {
-        mLock = new Object();
-        mListeners = new HashSet<>();
+        // Callbacks are typically only added and removed at startup/shutdown,
+        // but {@link #update} is often called at high-frequency. So, using a
+        // read-optimized data structure is appropriate here.
+        mListeners = new CopyOnWriteArraySet<>();
         mValue = initialValue;
     }
 
@@ -70,35 +64,24 @@ public class ConcurrentState<T> implements Updatable<T>, Observable<T> {
      */
     @Override
     public void update(T newValue) {
-        synchronized (mLock) {
-            mValue = newValue;
-            // Invoke executors.execute within mLock to guarantee that
-            // callbacks are serialized into their respective executor in
-            // the proper order.
-            for (ExecutorListenerPair<? super T> pair : mListeners) {
-                pair.run(newValue);
-            }
+        mValue = newValue;
+        for (ExecutorListenerPair pair : mListeners) {
+            pair.run();
         }
     }
 
     @CheckReturnValue
     @Nonnull
     @Override
-    public SafeCloseable addCallback(Callback<T> callback, Executor executor) {
-        synchronized (mLock) {
-            final ExecutorListenerPair<? super T> pair =
-                    new ExecutorListenerPair<>(executor, callback);
-            mListeners.add(pair);
-
-            return new SafeCloseable() {
-                @Override
-                public void close() {
-                    synchronized (mLock) {
-                        mListeners.remove(pair);
-                    }
-                }
-            };
-        }
+    public SafeCloseable addCallback(Runnable callback, Executor executor) {
+        final ExecutorListenerPair pair = new ExecutorListenerPair(executor, callback);
+        mListeners.add(pair);
+        return new SafeCloseable() {
+            @Override
+            public void close() {
+                mListeners.remove(pair);
+            }
+        };
     }
 
     /**
@@ -109,8 +92,6 @@ public class ConcurrentState<T> implements Updatable<T>, Observable<T> {
     @Nonnull
     @Override
     public T get() {
-        synchronized (mLock) {
-            return mValue;
-        }
+        return mValue;
     }
 }
diff --git a/src/com/android/camera/async/FilteredCallback.java b/src/com/android/camera/async/FilteredCallback.java
deleted file mode 100644
index 382a79c01..000000000
--- a/src/com/android/camera/async/FilteredCallback.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * Copyright (C) 2015 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.android.camera.async;
-
-import com.android.camera.util.Callback;
-import com.google.common.base.Objects;
-
-import javax.annotation.Nonnull;
-
-/**
- * Wraps a callback to filter out duplicate updates.
- */
-public class FilteredCallback<T> implements Callback<T> {
-    private final Callback<? super T> mCallback;
-    private T mLatestValue;
-
-    public FilteredCallback(Callback<? super T> callback) {
-        mCallback = callback;
-        mLatestValue = null;
-    }
-
-    @Override
-    public void onCallback(@Nonnull T newValue) {
-        if (!Objects.equal(mLatestValue, newValue)) {
-            mLatestValue = newValue;
-            mCallback.onCallback(newValue);
-        }
-    }
-}
diff --git a/src/com/android/camera/async/ForwardingObservable.java b/src/com/android/camera/async/ForwardingObservable.java
index 716a3cc82..03a1572fe 100644
--- a/src/com/android/camera/async/ForwardingObservable.java
+++ b/src/com/android/camera/async/ForwardingObservable.java
@@ -35,7 +35,7 @@ public abstract class ForwardingObservable<T> implements Observable<T> {
     @Nonnull
     @Override
     @CheckReturnValue
-    public SafeCloseable addCallback(Callback<T> callback, Executor executor) {
+    public SafeCloseable addCallback(Runnable callback, Executor executor) {
         return mDelegate.addCallback(callback, executor);
     }
 
diff --git a/src/com/android/camera/async/ListenableConcurrentState.java b/src/com/android/camera/async/ListenableConcurrentState.java
index b54763cfc..aa691d2d8 100644
--- a/src/com/android/camera/async/ListenableConcurrentState.java
+++ b/src/com/android/camera/async/ListenableConcurrentState.java
@@ -43,7 +43,7 @@ public class ListenableConcurrentState<T> implements Listenable<T> {
      * Sets the callback, removing any existing callback first.
      */
     @Override
-    public void setCallback(Callback<T> callback) {
+    public void setCallback(final Callback<T> callback) {
         synchronized (mLock) {
             if (mClosed) {
                 return;
@@ -52,7 +52,12 @@ public class ListenableConcurrentState<T> implements Listenable<T> {
                 // Unregister any existing callback
                 mExistingCallbackHandle.close();
             }
-            mExistingCallbackHandle = mState.addCallback(callback, mExecutor);
+            mExistingCallbackHandle = mState.addCallback(new Runnable() {
+                @Override
+                public void run() {
+                    callback.onCallback(mState.get());
+                }
+            }, mExecutor);
         }
     }
 
diff --git a/src/com/android/camera/async/Observable.java b/src/com/android/camera/async/Observable.java
index 7d56da09c..3ebda9a2f 100644
--- a/src/com/android/camera/async/Observable.java
+++ b/src/com/android/camera/async/Observable.java
@@ -16,7 +16,6 @@
 
 package com.android.camera.async;
 
-import com.android.camera.util.Callback;
 import com.google.common.base.Supplier;
 
 import java.util.concurrent.Executor;
@@ -33,8 +32,11 @@ import javax.annotation.concurrent.ThreadSafe;
 @ParametersAreNonnullByDefault
 public interface Observable<T> extends Supplier<T> {
     /**
-     * Adds the given callback, returning a handle to be closed when the
-     * callback must be deregistered.
+     * Adds the given callback to be invoked upon changes, returning a handle to
+     * be closed when the callback must be deregistered.
+     * <p>
+     * Note that the callback may be invoked multiple times even if the value
+     * has not actually changed.
      *
      * @param callback The callback to add.
      * @param executor The executor on which the callback will be invoked.
@@ -42,7 +44,7 @@ public interface Observable<T> extends Supplier<T> {
      */
     @Nonnull
     @CheckReturnValue
-    public SafeCloseable addCallback(Callback<T> callback, Executor executor);
+    public SafeCloseable addCallback(Runnable callback, Executor executor);
 
     /**
      * @return The current/latest value.
diff --git a/src/com/android/camera/async/ObservableCombiner.java b/src/com/android/camera/async/ObservableCombiner.java
index c7b94eaa9..d0ffb33b1 100644
--- a/src/com/android/camera/async/ObservableCombiner.java
+++ b/src/com/android/camera/async/ObservableCombiner.java
@@ -16,7 +16,6 @@
 
 package com.android.camera.async;
 
-import com.android.camera.util.Callback;
 import com.google.common.base.Function;
 import com.google.common.collect.ImmutableList;
 
@@ -27,7 +26,6 @@ import java.util.concurrent.Executor;
 import javax.annotation.CheckReturnValue;
 import javax.annotation.Nonnull;
 import javax.annotation.ParametersAreNonnullByDefault;
-import javax.annotation.concurrent.GuardedBy;
 import javax.annotation.concurrent.ThreadSafe;
 
 /**
@@ -43,34 +41,10 @@ final class ObservableCombiner<I, O> implements Observable<O> {
     private final ImmutableList<Observable<I>> mInputs;
     private final Function<List<I>, O> mFunction;
 
-    private final Object mLock;
-
-    @GuardedBy("mLock")
-    private final ConcurrentState<O> mListenerNotifier;
-
-    @GuardedBy("mLock")
-    private final List<SafeCloseable> mInputCallbackHandles;
-
-    @GuardedBy("mLock")
-    private int mNumRegisteredCallbacks;
-
-    /**
-     * The thread-safe callback to be registered with each input.
-     */
-    private final Updatable<I> mInputCallback = new Updatable<I>() {
-        public void update(I ignored) {
-            mListenerNotifier.update(get());
-        }
-    };
-
     private ObservableCombiner(List<? extends Observable<I>> inputs,
-            Function<List<I>, O> function, O initialValue) {
+            Function<List<I>, O> function) {
         mInputs = ImmutableList.copyOf(inputs);
         mFunction = function;
-        mListenerNotifier = new ConcurrentState<>(initialValue);
-        mLock = new Object();
-        mInputCallbackHandles = new ArrayList<>();
-        mNumRegisteredCallbacks = 0;
     }
 
     /**
@@ -86,70 +60,20 @@ final class ObservableCombiner<I, O> implements Observable<O> {
      */
     static <I, O> Observable<O> transform(List<? extends Observable<I>> inputs,
             Function<List<I>, O> function) {
-        // Compute the initial value.
-        ArrayList<I> deps = new ArrayList<>();
-        for (Observable<? extends I> input : inputs) {
-            deps.add(input.get());
-        }
-        O initialValue = function.apply(deps);
-
-        return new ObservableCombiner<>(inputs, function, initialValue);
-    }
-
-    @GuardedBy("mLock")
-    private void addCallbacksToInputs() {
-        for (Observable<I> observable : mInputs) {
-            final SafeCloseable callbackHandle =
-                    Observables.addThreadSafeCallback(observable, mInputCallback);
-
-            mInputCallbackHandles.add(callbackHandle);
-        }
-    }
-
-    @GuardedBy("mLock")
-    private void removeCallbacksFromInputs() {
-        for (SafeCloseable callbackHandle : mInputCallbackHandles) {
-            callbackHandle.close();
-        }
+        return new ObservableCombiner<>(inputs, function);
     }
 
     @Nonnull
     @Override
     @CheckReturnValue
-    public SafeCloseable addCallback(final Callback<O> callback, Executor executor) {
-        // When a callback is added to this, the "output", we must ensure that
-        // callbacks are registered with all of the inputs so that they can be
-        // forwarded properly.
-        // Instead of adding another callback to each input for each callback
-        // registered with the output, callbacks are registered when the first
-        // output callback is added, and removed when the last output callback
-        // is removed.
+    public SafeCloseable addCallback(Runnable callback, Executor executor) {
+        Lifetime callbackLifetime = new Lifetime();
 
-        synchronized (mLock) {
-            if (mNumRegisteredCallbacks == 0) {
-                addCallbacksToInputs();
-            }
-            mNumRegisteredCallbacks++;
+        for (Observable<I> input : mInputs) {
+            callbackLifetime.add(input.addCallback(callback, executor));
         }
 
-        // Wrap the callback in a {@link FilteredCallback} to prevent many
-        // duplicate/cascading updates even if the output does not change.
-        final SafeCloseable resultingCallbackHandle = mListenerNotifier.addCallback(
-                new FilteredCallback<O>(callback), executor);
-
-        return new SafeCloseable() {
-            @Override
-            public void close() {
-                resultingCallbackHandle.close();
-
-                synchronized (mLock) {
-                    mNumRegisteredCallbacks--;
-                    if (mNumRegisteredCallbacks == 0) {
-                        removeCallbacksFromInputs();
-                    }
-                }
-            }
-        };
+        return callbackLifetime;
     }
 
     @Nonnull
diff --git a/src/com/android/camera/async/Observables.java b/src/com/android/camera/async/Observables.java
index 900e02bf6..459198854 100644
--- a/src/com/android/camera/async/Observables.java
+++ b/src/com/android/camera/async/Observables.java
@@ -60,14 +60,8 @@ public class Observables {
             @CheckReturnValue
             @Nonnull
             @Override
-            public SafeCloseable addCallback(final Callback<T> callback,
-                    Executor executor) {
-                return input.addCallback(new Callback<F>() {
-                    @Override
-                    public void onCallback(F result) {
-                        callback.onCallback(function.apply(result));
-                    }
-                }, executor);
+            public SafeCloseable addCallback(Runnable callback, Executor executor) {
+                return input.addCallback(callback, executor);
             }
         };
     }
@@ -86,7 +80,7 @@ public class Observables {
      * @return An observable which has the given constant value.
      */
     @Nonnull
-    public static <T> Observable<T> of(final @Nullable T constant) {
+    public static <T> Observable<T> of(final T constant) {
         return new Observable<T>() {
             @Nonnull
             @Override
@@ -97,7 +91,7 @@ public class Observables {
             @CheckReturnValue
             @Nonnull
             @Override
-            public SafeCloseable addCallback(Callback<T> callback, Executor executor) {
+            public SafeCloseable addCallback(Runnable callback, Executor executor) {
                 return NOOP_CALLBACK_HANDLE;
             }
         };
@@ -105,12 +99,12 @@ public class Observables {
 
     @Nonnull
     @CheckReturnValue
-    public static <T> SafeCloseable addThreadSafeCallback(Observable<T> observable,
+    public static <T> SafeCloseable addThreadSafeCallback(final Observable<T> observable,
             final Updatable<T> callback) {
-        return observable.addCallback(new Callback<T>() {
+        return observable.addCallback(new Runnable() {
             @Override
-            public void onCallback(T result) {
-                callback.update(result);
+            public void run() {
+                callback.update(observable.get());
             }
         }, MoreExecutors.sameThreadExecutor());
     }
diff --git a/src/com/android/camera/one/v2/common/BasicCameraFactory.java b/src/com/android/camera/one/v2/common/BasicCameraFactory.java
index 4cca82bc8..4b6ba1cca 100644
--- a/src/com/android/camera/one/v2/common/BasicCameraFactory.java
+++ b/src/com/android/camera/one/v2/common/BasicCameraFactory.java
@@ -98,12 +98,9 @@ public class BasicCameraFactory {
         // changes to apply the new setting.
         // Also, de-register these callbacks when the camera is closed (to
         // not leak memory).
-        SafeCloseable zoomCallback = zoom.addCallback(
-                new CallbackRunnable<Float>(mPreviewStarter), threadPool);
+        SafeCloseable zoomCallback = zoom.addCallback(mPreviewStarter, threadPool);
         lifetime.add(zoomCallback);
-        SafeCloseable flashCallback = flash.addCallback(
-                new CallbackRunnable<OneCamera.PhotoCaptureParameters.Flash>(mPreviewStarter),
-                threadPool);
+        SafeCloseable flashCallback = flash.addCallback(mPreviewStarter, threadPool);
         lifetime.add(flashCallback);
 
         int sensorOrientation =
diff --git a/src/com/android/camera/settings/SettingObserver.java b/src/com/android/camera/settings/SettingObserver.java
index a38f3aff9..8d68cbfcf 100644
--- a/src/com/android/camera/settings/SettingObserver.java
+++ b/src/com/android/camera/settings/SettingObserver.java
@@ -20,8 +20,6 @@ import com.android.camera.async.ExecutorCallback;
 import com.android.camera.async.FilteredUpdatable;
 import com.android.camera.async.Observable;
 import com.android.camera.async.SafeCloseable;
-import com.android.camera.async.Updatable;
-import com.android.camera.util.Callback;
 
 import java.util.concurrent.Executor;
 
@@ -36,18 +34,17 @@ import javax.annotation.concurrent.ThreadSafe;
 @ThreadSafe
 public final class SettingObserver<T> implements Observable<T> {
     private class Listener implements SettingsManager.OnSettingChangedListener, SafeCloseable {
-        private final Updatable<? super T> mCallback;
+        private final Runnable mRunnable;
+        private final Executor mExecutor;
 
-        private Listener(Updatable<? super T> callback) {
-            mCallback = callback;
+        private Listener(Runnable runnable, Executor executor) {
+            mRunnable = runnable;
+            mExecutor = executor;
         }
 
         @Override
         public void onSettingChanged(SettingsManager settingsManager, String key) {
-            T t = get();
-            if (t != null) {
-                mCallback.update(t);
-            }
+            mExecutor.execute(mRunnable);
         }
 
         @Override
@@ -89,9 +86,8 @@ public final class SettingObserver<T> implements Observable<T> {
     @CheckReturnValue
     @Nonnull
     @Override
-    public SafeCloseable addCallback(@Nonnull Callback<T> callback, @Nonnull Executor executor) {
-        final Listener listener =
-                new Listener(new FilteredUpdatable<>(new ExecutorCallback<>(callback, executor)));
+    public SafeCloseable addCallback(@Nonnull Runnable callback, @Nonnull Executor executor) {
+        Listener listener = new Listener(callback, executor);
         mSettingsManager.addListener(listener);
         return listener;
     }