Source code
Revision control
Copy as Markdown
Other Tools
/*
* Copyright 2016 The WebRTC Project Authors. All rights reserved.
*
* Use of this source code is governed by a BSD-style license
* that can be found in the LICENSE file in the root of the source
* tree. An additional intellectual property rights grant can be found
* in the file PATENTS. All contributing project authors may
* be found in the AUTHORS file in the root of the source tree.
*/
package org.appspot.apprtc;
import android.util.Log;
import androidx.annotation.Nullable;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.UnknownHostException;
import java.nio.charset.Charset;
import java.util.concurrent.ExecutorService;
import org.webrtc.ThreadUtils;
/**
* Replacement for WebSocketChannelClient for direct communication between two IP addresses. Handles
* the signaling between the two clients using a TCP connection.
* <p>
* All public methods should be called from a looper executor thread
* passed in a constructor, otherwise exception will be thrown.
* All events are dispatched on the same thread.
*/
public class TCPChannelClient {
private static final String TAG = "TCPChannelClient";
private final ExecutorService executor;
private final ThreadUtils.ThreadChecker executorThreadCheck;
private final TCPChannelEvents eventListener;
private TCPSocket socket;
/**
* Callback interface for messages delivered on TCP Connection. All callbacks are invoked from the
* looper executor thread.
*/
public interface TCPChannelEvents {
void onTCPConnected(boolean server);
void onTCPMessage(String message);
void onTCPError(String description);
void onTCPClose();
}
/**
* Initializes the TCPChannelClient. If IP is a local IP address, starts a listening server on
* that IP. If not, instead connects to the IP.
*
* @param eventListener Listener that will receive events from the client.
* @param ip IP address to listen on or connect to.
* @param port Port to listen on or connect to.
*/
public TCPChannelClient(
ExecutorService executor, TCPChannelEvents eventListener, String ip, int port) {
this.executor = executor;
executorThreadCheck = new ThreadUtils.ThreadChecker();
executorThreadCheck.detachThread();
this.eventListener = eventListener;
InetAddress address;
try {
address = InetAddress.getByName(ip);
} catch (UnknownHostException e) {
reportError("Invalid IP address.");
return;
}
if (address.isAnyLocalAddress()) {
socket = new TCPSocketServer(address, port);
} else {
socket = new TCPSocketClient(address, port);
}
socket.start();
}
/**
* Disconnects the client if not already disconnected. This will fire the onTCPClose event.
*/
public void disconnect() {
executorThreadCheck.checkIsOnValidThread();
socket.disconnect();
}
/**
* Sends a message on the socket.
*
* @param message Message to be sent.
*/
public void send(String message) {
executorThreadCheck.checkIsOnValidThread();
socket.send(message);
}
/**
* Helper method for firing onTCPError events. Calls onTCPError on the executor thread.
*/
private void reportError(final String message) {
Log.e(TAG, "TCP Error: " + message);
executor.execute(new Runnable() {
@Override
public void run() {
eventListener.onTCPError(message);
}
});
}
/**
* Base class for server and client sockets. Contains a listening thread that will call
* eventListener.onTCPMessage on new messages.
*/
private abstract class TCPSocket extends Thread {
// Lock for editing out and rawSocket
protected final Object rawSocketLock;
@Nullable
private PrintWriter out;
@Nullable
private Socket rawSocket;
/**
* Connect to the peer, potentially a slow operation.
*
* @return Socket connection, null if connection failed.
*/
@Nullable
public abstract Socket connect();
/** Returns true if sockets is a server rawSocket. */
public abstract boolean isServer();
TCPSocket() {
rawSocketLock = new Object();
}
/**
* The listening thread.
*/
@Override
public void run() {
Log.d(TAG, "Listening thread started...");
// Receive connection to temporary variable first, so we don't block.
Socket tempSocket = connect();
BufferedReader in;
Log.d(TAG, "TCP connection established.");
synchronized (rawSocketLock) {
if (rawSocket != null) {
Log.e(TAG, "Socket already existed and will be replaced.");
}
rawSocket = tempSocket;
// Connecting failed, error has already been reported, just exit.
if (rawSocket == null) {
return;
}
try {
out = new PrintWriter(
new OutputStreamWriter(rawSocket.getOutputStream(), Charset.forName("UTF-8")), true);
in = new BufferedReader(
new InputStreamReader(rawSocket.getInputStream(), Charset.forName("UTF-8")));
} catch (IOException e) {
reportError("Failed to open IO on rawSocket: " + e.getMessage());
return;
}
}
Log.v(TAG, "Execute onTCPConnected");
executor.execute(new Runnable() {
@Override
public void run() {
Log.v(TAG, "Run onTCPConnected");
eventListener.onTCPConnected(isServer());
}
});
while (true) {
final String message;
try {
message = in.readLine();
} catch (IOException e) {
synchronized (rawSocketLock) {
// If socket was closed, this is expected.
if (rawSocket == null) {
break;
}
}
reportError("Failed to read from rawSocket: " + e.getMessage());
break;
}
// No data received, rawSocket probably closed.
if (message == null) {
break;
}
executor.execute(new Runnable() {
@Override
public void run() {
Log.v(TAG, "Receive: " + message);
eventListener.onTCPMessage(message);
}
});
}
Log.d(TAG, "Receiving thread exiting...");
// Close the rawSocket if it is still open.
disconnect();
}
/** Closes the rawSocket if it is still open. Also fires the onTCPClose event. */
public void disconnect() {
try {
synchronized (rawSocketLock) {
if (rawSocket != null) {
rawSocket.close();
rawSocket = null;
out = null;
executor.execute(new Runnable() {
@Override
public void run() {
eventListener.onTCPClose();
}
});
}
}
} catch (IOException e) {
reportError("Failed to close rawSocket: " + e.getMessage());
}
}
/**
* Sends a message on the socket. Should only be called on the executor thread.
*/
public void send(String message) {
Log.v(TAG, "Send: " + message);
synchronized (rawSocketLock) {
if (out == null) {
reportError("Sending data on closed socket.");
return;
}
out.write(message + "\n");
out.flush();
}
}
}
private class TCPSocketServer extends TCPSocket {
// Server socket is also guarded by rawSocketLock.
@Nullable
private ServerSocket serverSocket;
final private InetAddress address;
final private int port;
public TCPSocketServer(InetAddress address, int port) {
this.address = address;
this.port = port;
}
/** Opens a listening socket and waits for a connection. */
@Nullable
@Override
public Socket connect() {
Log.d(TAG, "Listening on [" + address.getHostAddress() + "]:" + Integer.toString(port));
final ServerSocket tempSocket;
try {
tempSocket = new ServerSocket(port, 0, address);
} catch (IOException e) {
reportError("Failed to create server socket: " + e.getMessage());
return null;
}
synchronized (rawSocketLock) {
if (serverSocket != null) {
Log.e(TAG, "Server rawSocket was already listening and new will be opened.");
}
serverSocket = tempSocket;
}
try {
return tempSocket.accept();
} catch (IOException e) {
reportError("Failed to receive connection: " + e.getMessage());
return null;
}
}
/** Closes the listening socket and calls super. */
@Override
public void disconnect() {
try {
synchronized (rawSocketLock) {
if (serverSocket != null) {
serverSocket.close();
serverSocket = null;
}
}
} catch (IOException e) {
reportError("Failed to close server socket: " + e.getMessage());
}
super.disconnect();
}
@Override
public boolean isServer() {
return true;
}
}
private class TCPSocketClient extends TCPSocket {
final private InetAddress address;
final private int port;
public TCPSocketClient(InetAddress address, int port) {
this.address = address;
this.port = port;
}
/** Connects to the peer. */
@Nullable
@Override
public Socket connect() {
Log.d(TAG, "Connecting to [" + address.getHostAddress() + "]:" + Integer.toString(port));
try {
return new Socket(address, port);
} catch (IOException e) {
reportError("Failed to connect: " + e.getMessage());
return null;
}
}
@Override
public boolean isServer() {
return false;
}
}
}