fix: resolve allowHalfOpen behavior (#230)

Co-authored-by: Apoorv <[email protected]>

Author: akai07

__tests__/allowHalfOpen.test.js

@@ -0,0 +1,109 @@
+import { expect, test, jest } from '@jest/globals';
+import net from '../src/index';
+import { nativeEventEmitter } from '../src/Globals';
+import { NativeModules } from 'react-native';
+
+const Sockets = NativeModules.TcpSockets;
+
+jest.mock('../src/Globals', () => {
+    const { EventEmitter } = require('events');
+    const emitter = new EventEmitter();
+    const originalAddListener = emitter.addListener.bind(emitter);
+    // @ts-ignore
+    emitter.addListener = (event, listener) => {
+        originalAddListener(event, listener);
+        return { remove: () => emitter.removeListener(event, listener) };
+    };
+
+    let idCounter = 1000;
+    return {
+        __esModule: true,
+        nativeEventEmitter: emitter,
+        getNextId: () => idCounter++,
+    };
+});
+
+test('allowHalfOpen: false (default) should call Sockets.end() on end event', () => {
+    return new Promise((resolve, reject) => {
+        // Reset mocks
+        Sockets.end.mockClear();
+
+        const server = net.createServer(); // allowHalfOpen default false
+        // @ts-ignore
+        const serverId = server._id;
+        server.listen(12345);
+
+        server.on('connection', (socket) => {
+            socket.on('end', () => {
+                try {
+                    // When we receive 'end', if allowHalfOpen is false, socket.end() should be called
+                    // which calls Sockets.end(id)
+                    expect(Sockets.end).toHaveBeenCalled();
+                    resolve(undefined);
+                } catch (e) {
+                    reject(e);
+                }
+            });
+        });
+
+        // Simulate connection
+        nativeEventEmitter.emit('connection', {
+            id: serverId,
+            info: {
+                id: 456,
+                connection: {
+                    localAddress: '127.0.0.1',
+                    localPort: 12345,
+                    remoteAddress: '127.0.0.1',
+                    remotePort: 54321,
+                    remoteFamily: 'IPv4',
+                },
+            },
+        });
+
+        // Simulate 'end' event from native for socket 456
+        nativeEventEmitter.emit('end', { id: 456 });
+    });
+});
+
+test('allowHalfOpen: true should NOT call Sockets.end() on end event', () => {
+    return new Promise((resolve, reject) => {
+        // Reset mocks
+        Sockets.end.mockClear();
+
+        const server = net.createServer({ allowHalfOpen: true });
+        // @ts-ignore
+        const serverId = server._id;
+        server.listen(12346);
+
+        server.on('connection', (socket) => {
+            socket.on('end', () => {
+                try {
+                    // When we receive 'end', if allowHalfOpen is true, socket.end() should NOT be called
+                    expect(Sockets.end).not.toHaveBeenCalled();
+                    resolve(undefined);
+                } catch (e) {
+                    reject(e);
+                }
+            });
+        });
+
+        // Simulate connection
+        nativeEventEmitter.emit('connection', {
+            id: serverId,
+            info: {
+                id: 457,
+                connection: {
+                    localAddress: '127.0.0.1',
+                    localPort: 12346,
+                    remoteAddress: '127.0.0.1',
+                    remotePort: 54321,
+                    remoteFamily: 'IPv4',
+                },
+            },
+        });
+
+        // Simulate 'end' event from native for socket 457
+        nativeEventEmitter.emit('end', { id: 457 });
+    });
+});

__tests__/server_options.test.js

@@ -0,0 +1,53 @@
+import { expect, test, jest } from '@jest/globals';
+
+jest.mock('../src/Globals', () => {
+    const { EventEmitter } = require('events');
+    const emitter = new EventEmitter();
+    const originalAddListener = emitter.addListener.bind(emitter);
+    // @ts-ignore
+    emitter.addListener = (event, listener) => {
+        originalAddListener(event, listener);
+        return { remove: () => emitter.removeListener(event, listener) };
+    };
+    return {
+        __esModule: true,
+        nativeEventEmitter: emitter,
+        getNextId: () => 123,
+    };
+});
+
+import net from '../src/index';
+import { nativeEventEmitter } from '../src/Globals';
+
+test('server option pauseOnConnect should pause the socket', () => {
+    return new Promise((resolve, reject) => {
+        const server = net.createServer({ pauseOnConnect: true });
+
+        server.listen(12345);
+
+        server.on('connection', (socket) => {
+            try {
+                // Check if socket is paused
+                // @ts-ignore
+                expect(socket._paused).toBe(true);
+                resolve(undefined);
+            } catch (error) {
+                reject(error);
+            }
+        });
+
+        nativeEventEmitter.emit('connection', {
+            id: 123,
+            info: {
+                id: 456,
+                connection: {
+                    localAddress: '127.0.0.1',
+                    localPort: 12345,
+                    remoteAddress: '127.0.0.1',
+                    remotePort: 54321,
+                    remoteFamily: 'IPv4',
+                },
+            },
+        });
+    });
+});

android/src/main/java/com/asterinet/react/tcpsocket/TcpEventListener.java

@@ -92,6 +92,12 @@ public void onData(int id, byte[] data) {
         sendEvent("data", eventParams);
     }
 
+    public void onEnd(int id) {
+        WritableMap eventParams = Arguments.createMap();
+        eventParams.putInt("id", id);
+        sendEvent("end", eventParams);
+    }
+
     public void onWritten(int id, int msgId, @Nullable Exception e) {
         String error = null;
         if (e != null) {

android/src/main/java/com/asterinet/react/tcpsocket/TcpSocketClient.java

@@ -267,7 +267,8 @@ public void run() {
                     if (bufferCount > 0) {
                         receiverListener.onData(socketId, Arrays.copyOfRange(buffer, 0, bufferCount));
                     } else if (bufferCount == -1) {
-                        clientSocket.destroy();
+                        receiverListener.onEnd(socketId);
+                        break;
                     }
                 }
             } catch (IOException | InterruptedException ioe) {

ios/TcpSocketClient.h

@@ -40,6 +40,7 @@ typedef enum RCTTCPError RCTTCPError;
 - (void)onSecureConnection:(TcpSocketClient *)client
                   toClient:(NSNumber *)clientID;
 - (void)onData:(NSNumber *)clientID data:(NSData *)data;
+- (void)onEnd:(NSNumber *)clientID;
 - (void)onClose:(TcpSocketClient *)client withError:(NSError *)err;
 - (void)onError:(TcpSocketClient *)client withError:(NSError *)err;
 - (void)onWrittenData:(TcpSocketClient *)client msgId:(NSNumber *)msgId;

ios/TcpSocketClient.m

@@ -580,9 +580,9 @@ - (void)socket:(GCDAsyncSocket *)sock
 }
 
 - (void)socketDidCloseReadStream:(GCDAsyncSocket *)sock {
-    // TODO : investigate for half-closed sockets
-    // for now close the stream completely
-    [sock disconnect];
+    if (_clientDelegate) {
+        [_clientDelegate onEnd:[sock userData]];
+    }
 }
 
 - (void)socketDidDisconnect:(GCDAsyncSocket *)sock withError:(NSError *)err {

ios/TcpSockets.m

@@ -22,7 +22,7 @@ @implementation TcpSockets {
 - (NSArray<NSString *> *)supportedEvents {
     return @[
         @"connect", @"listening", @"connection", @"secureConnection", @"data",
-        @"close", @"error", @"written"
+        @"close", @"error", @"written", @"end"
     ];
 }
 
@@ -294,7 +294,17 @@ - (void)onSocketConnection:(TcpSocketClient *)client
 - (void)onData:(NSNumber *)clientID data:(NSData *)data {
     NSString *base64String = [data base64EncodedStringWithOptions:0];
     [self sendEventWithName:@"data"
-                       body:@{@"id" : clientID, @"data" : base64String}];
+                       body:@{
+                           @"id" : clientID,
+                           @"data" : base64String
+                       }];
+}
+
+- (void)onEnd:(NSNumber *)clientID {
+    [self sendEventWithName:@"end"
+                       body:@{
+                           @"id" : clientID
+                       }];
 }
 
 - (void)onClose:(NSNumber *)clientID withError:(NSError *)err {

jest.setup.js

@@ -14,6 +14,10 @@ jest.mock('react-native', () => {
                 destroy: jest.fn(),
                 write: jest.fn(),
                 listen: jest.fn(),
+                pause: jest.fn(),
+                resume: jest.fn(),
+                setKeepAlive: jest.fn(),
+                setNoDelay: jest.fn(),
             },
         },
         NativeEventEmitter: NativeEventEmitter,

src/Server.js

@@ -38,6 +38,7 @@ export default class Server extends EventEmitter {
         this._id = getNextId();
         /** @protected @readonly */
         this._eventEmitter = nativeEventEmitter;
+        // console.log('Server eventEmitter:', this._eventEmitter);
         /** @private @type {Set<Socket>} */
         this._connections = new Set();
         /** @private */
@@ -249,6 +250,14 @@ export default class Server extends EventEmitter {
                 const keepAliveDelay = this._serverOptions.keepAliveInitialDelay || 0;
                 newSocket.setKeepAlive(this._serverOptions.keepAlive, keepAliveDelay);
             }
+
+            if (this._serverOptions.allowHalfOpen !== undefined) {
+                newSocket.allowHalfOpen = this._serverOptions.allowHalfOpen;
+            }
+
+            if (this._serverOptions.pauseOnConnect) {
+                newSocket.pause();
+            }
         }
 
         return newSocket;