-
-
Notifications
You must be signed in to change notification settings - Fork 84
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: Implement backpressure handling (#115)
* Add drain event * Add socket.pause() method * Add socket.resume() method
- Loading branch information
Showing
19 changed files
with
545 additions
and
280 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
117 changes: 117 additions & 0 deletions
117
android/src/main/java/com/asterinet/react/tcpsocket/TcpEventListener.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,117 @@ | ||
| package com.asterinet.react.tcpsocket; | ||
|
|
||
| import android.util.Base64; | ||
|
|
||
| import com.facebook.react.bridge.Arguments; | ||
| import com.facebook.react.bridge.ReactContext; | ||
| import com.facebook.react.bridge.WritableMap; | ||
| import com.facebook.react.modules.core.DeviceEventManagerModule; | ||
|
|
||
| import java.net.Inet6Address; | ||
| import java.net.InetAddress; | ||
| import java.net.InetSocketAddress; | ||
| import java.net.ServerSocket; | ||
| import java.net.Socket; | ||
|
|
||
| import javax.annotation.Nullable; | ||
|
|
||
| public class TcpEventListener { | ||
|
|
||
| private final DeviceEventManagerModule.RCTDeviceEventEmitter rctEvtEmitter; | ||
|
|
||
| public TcpEventListener(final ReactContext reactContext) { | ||
| rctEvtEmitter = reactContext.getJSModule(DeviceEventManagerModule.RCTDeviceEventEmitter.class); | ||
| } | ||
|
|
||
| public void onConnection(int serverId, int clientId, Socket socket) { | ||
| WritableMap eventParams = Arguments.createMap(); | ||
| eventParams.putInt("id", serverId); | ||
|
|
||
| WritableMap infoParams = Arguments.createMap(); | ||
| infoParams.putInt("id", clientId); | ||
|
|
||
| WritableMap connectionParams = Arguments.createMap(); | ||
| InetSocketAddress remoteAddress = (InetSocketAddress) socket.getRemoteSocketAddress(); | ||
|
|
||
| connectionParams.putString("localAddress", socket.getLocalAddress().getHostAddress()); | ||
| connectionParams.putInt("localPort", socket.getLocalPort()); | ||
| connectionParams.putString("remoteAddress", remoteAddress.getAddress().getHostAddress()); | ||
| connectionParams.putInt("remotePort", socket.getPort()); | ||
| connectionParams.putString("remoteFamily", remoteAddress.getAddress() instanceof Inet6Address ? "IPv6" : "IPv4"); | ||
|
|
||
| infoParams.putMap("connection", connectionParams); | ||
| eventParams.putMap("info", infoParams); | ||
|
|
||
| sendEvent("connection", eventParams); | ||
| } | ||
|
|
||
| public void onConnect(int id, TcpSocketClient client) { | ||
| WritableMap eventParams = Arguments.createMap(); | ||
| eventParams.putInt("id", id); | ||
| WritableMap connectionParams = Arguments.createMap(); | ||
| Socket socket = client.getSocket(); | ||
| InetSocketAddress remoteAddress = (InetSocketAddress) socket.getRemoteSocketAddress(); | ||
|
|
||
| connectionParams.putString("localAddress", socket.getLocalAddress().getHostAddress()); | ||
| connectionParams.putInt("localPort", socket.getLocalPort()); | ||
| connectionParams.putString("remoteAddress", remoteAddress.getAddress().getHostAddress()); | ||
| connectionParams.putInt("remotePort", socket.getPort()); | ||
| connectionParams.putString("remoteFamily", remoteAddress.getAddress() instanceof Inet6Address ? "IPv6" : "IPv4"); | ||
| eventParams.putMap("connection", connectionParams); | ||
| sendEvent("connect", eventParams); | ||
| } | ||
|
|
||
| public void onListen(int id, TcpSocketServer server) { | ||
| WritableMap eventParams = Arguments.createMap(); | ||
| eventParams.putInt("id", id); | ||
| WritableMap connectionParams = Arguments.createMap(); | ||
| ServerSocket serverSocket = server.getServerSocket(); | ||
| InetAddress address = serverSocket.getInetAddress(); | ||
|
|
||
| connectionParams.putString("localAddress", serverSocket.getInetAddress().getHostAddress()); | ||
| connectionParams.putInt("localPort", serverSocket.getLocalPort()); | ||
| connectionParams.putString("localFamily", address instanceof Inet6Address ? "IPv6" : "IPv4"); | ||
| eventParams.putMap("connection", connectionParams); | ||
| sendEvent("listening", eventParams); | ||
| } | ||
|
|
||
| public void onData(int id, byte[] data) { | ||
| WritableMap eventParams = Arguments.createMap(); | ||
| eventParams.putInt("id", id); | ||
| eventParams.putString("data", Base64.encodeToString(data, Base64.NO_WRAP)); | ||
|
|
||
| sendEvent("data", eventParams); | ||
| } | ||
|
|
||
| public void onWritten(int id, int msgId, @Nullable String error) { | ||
| WritableMap eventParams = Arguments.createMap(); | ||
| eventParams.putInt("id", id); | ||
| eventParams.putInt("msgId", msgId); | ||
| eventParams.putString("err", error); | ||
|
|
||
| sendEvent("written", eventParams); | ||
| } | ||
|
|
||
| public void onClose(int id, String error) { | ||
| if (error != null) { | ||
| onError(id, error); | ||
| } | ||
| WritableMap eventParams = Arguments.createMap(); | ||
| eventParams.putInt("id", id); | ||
| eventParams.putBoolean("hadError", error != null); | ||
|
|
||
| sendEvent("close", eventParams); | ||
| } | ||
|
|
||
| public void onError(int id, String error) { | ||
| WritableMap eventParams = Arguments.createMap(); | ||
| eventParams.putInt("id", id); | ||
| eventParams.putString("error", error); | ||
|
|
||
| sendEvent("error", eventParams); | ||
| } | ||
|
|
||
| private void sendEvent(String eventName, WritableMap params) { | ||
| rctEvtEmitter.emit(eventName, params); | ||
| } | ||
| } |
66 changes: 30 additions & 36 deletions
66
android/src/main/java/com/asterinet/react/tcpsocket/TcpReceiverTask.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,70 +1,64 @@ | ||
| package com.asterinet.react.tcpsocket; | ||
|
|
||
| import android.os.AsyncTask; | ||
| import android.util.Pair; | ||
|
|
||
| import java.io.BufferedInputStream; | ||
| import java.io.IOException; | ||
| import java.net.InetSocketAddress; | ||
| import java.util.Arrays; | ||
| import java.net.Socket; | ||
| import java.util.Arrays; | ||
|
|
||
| /** | ||
| * This is a specialized AsyncTask that receives data from a socket in the background, and | ||
| * This is a specialized Runnable that receives data from a socket in the background, and | ||
| * notifies it's listener when data is received. This is not threadsafe, the listener | ||
| * should handle synchronicity. | ||
| */ | ||
| class TcpReceiverTask extends AsyncTask<Pair<TcpSocketClient, TcpReceiverTask.OnDataReceivedListener>, Void, Void> { | ||
| public class TcpReceiverTask implements Runnable { | ||
|
|
||
| private final TcpSocketClient clientSocket; | ||
| private final TcpEventListener receiverListener; | ||
| private boolean paused = false; | ||
|
|
||
| public TcpReceiverTask(TcpSocketClient clientSocket, TcpEventListener receiverListener) { | ||
| this.clientSocket = clientSocket; | ||
| this.receiverListener = receiverListener; | ||
| } | ||
|
|
||
| /** | ||
| * An infinite loop to block and read data from the socket. | ||
| */ | ||
| @SafeVarargs | ||
| @Override | ||
| protected final Void doInBackground(Pair<TcpSocketClient, TcpReceiverTask.OnDataReceivedListener>... params) { | ||
| if (params.length > 1) { | ||
| throw new IllegalArgumentException("This task is only for a single socket/listener pair."); | ||
| } | ||
|
|
||
| TcpSocketClient clientSocket = params[0].first; | ||
| OnDataReceivedListener receiverListener = params[0].second; | ||
| public void run() { | ||
| int socketId = clientSocket.getId(); | ||
| Socket socket = clientSocket.getSocket(); | ||
| byte[] buffer = new byte[8192]; | ||
| int bufferCount; | ||
| byte[] buffer = new byte[16384]; | ||
| try { | ||
| BufferedInputStream in = new BufferedInputStream(socket.getInputStream()); | ||
| while (!isCancelled() && !socket.isClosed()) { | ||
| bufferCount = in.read(buffer); | ||
| while (!socket.isClosed()) { | ||
| int bufferCount = in.read(buffer); | ||
| waitIfPaused(); | ||
| if (bufferCount > 0) { | ||
| receiverListener.onData(socketId, Arrays.copyOfRange(buffer, 0, bufferCount)); | ||
| } else if (bufferCount == -1) { | ||
| clientSocket.destroy(); | ||
| } | ||
| } | ||
| } catch (IOException ioe) { | ||
| } catch (IOException | InterruptedException ioe) { | ||
| if (receiverListener != null && !socket.isClosed()) { | ||
| receiverListener.onError(socketId, ioe.getMessage()); | ||
| } | ||
| this.cancel(false); | ||
| } | ||
| return null; | ||
| } | ||
|
|
||
| /** | ||
| * Listener interface for receive events. | ||
| */ | ||
| @SuppressWarnings("WeakerAccess") | ||
| public interface OnDataReceivedListener { | ||
| void onConnection(Integer serverId, Integer clientId, Socket socket); | ||
|
|
||
| void onConnect(Integer id, TcpSocketClient client); | ||
|
|
||
| void onListen(Integer id, TcpSocketServer server); | ||
|
|
||
| void onData(Integer id, byte[] data); | ||
| public synchronized void pause() { | ||
| paused = true; | ||
| } | ||
|
|
||
| void onClose(Integer id, String error); | ||
| public synchronized void resume() { | ||
| paused = false; | ||
| notify(); | ||
| } | ||
|
|
||
| void onError(Integer id, String error); | ||
| private synchronized void waitIfPaused() throws InterruptedException { | ||
| while (paused) { | ||
| wait(); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.