diff --git a/README.md b/README.md index ed721f4..2f84b18 100644 --- a/README.md +++ b/README.md @@ -333,6 +333,8 @@ Here are listed all methods implemented in `react-native-tcp-socket` that imitat * [`setNoDelay([noDelay])`](https://nodejs.org/api/net.html#net_socket_setnodelay_nodelay) * [`setTimeout(timeout[, callback])`](https://nodejs.org/api/net.html#net_socket_settimeout_timeout_callback) * [`write(data[, encoding][, callback])`](https://nodejs.org/api/net.html#net_socket_write_data_encoding_callback) + * **[`sendFile(path, offset, length)`](#socketsendfile----omit-in-toc)** + * **[`receiveHttpBodyToFile(path)`](#socketreceivehttpbodytofile----omit-in-toc)** * [`pause()`](https://nodejs.org/api/net.html#net_socket_pause) * `ref()` - _Will not have any effect_ * [`resume()`](https://nodejs.org/api/net.html#net_socket_resume) @@ -362,6 +364,8 @@ Here are listed all methods implemented in `react-native-tcp-socket` that imitat * [`'drain'`](https://nodejs.org/api/net.html#net_event_drain) * [`'error'`](https://nodejs.org/api/net.html#net_event_error_1) * [`'timeout'`](https://nodejs.org/api/net.html#net_event_timeout) + * **[`'fileProgress'`](#socketsendfile----omit-in-toc)** + * **[`'fileEnd'`](#socketreceivehttpbodytofile----omit-in-toc)** ##### `net.createConnection()` `net.createConnection(options[, callback])` creates a TCP connection using the given `options`. The `options` parameter must be an `object` with the following properties: @@ -378,6 +382,54 @@ Here are listed all methods implemented in `react-native-tcp-socket` that imitat **Note**: The platforms marked as ❌ use the default value. +##### `Socket.sendFile()` +`socket.sendFile(path, offset, length)` sends `length` bytes of the file at `path`, starting at `offset`. The file is read and written to the socket natively: its bytes never go through JS, so a large file is sent much faster than with `write()` and without loading it in memory. It works on iOS/macOS and Android, on client sockets and on the sockets of a server. + +| Parameter | Type | Description | +| ------------ | ---------- | -------------------------------------------------------------------- | +| **`path`** | `` | Absolute path of the file, with or without the `file://` scheme. | +| **`offset`** | `` | Position of the first byte to send. | +| **`length`** | `` | Number of bytes to send. | + +It returns a `Promise` resolved once the bytes are written out, and rejected if the file cannot be read or if the socket fails or closes before the end. The file is sent after the pending `write()` calls. Wait for the promise before writing to the socket or sending another file. + +While the file is sent, `'fileProgress'` is emitted about every 250 ms with `(bytes, total)`: the number of bytes sent and `length`. + +```js +server.on('connection', async (socket) => { + socket.write(`HTTP/1.1 200 OK\r\nContent-Length: ${size}\r\nConnection: close\r\n\r\n`); + await socket.sendFile(path, 0, size); + socket.end(); +}); +``` + +##### `Socket.receiveHttpBodyToFile()` +`socket.receiveHttpBodyToFile(path)` writes the body of an HTTP response to the file at `path`, natively. From this call on, the incoming bytes are parsed as one HTTP response instead of being emitted as `'data'` events: call it before sending the request. It works on iOS/macOS and Android. + +| Response status | Behaviour | +| --------------- | ------------------------------------------------------------------------------------------ | +| `200` | The file is created, or replaced if it exists. | +| `206` | The body is appended to the file, to resume a download with a `Range` request. | +| Any other | The transfer fails and the file is left untouched. | + +The body ends after `Content-Length` bytes, or with the connection when the response has no `Content-Length`. It is written as received: `Transfer-Encoding: chunked` and compressed bodies are not decoded. Only one response is read per socket, the bytes following the body are ignored. + +* `'fileProgress'` is emitted about every 250 ms with `(bytes, total)`: the number of body bytes written by this call and the `Content-Length`, or `-1` when the response has none. +* `'fileEnd'` is emitted once with `(bytes, error)`. `error` is `null` when the whole body is written, or a message such as `'HTTP 404'` or `'Incomplete: 1000/2000 bytes'`. + +```js +const socket = net.createConnection({ host, port }, () => { + socket.receiveHttpBodyToFile(path); + socket.write(`GET /model.bin HTTP/1.1\r\nHost: ${host}\r\nConnection: close\r\n\r\n`); +}); + +socket.on('fileProgress', (bytes, total) => console.log(`${bytes}/${total}`)); +socket.on('fileEnd', (bytes, error) => { + socket.destroy(); + if (error) console.warn(`Download failed after ${bytes} bytes: ${error}`); +}); +``` + #### Server * **Methods:** * [`address()`](https://nodejs.org/api/net.html#net_server_address) diff --git a/__tests__/fileTransfer.test.js b/__tests__/fileTransfer.test.js new file mode 100644 index 0000000..a18dfcd --- /dev/null +++ b/__tests__/fileTransfer.test.js @@ -0,0 +1,154 @@ +import { expect, test, jest, beforeEach } 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 = 2000; + return { + __esModule: true, + nativeEventEmitter: emitter, + getNextId: () => idCounter++, + }; +}); + +const FILE_PATH = '/data/model.bin'; + +function createConnectedSocket() { + const socket = new net.Socket(); + socket.connect({ port: 8080, host: '192.168.1.20' }); + // @ts-ignore + const id = socket._id; + nativeEventEmitter.emit('connect', { + id, + connection: { + localAddress: '192.168.1.10', + localPort: 54321, + remoteAddress: '192.168.1.20', + remotePort: 8080, + remoteFamily: 'IPv4', + }, + }); + return { socket, id }; +} + +beforeEach(() => { + Sockets.sendFile.mockClear(); + Sockets.receiveHttpBodyToFile.mockClear(); +}); + +test('sendFile: rejects when the socket is not connected', async () => { + const socket = new net.Socket(); + + await expect(socket.sendFile(FILE_PATH, 0, 10)).rejects.toThrow('Socket is closed.'); + expect(Sockets.sendFile).not.toHaveBeenCalled(); +}); + +test('sendFile: resolves once the native side reports the file as written', async () => { + const { socket, id } = createConnectedSocket(); + + const sent = socket.sendFile(FILE_PATH, 100, 900); + + expect(Sockets.sendFile).toHaveBeenCalledWith(id, FILE_PATH, 100, 900, 0); + expect(socket.bytesWritten).toBe(900); + nativeEventEmitter.emit('written', { id, msgId: 0 }); + await expect(sent).resolves.toBeUndefined(); +}); + +test('sendFile: stays ordered with write()', async () => { + const { socket, id } = createConnectedSocket(); + + socket.write('HTTP/1.1 200 OK\r\n\r\n'); + const sent = socket.sendFile(FILE_PATH, 0, 10); + + expect(Sockets.write).toHaveBeenLastCalledWith(id, expect.any(String), 0); + expect(Sockets.sendFile).toHaveBeenCalledWith(id, FILE_PATH, 0, 10, 1); + nativeEventEmitter.emit('written', { id, msgId: 0 }); + nativeEventEmitter.emit('written', { id, msgId: 1 }); + await expect(sent).resolves.toBeUndefined(); +}); + +test('sendFile: rejects with the error of the written event', async () => { + const { socket, id } = createConnectedSocket(); + + const sent = socket.sendFile(FILE_PATH, 0, 10); + nativeEventEmitter.emit('written', { id, msgId: 0, err: 'Broken pipe' }); + + await expect(sent).rejects.toThrow('Broken pipe'); +}); + +test('sendFile: rejects when the socket reports an error', async () => { + const { socket, id } = createConnectedSocket(); + + const sent = socket.sendFile(FILE_PATH, 0, 10); + nativeEventEmitter.emit('error', { id, error: 'Cannot open the file to send' }); + + await expect(sent).rejects.toThrow('Cannot open the file to send'); +}); + +test('sendFile: rejects when the socket closes before the end', async () => { + const { socket, id } = createConnectedSocket(); + + const sent = socket.sendFile(FILE_PATH, 0, 10); + nativeEventEmitter.emit('close', { id, error: false }); + + await expect(sent).rejects.toThrow('Socket closed before the file was sent.'); +}); + +test('sendFile: emits fileProgress for its own socket only', () => { + const { socket, id } = createConnectedSocket(); + const onProgress = jest.fn(); + socket.on('fileProgress', onProgress); + + socket.sendFile(FILE_PATH, 0, 1000).catch(() => {}); + nativeEventEmitter.emit('fileProgress', { id: id + 1, bytes: 1, total: 2 }); + nativeEventEmitter.emit('fileProgress', { id, bytes: 250, total: 1000 }); + + expect(onProgress).toHaveBeenCalledTimes(1); + expect(onProgress).toHaveBeenCalledWith(250, 1000); +}); + +test('receiveHttpBodyToFile: throws when the socket is not connected', () => { + const socket = new net.Socket(); + + expect(() => socket.receiveHttpBodyToFile(FILE_PATH)).toThrow('Socket is closed.'); + expect(Sockets.receiveHttpBodyToFile).not.toHaveBeenCalled(); +}); + +test('receiveHttpBodyToFile: emits fileProgress then fileEnd without error', () => { + const { socket, id } = createConnectedSocket(); + const onProgress = jest.fn(); + const onEnd = jest.fn(); + socket.on('fileProgress', onProgress); + socket.on('fileEnd', onEnd); + + expect(socket.receiveHttpBodyToFile(FILE_PATH)).toBe(socket); + nativeEventEmitter.emit('fileProgress', { id, bytes: 500, total: 1000 }); + nativeEventEmitter.emit('fileEnd', { id, bytes: 1000, error: null }); + + expect(Sockets.receiveHttpBodyToFile).toHaveBeenCalledWith(id, FILE_PATH); + expect(onProgress).toHaveBeenCalledWith(500, 1000); + expect(onEnd).toHaveBeenCalledWith(1000, null); +}); + +test('receiveHttpBodyToFile: reports the failure in fileEnd', () => { + const { socket, id } = createConnectedSocket(); + const onEnd = jest.fn(); + socket.on('fileEnd', onEnd); + + socket.receiveHttpBodyToFile(FILE_PATH); + nativeEventEmitter.emit('fileEnd', { id, bytes: 0, error: 'HTTP 404' }); + + expect(onEnd).toHaveBeenCalledWith(0, 'HTTP 404'); +}); diff --git a/android/src/main/java/com/asterinet/react/tcpsocket/HttpBodyFileReceiver.java b/android/src/main/java/com/asterinet/react/tcpsocket/HttpBodyFileReceiver.java new file mode 100644 index 0000000..6b37f35 --- /dev/null +++ b/android/src/main/java/com/asterinet/react/tcpsocket/HttpBodyFileReceiver.java @@ -0,0 +1,144 @@ +package com.asterinet.react.tcpsocket; + +import java.io.BufferedOutputStream; +import java.io.ByteArrayOutputStream; +import java.io.FileOutputStream; +import java.io.IOException; +import java.io.OutputStream; +import java.nio.charset.StandardCharsets; + +import javax.annotation.Nullable; + +/** + * Parses an HTTP response header, then streams the body straight to a file, without passing the + * bytes through JS. A 200 response truncates the file, a 206 response appends to it (resume). + * Reports progress and completion through the TcpEventListener. + */ +class HttpBodyFileReceiver { + static final long PROGRESS_INTERVAL_MS = 250; + private static final int MAX_HEADER_BYTES = 64 * 1024; + private static final int FILE_BUFFER_BYTES = 256 * 1024; + + private final int socketId; + private final String path; + private final TcpEventListener listener; + private final ByteArrayOutputStream header = new ByteArrayOutputStream(); + @Nullable + private OutputStream out; + private long expected = -1; + private long received = 0; + private long lastProgressAt = 0; + private boolean done = false; + private boolean isPartialContent = false; + + HttpBodyFileReceiver(int socketId, String path, TcpEventListener listener) { + this.socketId = socketId; + this.path = path.startsWith("file://") ? path.substring("file://".length()) : path; + this.listener = listener; + } + + void onData(byte[] buffer, int count) { + if (done) return; + try { + if (out != null) { + writeBody(buffer, 0, count); + return; + } + header.write(buffer, 0, count); + byte[] head = header.toByteArray(); + int headerEnd = indexOfHeaderEnd(head); + if (headerEnd < 0) { + if (head.length > MAX_HEADER_BYTES) finish("HTTP header too large"); + return; + } + if (!parseHeader(new String(head, 0, headerEnd, StandardCharsets.ISO_8859_1))) return; + // 206 Partial Content answers a Range request: resume by appending to the partial file + out = new BufferedOutputStream(new FileOutputStream(path, isPartialContent), FILE_BUFFER_BYTES); + int bodyStart = headerEnd + 4; + writeBody(head, bodyStart, head.length - bodyStart); + } catch (IOException e) { + finish(e.getMessage()); + } + } + + void onEnd() { + if (done) return; + if (out == null) { + finish("Connection closed before the HTTP header"); + } else if (expected >= 0 && received < expected) { + finish("Incomplete: " + received + "/" + expected + " bytes"); + } else { + finish(null); + } + } + + void onError(@Nullable String message) { + if (!done) finish(message == null ? "Socket error" : message); + } + + private void writeBody(byte[] buffer, int offset, int length) throws IOException { + int toWrite = expected >= 0 ? (int) Math.min(length, expected - received) : length; + if (toWrite > 0 && out != null) { + out.write(buffer, offset, toWrite); + received += toWrite; + } + long now = System.currentTimeMillis(); + if (now - lastProgressAt >= PROGRESS_INTERVAL_MS) { + lastProgressAt = now; + listener.onFileProgress(socketId, received, expected); + } + if (expected >= 0 && received >= expected) finish(null); + } + + private boolean parseHeader(String head) { + String[] lines = head.split("\r\n"); + String[] statusParts = lines[0].split(" "); + int status = statusParts.length > 1 ? parseIntOr(statusParts[1], -1) : -1; + if (status != 200 && status != 206) { + finish("HTTP " + status); + return false; + } + isPartialContent = status == 206; + for (String line : lines) { + if (line.toLowerCase().startsWith("content-length:")) { + expected = parseLongOr(line.substring("content-length:".length()).trim(), -1); + } + } + return true; + } + + private void finish(@Nullable String error) { + done = true; + if (out != null) { + try { + out.close(); + } catch (IOException e) { + if (error == null) error = e.getMessage(); + } + } + listener.onFileEnd(socketId, received, error); + } + + private static int indexOfHeaderEnd(byte[] data) { + for (int i = 0; i + 3 < data.length; i++) { + if (data[i] == '\r' && data[i + 1] == '\n' && data[i + 2] == '\r' && data[i + 3] == '\n') return i; + } + return -1; + } + + private static int parseIntOr(String value, int fallback) { + try { + return Integer.parseInt(value); + } catch (NumberFormatException e) { + return fallback; + } + } + + private static long parseLongOr(String value, long fallback) { + try { + return Long.parseLong(value); + } catch (NumberFormatException e) { + return fallback; + } + } +} diff --git a/android/src/main/java/com/asterinet/react/tcpsocket/TcpEventListener.java b/android/src/main/java/com/asterinet/react/tcpsocket/TcpEventListener.java index 7c1c6ee..3b18604 100644 --- a/android/src/main/java/com/asterinet/react/tcpsocket/TcpEventListener.java +++ b/android/src/main/java/com/asterinet/react/tcpsocket/TcpEventListener.java @@ -112,6 +112,22 @@ public void onWritten(int id, int msgId, @Nullable Exception e) { sendEvent("written", eventParams); } + public void onFileProgress(int id, long bytes, long total) { + WritableMap eventParams = Arguments.createMap(); + eventParams.putInt("id", id); + eventParams.putDouble("bytes", bytes); + eventParams.putDouble("total", total); + sendEvent("fileProgress", eventParams); + } + + public void onFileEnd(int id, long bytes, @Nullable String error) { + WritableMap eventParams = Arguments.createMap(); + eventParams.putInt("id", id); + eventParams.putDouble("bytes", bytes); + eventParams.putString("error", error); + sendEvent("fileEnd", eventParams); + } + public void onClose(int id, Exception e) { if (e != null) { onError(id, e); diff --git a/android/src/main/java/com/asterinet/react/tcpsocket/TcpSocketClient.java b/android/src/main/java/com/asterinet/react/tcpsocket/TcpSocketClient.java index 7511570..f717af2 100644 --- a/android/src/main/java/com/asterinet/react/tcpsocket/TcpSocketClient.java +++ b/android/src/main/java/com/asterinet/react/tcpsocket/TcpSocketClient.java @@ -7,7 +7,9 @@ import com.facebook.react.bridge.ReadableArray; import java.io.BufferedInputStream; +import java.io.FileInputStream; import java.io.IOException; +import java.io.OutputStream; import java.net.InetAddress; import java.net.InetSocketAddress; import java.net.Socket; @@ -20,12 +22,15 @@ import javax.net.ssl.SSLSocketFactory; class TcpSocketClient extends TcpSocket { + private static final int DATA_BUFFER_BYTES = 16384; + private static final int FILE_BUFFER_BYTES = 65536; private final ExecutorService listenExecutor; private final ExecutorService writeExecutor; private final TcpEventListener receiverListener; private TcpReceiverTask receiverTask; private Socket socket; private boolean closed = true; + private volatile HttpBodyFileReceiver fileReceiver; TcpSocketClient(TcpEventListener receiverListener, Integer id, Socket socket) { super(id); @@ -171,6 +176,57 @@ public void run() { }); } + /** + * Streams `length` bytes of the file at `path`, from `offset`, without passing them through JS. + * Runs on the write executor, so it stays ordered with `write` calls. + */ + public void sendFile(final int msgId, final String rawPath, final long offset, final long length) { + writeExecutor.execute(new Runnable() { + @Override + public void run() { + final Socket s = socket; + if (s == null) { + receiverListener.onWritten(getId(), msgId, new IOException("Attempted to write to closed socket")); + return; + } + final String path = rawPath.startsWith("file://") ? rawPath.substring("file://".length()) : rawPath; + try (FileInputStream in = new FileInputStream(path)) { + in.getChannel().position(offset); + OutputStream out = s.getOutputStream(); + byte[] buffer = new byte[FILE_BUFFER_BYTES]; + long sent = 0; + long lastProgressAt = 0; + while (sent < length) { + int read = in.read(buffer, 0, (int) Math.min(buffer.length, length - sent)); + if (read < 0) throw new IOException("Unexpected end of file after " + sent + " bytes"); + out.write(buffer, 0, read); + sent += read; + long now = System.currentTimeMillis(); + if (now - lastProgressAt >= HttpBodyFileReceiver.PROGRESS_INTERVAL_MS) { + lastProgressAt = now; + receiverListener.onFileProgress(getId(), sent, length); + } + } + out.flush(); + receiverListener.onFileProgress(getId(), sent, length); + receiverListener.onWritten(getId(), msgId, null); + } catch (IOException e) { + receiverListener.onWritten(getId(), msgId, e); + receiverListener.onError(getId(), e); + } + } + }); + } + + /** + * From now on, incoming bytes are parsed as an HTTP response whose body is written to `path` + * (`fileProgress` / `fileEnd` events) instead of being emitted as `data` events. + * Call it before sending the request. + */ + public void receiveHttpBodyToFile(final String path) { + fileReceiver = new HttpBodyFileReceiver(getId(), path, receiverListener); + } + public ReadableMap getPeerCertificate() { return SSLCertificateHelper.getCertificateInfo(socket, true); } @@ -259,20 +315,28 @@ public void run() { return; } - byte[] buffer = new byte[16384]; + byte[] buffer = new byte[FILE_BUFFER_BYTES]; try { BufferedInputStream in = new BufferedInputStream(socket.getInputStream()); while (!socket.isClosed()) { - int bufferCount = in.read(buffer); + // `data` events keep their size, a file is read with larger chunks + int maxCount = clientSocket.fileReceiver != null ? FILE_BUFFER_BYTES : DATA_BUFFER_BYTES; + int bufferCount = in.read(buffer, 0, maxCount); waitIfPaused(); - if (bufferCount > 0) { + HttpBodyFileReceiver fileReceiver = clientSocket.fileReceiver; + if (bufferCount > 0 && fileReceiver != null) { + fileReceiver.onData(buffer, bufferCount); + } else if (bufferCount > 0) { receiverListener.onData(socketId, Arrays.copyOfRange(buffer, 0, bufferCount)); } else if (bufferCount == -1) { + if (fileReceiver != null) fileReceiver.onEnd(); receiverListener.onEnd(socketId); break; } } } catch (IOException | InterruptedException ioe) { + HttpBodyFileReceiver fileReceiver = clientSocket.fileReceiver; + if (fileReceiver != null) fileReceiver.onError(ioe.getMessage()); if (receiverListener != null && socket != null && !socket.isClosed() && !clientSocket.closed) { receiverListener.onError(socketId, ioe); } diff --git a/android/src/main/java/com/asterinet/react/tcpsocket/TcpSocketModule.java b/android/src/main/java/com/asterinet/react/tcpsocket/TcpSocketModule.java index 121edb4..3f112c7 100644 --- a/android/src/main/java/com/asterinet/react/tcpsocket/TcpSocketModule.java +++ b/android/src/main/java/com/asterinet/react/tcpsocket/TcpSocketModule.java @@ -133,6 +133,24 @@ public void write(final int cId, @NonNull final String base64String, final int m socketClient.write(msgId, data); } + @SuppressWarnings("unused") + @ReactMethod + public void sendFile(final int cId, @NonNull final String path, final double offset, final double length, final int msgId) { + TcpSocketClient socketClient = findTcpClient(cId); + if (socketClient == null) { + tcpEvtListener.onWritten(cId, msgId, new IOException("Socket is closed")); + return; + } + socketClient.sendFile(msgId, path, (long) offset, (long) length); + } + + @SuppressWarnings("unused") + @ReactMethod + public void receiveHttpBodyToFile(final int cId, @NonNull final String path) { + TcpSocketClient socketClient = findTcpClient(cId); + if (socketClient != null) socketClient.receiveHttpBodyToFile(path); + } + @SuppressLint("StaticFieldLeak") @SuppressWarnings("unused") @ReactMethod @@ -450,6 +468,12 @@ private void selectNetwork(@Nullable final String iface, @Nullable final String mNetworkMap.put(iface + ipAddress, currentNetwork.getNetwork()); } + @Nullable + private TcpSocketClient findTcpClient(final int id) { + final TcpSocket socket = socketMap.get(id); + return socket instanceof TcpSocketClient ? (TcpSocketClient) socket : null; + } + private TcpSocketClient getTcpClient(final int id) { TcpSocket socket = socketMap.get(id); if (socket == null) { diff --git a/ios/TcpSocketClient.h b/ios/TcpSocketClient.h index ab17fc9..f2d40b3 100644 --- a/ios/TcpSocketClient.h +++ b/ios/TcpSocketClient.h @@ -44,6 +44,12 @@ typedef enum RCTTCPError RCTTCPError; - (void)onClose:(TcpSocketClient *)client withError:(NSError *)err; - (void)onError:(TcpSocketClient *)client withError:(NSError *)err; - (void)onWrittenData:(TcpSocketClient *)client msgId:(NSNumber *)msgId; +- (void)onFileProgress:(NSNumber *)clientID + bytes:(unsigned long long)bytes + total:(long long)total; +- (void)onFileEnd:(NSNumber *)clientID + bytes:(unsigned long long)bytes + error:(NSString *)error; - (NSNumber *)getNextId; @end @@ -108,6 +114,21 @@ typedef enum RCTTCPError RCTTCPError; */ - (void)writeData:(NSData *)data msgId:(NSNumber *)msgId; +/** + * Streams `length` bytes of the file at `path`, from `offset`, without passing + * them through JS. Reports `written` for `msgId` once everything is queued out. + */ +- (void)sendFile:(NSString *)path + offset:(unsigned long long)offset + length:(unsigned long long)length + msgId:(NSNumber *)msgId; + +/** + * From now on, incoming bytes are parsed as an HTTP response whose body is + * written to `path` instead of being emitted as `data` events. + */ +- (void)receiveHttpBodyToFile:(NSString *)path; + - (void)startTLS:(NSDictionary *)tlsOptions; /** @@ -130,6 +151,12 @@ typedef enum RCTTCPError RCTTCPError; + (BOOL)hasIdentity:(NSDictionary *)aliases; +/** + * The serial queue shared by the socket delegate callbacks and the module + * methods + */ ++ (dispatch_queue_t)sharedQueue; + /** * Get peer certificate information * @return NSDictionary with certificate information or nil if not available diff --git a/ios/TcpSocketClient.m b/ios/TcpSocketClient.m index 7490ec4..c46b744 100644 --- a/ios/TcpSocketClient.m +++ b/ios/TcpSocketClient.m @@ -67,6 +67,22 @@ @interface TcpSocketClient () { SecTrustRef _peerTrust; SecIdentityRef _clientIdentity; NSDictionary *_tlsOptionsPending; + // Native file sending + NSFileHandle *_sendFileHandle; + NSNumber *_sendFileMsgId; + unsigned long long _sendFileLength; + unsigned long long _sendFileSent; + NSTimeInterval _sendFileLastProgress; + // Native HTTP body receiving + BOOL _receivingFile; + BOOL _receiveDone; + BOOL _receivePartialContent; + NSString *_receivePath; + NSMutableData *_receiveHeader; + NSFileHandle *_receiveFileHandle; + long long _receiveExpected; + unsigned long long _receiveReceived; + NSTimeInterval _receiveLastProgress; } - (id)initWithClientId:(NSNumber *)clientID @@ -416,7 +432,286 @@ - (void)dropPendingSend:(NSNumber *)key { } } +static const long kSendFileTag = -4242; +static const NSUInteger kSendFileChunk = 256 * 1024; +static const NSTimeInterval kFileProgressInterval = 0.25; +static const NSUInteger kMaxHttpHeaderBytes = 64 * 1024; + +static NSString *TcpStripFileScheme(NSString *path) { + return [path hasPrefix:@"file://"] ? [path substringFromIndex:7] : path; +} + +static NSError *TcpFileError(NSException *exception) { + return [NSError errorWithDomain:RCTTCPErrorDomain + code:RCTTCPOtherError + userInfo:@{ + NSLocalizedDescriptionKey : exception.reason + ?: exception.name + }]; +} + +// The NSFileHandle methods reporting an NSError need iOS 13 / macOS 10.15 / +// tvOS 13. The older ones raise an exception instead. +static BOOL TcpSeekFile(NSFileHandle *handle, unsigned long long offset, + NSError **error) { + if (@available(iOS 13.0, macOS 10.15, tvOS 13.0, *)) { + return [handle seekToOffset:offset error:error]; + } + @try { + [handle seekToFileOffset:offset]; + return YES; + } @catch (NSException *exception) { + *error = TcpFileError(exception); + return NO; + } +} + +static NSData *TcpReadFile(NSFileHandle *handle, NSUInteger length, + NSError **error) { + if (@available(iOS 13.0, macOS 10.15, tvOS 13.0, *)) { + return [handle readDataUpToLength:length error:error]; + } + @try { + return [handle readDataOfLength:length]; + } @catch (NSException *exception) { + *error = TcpFileError(exception); + return nil; + } +} + +static BOOL TcpWriteFile(NSFileHandle *handle, NSData *data, NSError **error) { + if (@available(iOS 13.0, macOS 10.15, tvOS 13.0, *)) { + return [handle writeData:data error:error]; + } + @try { + [handle writeData:data]; + return YES; + } @catch (NSException *exception) { + *error = TcpFileError(exception); + return NO; + } +} + +#pragma mark - Native file sending + +- (void)sendFile:(NSString *)path + offset:(unsigned long long)offset + length:(unsigned long long)length + msgId:(NSNumber *)msgId { + dispatch_async([self methodQueue], ^{ + NSError *error = nil; + NSFileHandle *handle = + [NSFileHandle fileHandleForReadingAtPath:TcpStripFileScheme(path)]; + if (!handle || !TcpSeekFile(handle, offset, &error)) { + [self failSendFile:error ? error.localizedDescription + : @"Cannot open the file to send"]; + return; + } + self->_sendFileHandle = handle; + self->_sendFileMsgId = msgId; + self->_sendFileLength = length; + self->_sendFileSent = 0; + self->_sendFileLastProgress = 0; + [self sendNextFileChunk]; + }); +} + +- (void)sendNextFileChunk { + if (!_sendFileHandle) { + return; + } + if (_sendFileSent >= _sendFileLength) { + [_sendFileHandle closeFile]; + _sendFileHandle = nil; + [_clientDelegate onFileProgress:_id + bytes:_sendFileSent + total:(long long)_sendFileLength]; + [_clientDelegate onWrittenData:self msgId:_sendFileMsgId]; + return; + } + NSError *error = nil; + NSUInteger toRead = (NSUInteger)MIN((unsigned long long)kSendFileChunk, + _sendFileLength - _sendFileSent); + NSData *chunk = TcpReadFile(_sendFileHandle, toRead, &error); + if (!chunk || chunk.length == 0) { + [self failSendFile:error ? error.localizedDescription + : @"Unexpected end of file"]; + return; + } + _sendFileSent += chunk.length; + [_tcpSocket writeData:chunk withTimeout:-1 tag:kSendFileTag]; +} + +- (void)failSendFile:(NSString *)message { + [_sendFileHandle closeFile]; + _sendFileHandle = nil; + [_clientDelegate + onError:self + withError:[NSError + errorWithDomain:RCTTCPErrorDomain + code:RCTTCPSendFailedError + userInfo:@{NSLocalizedDescriptionKey : message}]]; +} + +#pragma mark - Native HTTP body receiving + +- (void)receiveHttpBodyToFile:(NSString *)path { + dispatch_async([self methodQueue], ^{ + self->_receivePath = TcpStripFileScheme(path); + self->_receiveHeader = [NSMutableData data]; + self->_receiveFileHandle = nil; + self->_receiveExpected = -1; + self->_receiveReceived = 0; + self->_receiveLastProgress = 0; + self->_receiveDone = NO; + self->_receivePartialContent = NO; + self->_receivingFile = YES; + }); +} + +- (void)finishReceive:(NSString *)error { + _receiveDone = YES; + [_receiveFileHandle closeFile]; + _receiveFileHandle = nil; + [_clientDelegate onFileEnd:_id bytes:_receiveReceived error:error]; +} + +- (BOOL)parseReceiveHeader:(NSString *)head { + NSArray *lines = [head componentsSeparatedByString:@"\r\n"]; + NSArray *statusParts = + [lines.firstObject componentsSeparatedByString:@" "]; + NSInteger status = + statusParts.count > 1 ? [statusParts[1] integerValue] : -1; + if (status != 200 && status != 206) { + [self finishReceive:[NSString + stringWithFormat:@"HTTP %ld", (long)status]]; + return NO; + } + _receivePartialContent = status == 206; + for (NSString *line in lines) { + if ([line.lowercaseString hasPrefix:@"content-length:"]) { + NSString *value = [[line substringFromIndex:15] + stringByTrimmingCharactersInSet:[NSCharacterSet + whitespaceCharacterSet]]; + _receiveExpected = [value longLongValue]; + } + } + return YES; +} + +- (void)writeReceivedBody:(NSData *)body { + NSData *toWrite = body; + if (_receiveExpected >= 0) { + unsigned long long remaining = + (unsigned long long)_receiveExpected - _receiveReceived; + if (body.length > remaining) { + toWrite = + [body subdataWithRange:NSMakeRange(0, (NSUInteger)remaining)]; + } + } + if (toWrite.length > 0) { + NSError *error = nil; + if (!TcpWriteFile(_receiveFileHandle, toWrite, &error)) { + [self finishReceive:error.localizedDescription + ?: @"Cannot write the file"]; + return; + } + _receiveReceived += toWrite.length; + } + NSTimeInterval now = [NSDate timeIntervalSinceReferenceDate]; + if (now - _receiveLastProgress >= kFileProgressInterval) { + _receiveLastProgress = now; + [_clientDelegate onFileProgress:_id + bytes:_receiveReceived + total:_receiveExpected]; + } + if (_receiveExpected >= 0 && + _receiveReceived >= (unsigned long long)_receiveExpected) { + [self finishReceive:nil]; + } +} + +- (void)handleReceivedFileData:(NSData *)data { + if (_receiveDone) { + return; + } + if (_receiveFileHandle) { + [self writeReceivedBody:data]; + return; + } + [_receiveHeader appendData:data]; + NSData *headerEnd = [@"\r\n\r\n" dataUsingEncoding:NSASCIIStringEncoding]; + NSRange range = + [_receiveHeader rangeOfData:headerEnd + options:0 + range:NSMakeRange(0, _receiveHeader.length)]; + if (range.location == NSNotFound) { + if (_receiveHeader.length > kMaxHttpHeaderBytes) { + [self finishReceive:@"HTTP header too large"]; + } + return; + } + NSString *head = [[NSString alloc] + initWithData:[_receiveHeader + subdataWithRange:NSMakeRange(0, range.location)] + encoding:NSISOLatin1StringEncoding]; + if (![self parseReceiveHeader:head]) { + return; + } + // 206 Partial Content answers a Range request: resume by appending to the + // partial file. A 200 response starts the file again. + BOOL append = _receivePartialContent && [[NSFileManager defaultManager] + fileExistsAtPath:_receivePath]; + if (!append) { + [[NSFileManager defaultManager] createFileAtPath:_receivePath + contents:nil + attributes:nil]; + } + _receiveFileHandle = [NSFileHandle fileHandleForWritingAtPath:_receivePath]; + if (!_receiveFileHandle) { + [self finishReceive:@"Cannot create the destination file"]; + return; + } + if (append) { + [_receiveFileHandle seekToEndOfFile]; + } + NSUInteger bodyStart = range.location + range.length; + NSData *body = [_receiveHeader + subdataWithRange:NSMakeRange(bodyStart, + _receiveHeader.length - bodyStart)]; + _receiveHeader = nil; + [self writeReceivedBody:body]; +} + +- (void)endReceiveOnDisconnect { + if (!_receivingFile || _receiveDone) { + return; + } + if (!_receiveFileHandle) { + [self finishReceive:@"Connection closed before the HTTP header"]; + } else if (_receiveExpected >= 0 && + _receiveReceived < (unsigned long long)_receiveExpected) { + [self finishReceive:[NSString + stringWithFormat:@"Incomplete: %llu/%lld bytes", + _receiveReceived, + _receiveExpected]]; + } else { + [self finishReceive:nil]; + } +} + - (void)socket:(GCDAsyncSocket *)sock didWriteDataWithTag:(long)msgTag { + if (msgTag == kSendFileTag) { + NSTimeInterval now = [NSDate timeIntervalSinceReferenceDate]; + if (now - _sendFileLastProgress >= kFileProgressInterval) { + _sendFileLastProgress = now; + [_clientDelegate onFileProgress:_id + bytes:_sendFileSent + total:(long long)_sendFileLength]; + } + [self sendNextFileChunk]; + return; + } NSNumber *tagNum = [NSNumber numberWithLong:msgTag]; NSNumber *msgId = [self getPendingSend:tagNum]; if (msgId) { @@ -460,7 +755,11 @@ - (void)socket:(GCDAsyncSocket *)sock return; } - [_clientDelegate onData:@(tag) data:data]; + if (_receivingFile) { + [self handleReceivedFileData:data]; + } else { + [_clientDelegate onData:@(tag) data:data]; + } if (!_paused) { [sock readDataWithTimeout:-1 tag:tag]; } @@ -599,6 +898,10 @@ - (void)socketDidDisconnect:(GCDAsyncSocket *)sock withError:(NSError *)err { return; } + [self endReceiveOnDisconnect]; + [_sendFileHandle closeFile]; + _sendFileHandle = nil; + [_clientDelegate onClose:[sock userData] withError:(!err || err.code == GCDAsyncSocketClosedError ? nil : err)]; @@ -1033,7 +1336,19 @@ - (NSError *)badInvocationError:(NSString *)errMsg { } - (dispatch_queue_t)methodQueue { - return dispatch_get_main_queue(); + return [TcpSocketClient sharedQueue]; +} + ++ (dispatch_queue_t)sharedQueue { + // A private serial queue: native file transfers read and write the disk + // from the socket callbacks, which must not run on the main thread. + static dispatch_queue_t queue; + static dispatch_once_t onceToken; + dispatch_once(&onceToken, ^{ + queue = dispatch_queue_create("com.asterinet.react.tcpsocket.delegate", + DISPATCH_QUEUE_SERIAL); + }); + return queue; } @end diff --git a/ios/TcpSockets.m b/ios/TcpSockets.m index a5553c6..d9de131 100644 --- a/ios/TcpSockets.m +++ b/ios/TcpSockets.m @@ -19,10 +19,16 @@ @implementation TcpSockets { RCT_EXPORT_MODULE() +// Same serial queue as the socket delegate callbacks: `_clients` is only +// touched from it +- (dispatch_queue_t)methodQueue { + return [TcpSocketClient sharedQueue]; +} + - (NSArray *)supportedEvents { return @[ @"connect", @"listening", @"connection", @"secureConnection", @"data", - @"close", @"error", @"written", @"end" + @"close", @"error", @"written", @"end", @"fileProgress", @"fileEnd" ]; } @@ -101,6 +107,28 @@ - (TcpSocketClient *)createSocket:(nonnull NSNumber *)cId { [client writeData:data msgId:msgId]; } +RCT_EXPORT_METHOD(sendFile : (nonnull NSNumber *)cId path : (nonnull NSString *) + path offset : (double)offset length : (double) + length msgId : (nonnull NSNumber *)msgId) { + TcpSocketClient *client = [self findClient:cId]; + if (!client) + return; + + [client sendFile:path + offset:(unsigned long long)offset + length:(unsigned long long)length + msgId:msgId]; +} + +RCT_EXPORT_METHOD(receiveHttpBodyToFile : (nonnull NSNumber *) + cId path : (nonnull NSString *)path) { + TcpSocketClient *client = [self findClient:cId]; + if (!client) + return; + + [client receiveHttpBodyToFile:path]; +} + RCT_EXPORT_METHOD(end : (nonnull NSNumber *)cId) { [self endClient:cId]; } RCT_EXPORT_METHOD(destroy : (nonnull NSNumber *)cId) { @@ -315,6 +343,28 @@ - (void)onData:(NSNumber *)clientID data:(NSData *)data { }]; } +- (void)onFileProgress:(NSNumber *)clientID + bytes:(unsigned long long)bytes + total:(long long)total { + [self sendEventWithName:@"fileProgress" + body:@{ + @"id" : clientID, + @"bytes" : @(bytes), + @"total" : @(total) + }]; +} + +- (void)onFileEnd:(NSNumber *)clientID + bytes:(unsigned long long)bytes + error:(NSString *)error { + [self sendEventWithName:@"fileEnd" + body:@{ + @"id" : clientID, + @"bytes" : @(bytes), + @"error" : error ?: [NSNull null] + }]; +} + - (void)onEnd:(NSNumber *)clientID { [self sendEventWithName:@"end" body:@{ diff --git a/jest.setup.js b/jest.setup.js index 025f3b8..585f8e3 100644 --- a/jest.setup.js +++ b/jest.setup.js @@ -13,6 +13,8 @@ jest.mock('react-native', () => { end: jest.fn(), destroy: jest.fn(), write: jest.fn(), + sendFile: jest.fn(), + receiveHttpBodyToFile: jest.fn(), listen: jest.fn(), pause: jest.fn(), resume: jest.fn(), diff --git a/lib/types/Socket.d.ts b/lib/types/Socket.d.ts index abf5383..7031cfb 100644 --- a/lib/types/Socket.d.ts +++ b/lib/types/Socket.d.ts @@ -33,6 +33,8 @@ * @property {(err: Error) => void} error * @property {() => void} timeout * @property {() => void} secureConnect + * @property {(bytes: number, total: number) => void} fileProgress + * @property {(bytes: number, error: string | null) => void} fileEnd * * @extends {EventEmitter} */ @@ -190,6 +192,36 @@ export default class Socket extends EventEmitter void) | undefined): boolean; + /** + * Sends `length` bytes of the file at `path`, starting at `offset`. The file is read and written + * to the socket natively: its bytes never go through JS. + * + * The data is sent after the pending `socket.write()` calls. Wait for the returned promise + * before writing to the socket or sending another file. + * + * `'fileProgress'` is emitted with `(bytes, total)` while the file is sent. + * + * @param {string} path Absolute path of the file, with or without the `file://` scheme + * @param {number} offset Position of the first byte to send + * @param {number} length Number of bytes to send + * + * @return {Promise} Resolved once the bytes are written out + */ + sendFile(path: string, offset: number, length: number): Promise; + /** + * From now on, the incoming bytes are parsed natively as one HTTP response and its body is + * written to the file at `path`, instead of being emitted as `'data'` events. Call it before + * sending the request. + * + * A `200` response replaces the file. A `206` response is appended to it, to resume a download + * with a `Range` request. Any other status fails. + * + * `'fileProgress'` is emitted with `(bytes, total)` while the body is received, then `'fileEnd'` + * with `(bytes, error)`, `error` being `null` on success. + * + * @param {string} path Absolute path of the file, with or without the `file://` scheme + */ + receiveHttpBodyToFile(path: string): Socket; /** * Pauses the reading of data. That is, `'data'` events will not be emitted. Useful to throttle back an upload. */ @@ -218,6 +250,8 @@ export default class Socket extends EventEmitter void; timeout: () => void; secureConnect: () => void; + fileProgress: (bytes: number, total: number) => void; + fileEnd: (bytes: number, error: string | null) => void; }; import EventEmitter from "eventemitter3"; import { Buffer } from "buffer"; diff --git a/src/Socket.js b/src/Socket.js index 9913e5a..527b2be 100644 --- a/src/Socket.js +++ b/src/Socket.js @@ -40,6 +40,8 @@ import { nativeEventEmitter, getNextId } from './Globals'; * @property {(err: Error) => void} error * @property {() => void} timeout * @property {() => void} secureConnect + * @property {(bytes: number, total: number) => void} fileProgress + * @property {(bytes: number, error: string | null) => void} fileEnd * * @extends {EventEmitter} */ @@ -368,6 +370,85 @@ export default class Socket extends EventEmitter { return ok; } + /** + * Sends `length` bytes of the file at `path`, starting at `offset`. The file is read and written + * to the socket natively: its bytes never go through JS. + * + * The data is sent after the pending `socket.write()` calls. Wait for the returned promise + * before writing to the socket or sending another file. + * + * `'fileProgress'` is emitted with `(bytes, total)` while the file is sent. + * + * @param {string} path Absolute path of the file, with or without the `file://` scheme + * @param {number} offset Position of the first byte to send + * @param {number} length Number of bytes to send + * + * @return {Promise} Resolved once the bytes are written out + */ + sendFile(path, offset, length) { + if (this._pending || this._destroyed) return Promise.reject(new Error('Socket is closed.')); + + const currentMsgId = this._msgId; + this._msgId = (this._msgId + 1) % Number.MAX_SAFE_INTEGER; + return new Promise((resolve, reject) => { + const removeListeners = () => { + this._msgEvtEmitter.removeListener('written', msgEvtHandler); + this.removeListener('error', errorHandler); + this.removeListener('close', closeHandler); + }; + const msgEvtHandler = ( + /** @type {{id: number, msgId: number, err?: string}} */ evt + ) => { + const { msgId, err } = evt; + if (msgId !== currentMsgId) return; + removeListeners(); + this._lastRcvMsgId = msgId; + this._resetTimeout(); + if (this.writableNeedDrain && this._lastSentMsgId === msgId) { + this.writableNeedDrain = false; + this.emit('drain'); + } + if (err) reject(new Error(err)); + else resolve(); + }; + // A failed transfer is only reported as `'error'` on iOS + const errorHandler = (/** @type {Error | string} */ err) => { + removeListeners(); + reject(err instanceof Error ? err : new Error(err)); + }; + const closeHandler = () => { + removeListeners(); + reject(new Error('Socket closed before the file was sent.')); + }; + this._msgEvtEmitter.on('written', msgEvtHandler, this); + this.on('error', errorHandler); + this.on('close', closeHandler); + this._lastSentMsgId = currentMsgId; + this._bytesWritten += length; + NativeModules.TcpSockets.sendFile(this._id, path, offset, length, currentMsgId); + }); + } + + /** + * From now on, the incoming bytes are parsed natively as one HTTP response and its body is + * written to the file at `path`, instead of being emitted as `'data'` events. Call it before + * sending the request. + * + * A `200` response replaces the file. A `206` response is appended to it, to resume a download + * with a `Range` request. Any other status fails. + * + * `'fileProgress'` is emitted with `(bytes, total)` while the body is received, then `'fileEnd'` + * with `(bytes, error)`, `error` being `null` on success. + * + * @param {string} path Absolute path of the file, with or without the `file://` scheme + */ + receiveHttpBodyToFile(path) { + if (this._pending || this._destroyed) throw new Error('Socket is closed.'); + + NativeModules.TcpSockets.receiveHttpBodyToFile(this._id, path); + return this; + } + /** * Pauses the reading of data. That is, `'data'` events will not be emitted. Useful to throttle back an upload. */ @@ -486,6 +567,15 @@ export default class Socket extends EventEmitter { if (evt.id !== this._id) return; this._msgEvtEmitter.emit('written', evt); }); + this._fileProgressListener = this._eventEmitter.addListener('fileProgress', (evt) => { + if (evt.id !== this._id) return; + this._resetTimeout(); + this.emit('fileProgress', evt.bytes, evt.total); + }); + this._fileEndListener = this._eventEmitter.addListener('fileEnd', (evt) => { + if (evt.id !== this._id) return; + this.emit('fileEnd', evt.bytes, evt.error || null); + }); } /** @@ -498,6 +588,8 @@ export default class Socket extends EventEmitter { this._endListener?.remove(); this._connectListener?.remove(); this._writtenListener?.remove(); + this._fileProgressListener?.remove(); + this._fileEndListener?.remove(); } /**