Avoid Concurrent Exception in BaseDataProducer.

Add locks to protect the list of consumers.
We are seeing concurrent modification exceptions because
DeviceStateManager sends changes on a separate thread.

Bug: 241464751
Test: n/a could not generate a non-flaky test.
Change-Id: Ie49c0f4f16db39187c98609ac57734e468c3f507

libs/WindowManager/Jetpack/src/androidx/window/util/BaseDataProducer.java

@@ -16,6 +16,7 @@
 
 package androidx.window.util;
 
+import androidx.annotation.GuardedBy;
 import androidx.annotation.NonNull;
 
 import java.util.LinkedHashSet;
@@ -25,25 +26,45 @@ import java.util.function.Consumer;
 
 /**
  * Base class that provides the implementation for the callback mechanism of the
- * {@link DataProducer} API.
+ * {@link DataProducer} API.  This class is thread safe for adding, removing, and notifying
+ * consumers.
  *
  * @param <T> The type of data this producer returns through {@link DataProducer#getData}.
  */
 public abstract class BaseDataProducer<T> implements DataProducer<T> {
+
+    private final Object mLock = new Object();
+    @GuardedBy("mLock")
     private final Set<Consumer<T>> mCallbacks = new LinkedHashSet<>();
 
+    /**
+     * Adds a callback to the set of callbacks listening for data. Data is delivered through
+     * {@link BaseDataProducer#notifyDataChanged(Object)}. This method is thread safe. Callers
+     * should ensure that callbacks are thread safe.
+     * @param callback that will receive data from the producer.
+     */
     @Override
     public final void addDataChangedCallback(@NonNull Consumer<T> callback) {
-        mCallbacks.add(callback);
-        Optional<T> currentData = getCurrentData();
-        currentData.ifPresent(callback);
-        onListenersChanged(mCallbacks);
+        synchronized (mLock) {
+            mCallbacks.add(callback);
+            Optional<T> currentData = getCurrentData();
+            currentData.ifPresent(callback);
+            onListenersChanged(mCallbacks);
+        }
     }
 
+    /**
+     * Removes a callback to the set of callbacks listening for data. This method is thread safe
+     * for adding.
+     * @param callback that was registered in
+     * {@link BaseDataProducer#addDataChangedCallback(Consumer)}.
+     */
     @Override
     public final void removeDataChangedCallback(@NonNull Consumer<T> callback) {
-        mCallbacks.remove(callback);
-        onListenersChanged(mCallbacks);
+        synchronized (mLock) {
+            mCallbacks.remove(callback);
+            onListenersChanged(mCallbacks);
+        }
     }
 
     protected void onListenersChanged(Set<Consumer<T>> callbacks) {}
@@ -56,11 +77,14 @@ public abstract class BaseDataProducer<T> implements DataProducer<T> {
 
     /**
      * Called to notify all registered consumers that the data provided
-     * by {@link DataProducer#getData} has changed.
+     * by {@link DataProducer#getData} has changed. Calls to this are thread save but callbacks need
+     * to ensure thread safety.
      */
     protected void notifyDataChanged(T value) {
-        for (Consumer<T> callback : mCallbacks) {
-            callback.accept(value);
+        synchronized (mLock) {
+            for (Consumer<T> callback : mCallbacks) {
+                callback.accept(value);
+            }
         }
     }
 }