Remove peer connection and signaling calls from UI thread.

- Add separate looper threads for peer connection and websocket
signaling classes.
- To improve the connection speed start peer connection factory
initialization once EGL context is ready in parallel with the room
connection.
- Add asynchronious http request class and start using it in
webscoket signaling and room parameters extractor.
- Add helper looper based executor class.
- Port some of henrika changes from
https://webrtc-codereview.appspot.com/36629004/ to fix sensor
crashes on non L devices - will remove the change if CL will
be submitted soon.

R=jiayl@webrtc.org, wzh@webrtc.org

Review URL: https://webrtc-codereview.appspot.com/41369004

git-svn-id: http://webrtc.googlecode.com/svn/trunk@8006 4adac7df-926f-26a2-2b94-8c16560cd09d

talk/examples/android/src/org/appspot/apprtc/WebSocketRTCClient.java

@@ -26,17 +26,11 @@
  */
 package org.appspot.apprtc;
 
-import android.os.Handler;
-import android.os.Looper;
 import android.util.Log;
 
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.net.HttpURLConnection;
-import java.net.URL;
-import java.util.Scanner;
-
+import org.appspot.apprtc.util.AsyncHttpURLConnection;
+import org.appspot.apprtc.util.AsyncHttpURLConnection.AsyncHttpEvents;
+import org.appspot.apprtc.util.LooperExecutor;
 import org.appspot.apprtc.RoomParametersFetcher.RoomParametersFetcherEvents;
 import org.appspot.apprtc.WebSocketChannelClient.WebSocketChannelEvents;
 import org.appspot.apprtc.WebSocketChannelClient.WebSocketConnectionState;
@@ -56,7 +50,7 @@ import org.webrtc.SessionDescription;
  * be sent after WebSocket connection is established.
  */
 public class WebSocketRTCClient implements AppRTCClient,
-    RoomParametersFetcherEvents, WebSocketChannelEvents {
+    WebSocketChannelEvents {
   private static final String TAG = "WSRTCClient";
 
   private enum ConnectionState {
@@ -65,7 +59,7 @@ public class WebSocketRTCClient implements AppRTCClient,
   private enum MessageType {
     MESSAGE, BYE
   };
-  private final Handler uiHandler;
+  private final LooperExecutor executor;
   private boolean loopback;
   private boolean initiator;
   private SignalingEvents events;
@@ -77,14 +71,81 @@ public class WebSocketRTCClient implements AppRTCClient,
 
   public WebSocketRTCClient(SignalingEvents events) {
     this.events = events;
-    uiHandler = new Handler(Looper.getMainLooper());
+    executor = new LooperExecutor();
   }
 
   // --------------------------------------------------------------------
-  // RoomConnectionEvents interface implementation.
-  // All events are called on UI thread.
+  // AppRTCClient interface implementation.
+  // Asynchronously connect to an AppRTC room URL, e.g.
+  // https://apprtc.appspot.com/register/<room>, retrieve room parameters
+  // and connect to WebSocket server.
   @Override
-  public void onSignalingParametersReady(final SignalingParameters params) {
+  public void connectToRoom(final String url, final boolean loopback) {
+    executor.requestStart();
+    executor.execute(new Runnable() {
+      @Override
+      public void run() {
+        connectToRoomInternal(url, loopback);
+      }
+    });
+  }
+
+  @Override
+  public void disconnectFromRoom() {
+    executor.execute(new Runnable() {
+      @Override
+      public void run() {
+        disconnectFromRoomInternal();
+      }
+    });
+    executor.requestStop();
+  }
+
+  // Connects to room - function runs on a local looper thread.
+  private void connectToRoomInternal(String url, boolean loopback) {
+    Log.d(TAG, "Connect to room: " + url);
+    this.loopback = loopback;
+    roomState = ConnectionState.NEW;
+    // Create WebSocket client.
+    wsClient = new WebSocketChannelClient(executor, this);
+    // Get room parameters.
+    fetcher = new RoomParametersFetcher(loopback, url,
+      new RoomParametersFetcherEvents() {
+        @Override
+        public void onSignalingParametersReady(
+            final SignalingParameters params) {
+          executor.execute(new Runnable() {
+            @Override
+            public void run() {
+              signalingParametersReady(params);
+            }
+          });
+        }
+
+        @Override
+        public void onSignalingParametersError(String description) {
+          reportError(description);
+        }
+      }
+    );
+  }
+
+  // Disconnect from room and send bye messages - runs on a local looper thread.
+  private void disconnectFromRoomInternal() {
+    Log.d(TAG, "Disconnect. Room state: " + roomState);
+    if (roomState == ConnectionState.CONNECTED) {
+      Log.d(TAG, "Closing room.");
+      sendPostMessage(MessageType.BYE, byeMessageUrl, "");
+    }
+    roomState = ConnectionState.CLOSED;
+    if (wsClient != null) {
+      wsClient.disconnect(true);
+    }
+  }
+
+  // Callback issued when room parameters are extracted. Runs on local
+  // looper thread.
+  private void signalingParametersReady(final SignalingParameters params) {
     Log.d(TAG, "Room connection completed.");
     if (loopback && (!params.initiator || params.offerSdp != null)) {
       reportError("Loopback room is busy.");
@@ -100,15 +161,16 @@ public class WebSocketRTCClient implements AppRTCClient,
       + params.roomId + "/" + params.clientId;
     roomState = ConnectionState.CONNECTED;
 
-    // Connect to WebSocket server.
-    wsClient.connect(params.wssUrl, params.wssPostUrl);
-    wsClient.setClientParameters(params.roomId, params.clientId);
-
     // Fire connection and signaling parameters events.
     events.onConnectedToRoom(params);
+
+    // Connect to WebSocket server.
+    wsClient.connect(
+        params.wssUrl, params.wssPostUrl, params.roomId, params.clientId);
+
+    // For call receiver get sdp offer and ice candidates
+    // from room parameters and fire corresponding events.
     if (!params.initiator) {
-      // For call receiver get sdp offer and ice candidates
-      // from room parameters.
       if (params.offerSdp != null) {
         events.onRemoteDescription(params.offerSdp);
       }
@@ -120,14 +182,90 @@ public class WebSocketRTCClient implements AppRTCClient,
     }
   }
 
+  // Send local offer SDP to the other participant.
   @Override
-  public void onSignalingParametersError(final String description) {
-    reportError("Room connection error: " + description);
+  public void sendOfferSdp(final SessionDescription sdp) {
+    executor.execute(new Runnable() {
+      @Override
+      public void run() {
+        if (roomState != ConnectionState.CONNECTED) {
+          reportError("Sending offer SDP in non connected state.");
+          return;
+        }
+        JSONObject json = new JSONObject();
+        jsonPut(json, "sdp", sdp.description);
+        jsonPut(json, "type", "offer");
+        sendPostMessage(MessageType.MESSAGE, postMessageUrl, json.toString());
+        if (loopback) {
+          // In loopback mode rename this offer to answer and route it back.
+          SessionDescription sdpAnswer = new SessionDescription(
+              SessionDescription.Type.fromCanonicalForm("answer"),
+              sdp.description);
+          events.onRemoteDescription(sdpAnswer);
+        }
+      }
+    });
+  }
+
+  // Send local answer SDP to the other participant.
+  @Override
+  public void sendAnswerSdp(final SessionDescription sdp) {
+    executor.execute(new Runnable() {
+      @Override
+      public void run() {
+        if (loopback) {
+          Log.e(TAG, "Sending answer in loopback mode.");
+          return;
+        }
+        if (wsClient.getState() != WebSocketConnectionState.REGISTERED) {
+          reportError("Sending answer SDP in non registered state.");
+          return;
+        }
+        JSONObject json = new JSONObject();
+        jsonPut(json, "sdp", sdp.description);
+        jsonPut(json, "type", "answer");
+        wsClient.send(json.toString());
+      }
+    });
+  }
+
+  // Send Ice candidate to the other participant.
+  @Override
+  public void sendLocalIceCandidate(final IceCandidate candidate) {
+    executor.execute(new Runnable() {
+      @Override
+      public void run() {
+        JSONObject json = new JSONObject();
+        jsonPut(json, "type", "candidate");
+        jsonPut(json, "label", candidate.sdpMLineIndex);
+        jsonPut(json, "id", candidate.sdpMid);
+        jsonPut(json, "candidate", candidate.sdp);
+        if (initiator) {
+          // Call initiator sends ice candidates to GAE server.
+          if (roomState != ConnectionState.CONNECTED) {
+            reportError("Sending ICE candidate in non connected state.");
+            return;
+          }
+          sendPostMessage(MessageType.MESSAGE, postMessageUrl, json.toString());
+          if (loopback) {
+            events.onRemoteIceCandidate(candidate);
+          }
+        } else {
+          // Call receiver sends ice candidates to websocket server.
+          if (wsClient.getState() != WebSocketConnectionState.REGISTERED) {
+            reportError("Sending ICE candidate in non registered state.");
+            return;
+          }
+          wsClient.send(json.toString());
+        }
+      }
+    });
   }
 
   // --------------------------------------------------------------------
   // WebSocketChannelEvents interface implementation.
-  // All events are called on UI thread.
+  // All events are called by WebSocketChannelClient on a local looper thread
+  // (passed to WebSocket client constructor).
   @Override
   public void onWebSocketOpen() {
     Log.d(TAG, "Websocket connection completed. Registering...");
@@ -198,108 +336,12 @@ public class WebSocketRTCClient implements AppRTCClient,
     reportError("WebSocket error: " + description);
   }
 
-  // --------------------------------------------------------------------
-  // AppRTCClient interface implementation.
-  // Asynchronously connect to an AppRTC room URL, e.g.
-  // https://apprtc.appspot.com/register/<room>, retrieve room parameters
-  // and connect to WebSocket server.
-  @Override
-  public void connectToRoom(String url, boolean loopback) {
-    this.loopback = loopback;
-    // Create WebSocket client.
-    wsClient = new WebSocketChannelClient(this);
-    // Get room parameters.
-    roomState = ConnectionState.NEW;
-    fetcher = new RoomParametersFetcher(this, loopback);
-    fetcher.execute(url);
-  }
-
-  @Override
-  public void disconnect() {
-    Log.d(TAG, "Disconnect. Room state: " + roomState);
-    if (roomState == ConnectionState.CONNECTED) {
-      Log.d(TAG, "Closing room.");
-      sendPostMessage(MessageType.BYE, byeMessageUrl, "");
-    }
-    roomState = ConnectionState.CLOSED;
-    if (wsClient != null) {
-      wsClient.disconnect();
-    }
-  }
-
-  // Send local SDP (offer or answer, depending on role) to the
-  // other participant.  Note that it is important to send the output of
-  // create{Offer,Answer} and not merely the current value of
-  // getLocalDescription() because the latter may include ICE candidates that
-  // we might want to filter elsewhere.
-  @Override
-  public void sendOfferSdp(final SessionDescription sdp) {
-    if (roomState != ConnectionState.CONNECTED) {
-      reportError("Sending offer SDP in non connected state.");
-      return;
-    }
-    JSONObject json = new JSONObject();
-    jsonPut(json, "sdp", sdp.description);
-    jsonPut(json, "type", "offer");
-    sendPostMessage(MessageType.MESSAGE, postMessageUrl, json.toString());
-    if (loopback) {
-      // In loopback mode rename this offer to answer and route it back.
-      SessionDescription sdpAnswer = new SessionDescription(
-          SessionDescription.Type.fromCanonicalForm("answer"),
-          sdp.description);
-      events.onRemoteDescription(sdpAnswer);
-    }
-  }
-
-  @Override
-  public void sendAnswerSdp(final SessionDescription sdp) {
-    if (loopback) {
-      Log.e(TAG, "Sending answer in loopback mode.");
-      return;
-    }
-    if (wsClient.getState() != WebSocketConnectionState.REGISTERED) {
-      reportError("Sending answer SDP in non registered state.");
-      return;
-    }
-    JSONObject json = new JSONObject();
-    jsonPut(json, "sdp", sdp.description);
-    jsonPut(json, "type", "answer");
-    wsClient.send(json.toString());
-  }
-
-  // Send Ice candidate to the other participant.
-  @Override
-  public void sendLocalIceCandidate(final IceCandidate candidate) {
-    JSONObject json = new JSONObject();
-    jsonPut(json, "type", "candidate");
-    jsonPut(json, "label", candidate.sdpMLineIndex);
-    jsonPut(json, "id", candidate.sdpMid);
-    jsonPut(json, "candidate", candidate.sdp);
-    if (initiator) {
-      // Call initiator sends ice candidates to GAE server.
-      if (roomState != ConnectionState.CONNECTED) {
-        reportError("Sending ICE candidate in non connected state.");
-        return;
-      }
-      sendPostMessage(MessageType.MESSAGE, postMessageUrl, json.toString());
-      if (loopback) {
-        events.onRemoteIceCandidate(candidate);
-      }
-    } else {
-      // Call receiver sends ice candidates to websocket server.
-      if (wsClient.getState() != WebSocketConnectionState.REGISTERED) {
-        reportError("Sending ICE candidate in non registered state.");
-        return;
-      }
-      wsClient.send(json.toString());
-    }
-  }
-
   // --------------------------------------------------------------------
   // Helper functions.
   private void reportError(final String errorMessage) {
     Log.e(TAG, errorMessage);
-    uiHandler.post(new Runnable() {
+    executor.execute(new Runnable() {
+      @Override
       public void run() {
         if (roomState != ConnectionState.ERROR) {
           roomState = ConnectionState.ERROR;
@@ -318,81 +360,36 @@ public class WebSocketRTCClient implements AppRTCClient,
     }
   }
 
-  private class PostMessage {
-    PostMessage(MessageType type, String postUrl, String message) {
-      this.messageType = type;
-      this.postUrl = postUrl;
-      this.message = message;
-    }
-    public final MessageType messageType;
-    public final String postUrl;
-    public final String message;
-  }
-
-  // Queue a message for sending to the room  and send it if already connected.
-  private synchronized void sendPostMessage(
-      MessageType messageType, String url, String message) {
-    final PostMessage postMessage = new PostMessage(messageType, url, message);
-    Runnable runDrain = new Runnable() {
-      public void run() {
-        sendPostMessageAsync(postMessage);
-      }
-    };
-    new Thread(runDrain).start();
-  }
-
-  // Send all queued POST messages to app engine server.
-  private void sendPostMessageAsync(PostMessage postMessage) {
-    if (postMessage.messageType == MessageType.BYE) {
-      Log.d(TAG, "C->GAE: " + postMessage.postUrl);
+  // Send SDP or ICE candidate to a room server.
+  private void sendPostMessage(
+      final MessageType messageType, final String url, final String message) {
+    if (messageType == MessageType.BYE) {
+      Log.d(TAG, "C->GAE: " + url);
     } else {
-      Log.d(TAG, "C->GAE: " + postMessage.message);
+      Log.d(TAG, "C->GAE: " + message);
     }
-    try {
-      // Get connection.
-      HttpURLConnection connection =
-        (HttpURLConnection) new URL(postMessage.postUrl).openConnection();
-      byte[] postData = postMessage.message.getBytes("UTF-8");
-      connection.setUseCaches(false);
-      connection.setDoOutput(true);
-      connection.setDoInput(true);
-      connection.setRequestMethod("POST");
-      connection.setFixedLengthStreamingMode(postData.length);
-      connection.setRequestProperty(
-          "content-type", "text/plain; charset=utf-8");
-
-      // Send POST request.
-      OutputStream outStream = connection.getOutputStream();
-      outStream.write(postData);
-      outStream.close();
-
-      // Get response.
-      int responseCode = connection.getResponseCode();
-      if (responseCode != 200) {
-        reportError("Non-200 response to POST: "
-          + connection.getHeaderField(null));
-      }
-      InputStream responseStream = connection.getInputStream();
-      String response = drainStream(responseStream);
-      responseStream.close();
-      if (postMessage.messageType == MessageType.MESSAGE) {
-        JSONObject roomJson = new JSONObject(response);
-        String result = roomJson.getString("result");
-        if (!result.equals("SUCCESS")) {
-          reportError("Room POST error: " + result);
+    AsyncHttpURLConnection httpConnection = new AsyncHttpURLConnection(
+      "POST", url, message, new AsyncHttpEvents() {
+        @Override
+        public void OnHttpError(String errorMessage) {
+          reportError("GAE POST error: " + errorMessage);
         }
-      }
-    } catch (IOException e) {
-      reportError("GAE POST error: " + e.getMessage());
-    } catch (JSONException e) {
-      reportError("GAE POST JSON error: " + e.getMessage());
-    }
-  }
 
-  // Return the contents of an InputStream as a String.
-  private String drainStream(InputStream in) {
-    Scanner s = new Scanner(in).useDelimiter("\\A");
-    return s.hasNext() ? s.next() : "";
+        @Override
+        public void OnHttpComplete(String response) {
+          if (messageType == MessageType.MESSAGE) {
+            try {
+              JSONObject roomJson = new JSONObject(response);
+              String result = roomJson.getString("result");
+              if (!result.equals("SUCCESS")) {
+                reportError("GAE POST error: " + result);
+              }
+            } catch (JSONException e) {
+              reportError("GAE POST JSON error: " + e.toString());
+            }
+          }
+        }
+      });
+    httpConnection.send();
   }
-
 }