Add addTransferListener method to DataSource and provide BaseDataSource.

The new method allows to add transfer listeners (e.g. the BandwidthMeter) after
the data source has been created. To simplify the implementation for
subclasses, this change also introduces a BaseDataSource which handles
the list of listeners and the listener registration.

-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=202649563

RELEASENOTES.md

@@ -25,6 +25,10 @@
     gather bandwidth information.
   * Pass `TransferListener` to `MediaSource`s to listen to media data transfers.
     Always null at the moment.
+  * Add method to `DataSource` to add `TransferListener`s. Custom `DataSource`s
+    directly reading data should implement `BaseDataSource` to handle the
+    registration correctly. Custom `DataSource`'s forwarding to other sources
+    should forward all calls to `addTransferListener`.
 * Error handling:
   * Allow configuration of the Loader retry delay
     ([#3370](https://github.com/google/ExoPlayer/issues/3370)).

library/core/src/main/java/com/google/android/exoplayer2/upstream/BaseDataSource.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright (C) 2018 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.google.android.exoplayer2.upstream;
+
+import java.util.ArrayList;
+
+/**
+ * Base {@link DataSource} implementation to keep a list of {@link TransferListener}s.
+ *
+ * <p>Subclasses must call {@link #transferStarted(DataSpec)}, {@link #bytesTransferred(int)}, and
+ * {@link #transferEnded()} to inform listeners of data transfers.
+ */
+public abstract class BaseDataSource implements DataSource {
+
+  private final @DataSource.Type int type;
+  private final ArrayList<TransferListener<? super DataSource>> listeners;
+
+  /**
+   * Creates base data source for a data source of the specified type.
+   *
+   * @param type The {@link DataSource.Type} of the data source.
+   */
+  protected BaseDataSource(@DataSource.Type int type) {
+    this.type = type;
+    this.listeners = new ArrayList<>(/* initialCapacity= */ 1);
+  }
+
+  @Override
+  public final void addTransferListener(TransferListener<? super DataSource> transferListener) {
+    listeners.add(transferListener);
+  }
+
+  /**
+   * Notifies listeners that data transfer for the specified {@link DataSpec} started.
+   *
+   * @param dataSpec {@link DataSpec} describing the data being transferred.
+   */
+  protected final void transferStarted(DataSpec dataSpec) {
+    for (int i = 0; i < listeners.size(); i++) {
+      listeners.get(i).onTransferStart(/* source= */ this, dataSpec);
+    }
+  }
+
+  /**
+   * Notifies listeners that bytes were transferred.
+   *
+   * @param bytesTransferred The number of bytes transferred since the previous call to this method
+   *     (or if the first call, since the transfer was started).
+   */
+  protected final void bytesTransferred(int bytesTransferred) {
+    for (int i = 0; i < listeners.size(); i++) {
+      listeners.get(i).onBytesTransferred(/* source= */ this, bytesTransferred);
+    }
+  }
+
+  /** Notifies listeners that a transfer ended. */
+  protected final void transferEnded() {
+    for (int i = 0; i < listeners.size(); i++) {
+      listeners.get(i).onTransferEnd(/* source= */ this);
+    }
+  }
+}

library/core/src/main/java/com/google/android/exoplayer2/upstream/DataSource.java

@@ -16,9 +16,12 @@
 package com.google.android.exoplayer2.upstream;
 
 import android.net.Uri;
+import android.support.annotation.IntDef;
 import android.support.annotation.Nullable;
 import com.google.android.exoplayer2.C;
 import java.io.IOException;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -28,6 +31,15 @@ import java.util.Map;
  */
 public interface DataSource {
 
+  /** Type of a data source. */
+  @Retention(RetentionPolicy.SOURCE)
+  @IntDef({TYPE_REMOTE, TYPE_LOCAL})
+  public @interface Type {}
+  /** Data source which loads data from a remote (network) source. */
+  int TYPE_REMOTE = 0;
+  /** Data source which loads data from a local (on-device) source. */
+  int TYPE_LOCAL = 1;
+
   /**
    * A factory for {@link DataSource} instances.
    */
@@ -37,7 +49,16 @@ public interface DataSource {
      * Creates a {@link DataSource} instance.
      */
     DataSource createDataSource();
+  }
 
+  /**
+   * Adds a {@link TransferListener} to listen to data transfers. Must be called before the first
+   * call to {@link #open(DataSpec)}.
+   *
+   * @param transferListener A {@link TransferListener}.
+   */
+  default void addTransferListener(TransferListener<? super DataSource> transferListener) {
+    // TODO: Make non-default once all DataSources implement this method.
   }
 
   /**
@@ -102,5 +123,4 @@ public interface DataSource {
    * @throws IOException If an error occurs closing the source.
    */
   void close() throws IOException;
-
 }

library/core/src/test/java/com/google/android/exoplayer2/upstream/BaseDataSourceTest.java

@@ -0,0 +1,122 @@
+/*
+ * Copyright (C) 2018 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.google.android.exoplayer2.upstream;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import android.net.Uri;
+import android.support.annotation.Nullable;
+import com.google.android.exoplayer2.C;
+import java.io.IOException;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.robolectric.RobolectricTestRunner;
+
+/** Unit test for {@link BaseDataSource}. */
+@RunWith(RobolectricTestRunner.class)
+public class BaseDataSourceTest {
+
+  @Test
+  public void dataTransfer_withLocalSource_isReported() throws IOException {
+    TestSource testSource = new TestSource(DataSource.TYPE_LOCAL);
+    TestTransferListener transferListener = new TestTransferListener();
+    testSource.addTransferListener(transferListener);
+
+    DataSpec dataSpec = new DataSpec(Uri.EMPTY);
+    testSource.open(dataSpec);
+    testSource.read(/* buffer= */ null, /* offset= */ 0, /* readLength= */ 100);
+    testSource.close();
+
+    assertThat(transferListener.lastTransferStartSource).isSameAs(testSource);
+    assertThat(transferListener.lastBytesTransferredSource).isSameAs(testSource);
+    assertThat(transferListener.lastTransferEndSource).isSameAs(testSource);
+    assertThat(transferListener.lastTransferStartDataSpec).isEqualTo(dataSpec);
+    assertThat(transferListener.lastBytesTransferred).isEqualTo(100);
+  }
+
+  @Test
+  public void dataTransfer_withRemoteSource_isReported() throws IOException {
+    TestSource testSource = new TestSource(DataSource.TYPE_REMOTE);
+    TestTransferListener transferListener = new TestTransferListener();
+    testSource.addTransferListener(transferListener);
+
+    DataSpec dataSpec = new DataSpec(Uri.EMPTY);
+    testSource.open(dataSpec);
+    testSource.read(/* buffer= */ null, /* offset= */ 0, /* readLength= */ 100);
+    testSource.close();
+
+    assertThat(transferListener.lastTransferStartSource).isSameAs(testSource);
+    assertThat(transferListener.lastBytesTransferredSource).isSameAs(testSource);
+    assertThat(transferListener.lastTransferEndSource).isSameAs(testSource);
+    assertThat(transferListener.lastTransferStartDataSpec).isEqualTo(dataSpec);
+    assertThat(transferListener.lastBytesTransferred).isEqualTo(100);
+  }
+
+  private static final class TestSource extends BaseDataSource {
+
+    public TestSource(@DataSource.Type int type) {
+      super(type);
+    }
+
+    @Override
+    public long open(DataSpec dataSpec) throws IOException {
+      transferStarted(dataSpec);
+      return C.LENGTH_UNSET;
+    }
+
+    @Override
+    public int read(byte[] buffer, int offset, int readLength) throws IOException {
+      bytesTransferred(readLength);
+      return readLength;
+    }
+
+    @Override
+    public @Nullable Uri getUri() {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public void close() throws IOException {
+      transferEnded();
+    }
+  }
+
+  private static final class TestTransferListener implements TransferListener<DataSource> {
+
+    public Object lastTransferStartSource;
+    public DataSpec lastTransferStartDataSpec;
+    public Object lastBytesTransferredSource;
+    public int lastBytesTransferred;
+    public Object lastTransferEndSource;
+
+    @Override
+    public void onTransferStart(DataSource source, DataSpec dataSpec) {
+      lastTransferStartSource = source;
+      lastTransferStartDataSpec = dataSpec;
+    }
+
+    @Override
+    public void onBytesTransferred(DataSource source, int bytesTransferred) {
+      lastBytesTransferredSource = source;
+      lastBytesTransferred = bytesTransferred;
+    }
+
+    @Override
+    public void onTransferEnd(DataSource source) {
+      lastTransferEndSource = source;
+    }
+  }
+}