Data channel implementation for SessionDependencyFactory.

components/devtools_bridge/android/javatests2/src/org/chromium/components/devtools_bridge/SessionDependencyFactoryNativeTest.java

Index: components/devtools_bridge/android/javatests2/src/org/chromium/components/devtools_bridge/SessionDependencyFactoryNativeTest.java
diff --git a/components/devtools_bridge/android/javatests2/src/org/chromium/components/devtools_bridge/SessionDependencyFactoryNativeTest.java b/components/devtools_bridge/android/javatests2/src/org/chromium/components/devtools_bridge/SessionDependencyFactoryNativeTest.java
index 0be312e009a44ec20dfee295e895d68330e00ae5..324374847ea034b41c514c47723f5f3b7a38da9f 100644
--- a/components/devtools_bridge/android/javatests2/src/org/chromium/components/devtools_bridge/SessionDependencyFactoryNativeTest.java
+++ b/components/devtools_bridge/android/javatests2/src/org/chromium/components/devtools_bridge/SessionDependencyFactoryNativeTest.java
@@ -5,16 +5,21 @@
 package org.chromium.components.devtools_bridge;
 
 import android.test.InstrumentationTestCase;
+import android.test.suitebuilder.annotation.MediumTest;
 import android.test.suitebuilder.annotation.SmallTest;
 
 import junit.framework.Assert;
 
+import java.nio.ByteBuffer;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingDeque;
 
 /**
  * Tests for {@link SessionDependencyFactoryNative}
  */
 public class SessionDependencyFactoryNativeTest extends InstrumentationTestCase {
+    private static final int DATA_CHANNEL_ID = 0;
+
     private SessionDependencyFactoryNative mInstance;
     private AbstractPeerConnection mConnection;
     private ObserverMock mObserver;
@@ -123,6 +128,67 @@ public class SessionDependencyFactoryNativeTest extends InstrumentationTestCase
         mInstance.dispose();
     }
 
+    @SmallTest
+    public void testDataChannel() {
+        mInstance = newFactory();
+        mConnection = newConnection();
+        AbstractDataChannel channel = mConnection.createDataChannel(DATA_CHANNEL_ID);
+
+        channel.registerObserver(new DataChannelObserver());
+        channel.send(ByteBuffer.allocateDirect(1), AbstractDataChannel.MessageType.TEXT);
+        channel.send(ByteBuffer.allocateDirect(1), AbstractDataChannel.MessageType.BINARY);
+        channel.unregisterObserver();
+        channel.close();
+
+        channel.dispose();
+        mConnection.dispose();
+        mInstance.dispose();
+    }
+
+    @SmallTest
+    public void testDataChannelOpens() throws Exception {
+        mInstance = newFactory();
+        Pipe pipe = new Pipe(mInstance);
+
+        pipe.registerDatatChannelObservers();
+
+        pipe.negotiate();
+
+        pipe.dataChannelObserver(0).opened.await();
+        pipe.dataChannelObserver(1).opened.await();
+
+        pipe.unregisterDatatChannelObservers();
+
+        pipe.dispose();
+        mInstance.dispose();
+    }
+
+    @MediumTest
+    public void testPumpData() throws Exception {
+        mInstance = newFactory();
+        Pipe pipe = new Pipe(mInstance);
+        pipe.registerDatatChannelObservers();
+        pipe.negotiate();
+        pipe.dataChannelObserver(0).opened.await();
+
+        // Make sure data channel don't leave local references on stack
+        // of signaling thread. References causes failure like
+        // "Failed adding to JNI local ref table (has 512 entries)".
+        final int count = 1000;
+
+        for (int i = 0; i < count; i++) {
+            pipe.send(0, "A");
+        }
+
+        for (int i = 0; i < count; i++) {
+            pipe.dataChannelObserver(1).received.take();
+        }
+
+        pipe.unregisterDatatChannelObservers();
+        pipe.dispose();
+        mInstance.dispose();
+    }
+
     private SessionDependencyFactoryNative newFactory() {
         return new SessionDependencyFactoryNative();
     }
@@ -139,6 +205,9 @@ public class SessionDependencyFactoryNativeTest extends InstrumentationTestCase
         final ObserverMock mObserver1 = new ObserverMock();
         final ObserverMock mObserver2 = new ObserverMock();
 
+        DataChannelObserver mDataChannelObserver1 = new DataChannelObserver();
+        DataChannelObserver mDataChannelObserver2 = new DataChannelObserver();
+
         final AbstractPeerConnection mConnection1;
         final AbstractPeerConnection mConnection2;
 
@@ -153,8 +222,8 @@ public class SessionDependencyFactoryNativeTest extends InstrumentationTestCase
             mObserver1.iceCandidatesSink = mConnection2;
             mObserver2.iceCandidatesSink = mConnection1;
 
-            mDataChannel1 = mConnection1.createDataChannel(0);
-            mDataChannel2 = mConnection2.createDataChannel(1);
+            mDataChannel1 = mConnection1.createDataChannel(DATA_CHANNEL_ID);
+            mDataChannel2 = mConnection2.createDataChannel(DATA_CHANNEL_ID);
         }
 
         void dispose() {
@@ -188,6 +257,54 @@ public class SessionDependencyFactoryNativeTest extends InstrumentationTestCase
             mObserver1.connected.await();
             mObserver2.connected.await();
         }
+
+        void send(int channelIndex, String data) {
+            send(channelIndex, data.getBytes(), AbstractDataChannel.MessageType.TEXT);
+        }
+
+        void send(int channelIndex, byte[] bytes, AbstractDataChannel.MessageType type) {
+            ByteBuffer rawMessage = ByteBuffer.allocateDirect(bytes.length);
+            rawMessage.put(bytes);
+            rawMessage.limit(rawMessage.position());
+            rawMessage.position(0);
+            dataChannel(channelIndex).send(rawMessage, type);
+        }
+
+        AbstractDataChannel dataChannel(int channelIndex) {
+            switch (channelIndex) {
+                case 0:
+                    return mDataChannel1;
+
+                case 1:
+                    return mDataChannel2;
+
+                default:
+                    throw new ArrayIndexOutOfBoundsException();
+            }
+        }
+
+        DataChannelObserver dataChannelObserver(int channelIndex) {
+            switch (channelIndex) {
+                case 0:
+                    return mDataChannelObserver1;
+
+                case 1:
+                    return mDataChannelObserver2;
+
+                default:
+                    throw new ArrayIndexOutOfBoundsException();
+            }
+        }
+
+        void registerDatatChannelObservers() {
+            mDataChannel1.registerObserver(mDataChannelObserver1);
+            mDataChannel2.registerObserver(mDataChannelObserver2);
+        }
+
+        void unregisterDatatChannelObservers() {
+            mDataChannel1.unregisterObserver();
+            mDataChannel2.unregisterObserver();
+        }
     }
 
     private static class ObserverMock implements AbstractPeerConnection.Observer {
@@ -232,4 +349,28 @@ public class SessionDependencyFactoryNativeTest extends InstrumentationTestCase
             this.connected.countDown();
         }
     }
+
+    private static class DataChannelObserver implements AbstractDataChannel.Observer {
+        public final CountDownLatch opened = new CountDownLatch(1);
+        public final CountDownLatch closed = new CountDownLatch(1);
+        public final LinkedBlockingDeque<byte[]> received = new LinkedBlockingDeque<byte[]>();
+
+        public void onStateChange(AbstractDataChannel.State state) {
+            switch (state) {
+                case OPEN:
+                    opened.countDown();
+                    break;
+
+                case CLOSED:
+                    closed.countDown();
+                    break;
+            }
+        }
+
+        public void onMessage(ByteBuffer message) {
+            byte[] bytes = new byte[message.remaining()];
+            message.get(bytes);
+            received.add(bytes);
+        }
+    }
 }