Add support for socket timeouts (#303)

* Add support for socket timeouts

Signed-off-by: Wolfgang Haupt <[email protected]>

* Add new timeout option to Readme

Signed-off-by: Wolfgang Haupt <[email protected]>

Author: a1rwulf

README.md

@@ -196,6 +196,7 @@ Creates a client instance with the provided options. Note that this will not aut
     -   **port** The port where the ADB server is listening. Defaults to `5037`.
     -   **host** The host of the ADB server. Defaults to `'127.0.0.1'`.
     -   **bin** As the sole exception, this option provides the path to the `adb` binary, used for starting the server locally if initial connection fails. Defaults to `'adb'`.
+    -   **timeout** The socket timeout in ms as defined by [Net.connect][net-connect]'s options. Defaults to `0` which means no timeout.
 -   Returns: The client instance.
 
 #### adb.util.parsePublicKey(androidKey)

src/adb.ts

@@ -7,6 +7,7 @@ interface Options {
   host?: string;
   port?: number;
   bin?: string;
+  timeout?: number;
 }
 
 export default class Adb {
@@ -18,6 +19,7 @@ export default class Adb {
       bin: options.bin,
       host: options.host || process.env.ADB_HOST,
       port: options.port || 5037,
+      timeout: options.timeout || 0
     };
     if (!opts.port) {
       const port = parseInt(process.env.ADB_PORT || '5037', 10);

src/adb/client.ts

@@ -24,13 +24,15 @@ export default class Client extends EventEmitter {
   public readonly host: string;
   public readonly port: number | string;
   public readonly bin: string;
+  public readonly timeout: number;
 
-  constructor({ host = '127.0.0.1', port = 5037, bin = 'adb' }: ClientOptions = { port: 5037 }) {
+  constructor({ host = '127.0.0.1', port = 5037, bin = 'adb', timeout = 0 }: ClientOptions = { port: 5037 }) {
     super();
     this.host = host;
     this.port = port;
     this.bin = bin;
-    this.options = { host, port, bin };
+    this.timeout = timeout;
+    this.options = { host, port, bin, timeout };
   }
 
   public createTcpUsbBridge(serial: string, options: SocketOptions): TcpUsbServer {

src/adb/parser.ts

@@ -31,6 +31,15 @@ class UnexpectedDataError extends Error {
   }
 }
 
+class TimeoutError extends Error {
+  constructor() {
+    super(`Socket timeout reached.`);
+    Object.setPrototypeOf(this, TimeoutError.prototype);
+    this.name = 'TimeoutError';
+    Error.captureStackTrace(this, TimeoutError);
+  }
+}
+
 Bluebird.config({
   // Enable warnings
   // warnings: true,
@@ -46,6 +55,7 @@ export default class Parser {
   public static FailError = FailError;
   public static PrematureEOFError = PrematureEOFError;
   public static UnexpectedDataError = UnexpectedDataError;
+  public static TimeoutError = TimeoutError;
   private ended = false;
 
   constructor(public stream: Duplex) {
@@ -59,6 +69,7 @@ export default class Parser {
     let tryRead: () => void;
     let errorListener: (error: Error) => void;
     let endListener: () => void;
+    let timeoutListener: () => void;
     return new Bluebird<boolean>((resolve, reject, onCancel) => {
       tryRead = () => {
         while (this.stream.read()) {
@@ -72,9 +83,14 @@ export default class Parser {
         this.ended = true;
         return resolve(true);
       };
+      timeoutListener = () => {
+        this.stream.end()
+        return reject(new Parser.TimeoutError());
+      };
       this.stream.on('readable', tryRead);
       this.stream.on('error', errorListener);
       this.stream.on('end', endListener);
+      this.stream.on('timeout', timeoutListener);
       this.stream.read(0);
       this.stream.end();
       onCancel(() => {
@@ -84,6 +100,7 @@ export default class Parser {
       this.stream.removeListener('readable', tryRead);
       this.stream.removeListener('error', errorListener);
       this.stream.removeListener('end', endListener);
+      this.stream.removeListener('timeout', timeoutListener);
       // return r;
     });
   }
@@ -98,6 +115,7 @@ export default class Parser {
     let tryRead: () => void;
     let errorListener: (error: Error) => void;
     let endListener: () => void;
+    let timeoutListener: () => void;
 
     return new Bluebird<Buffer>((resolve, reject, onCancel) => {
       tryRead = () => {
@@ -116,9 +134,14 @@ export default class Parser {
         this.ended = true;
         return resolve(all);
       };
+      timeoutListener = () => {
+        this.stream.end()
+        return reject(new Parser.TimeoutError());
+      };
       this.stream.on('readable', tryRead);
       this.stream.on('error', errorListener);
       this.stream.on('end', endListener);
+      this.stream.on('timeout', timeoutListener);
       tryRead();
       onCancel(() => {
         // console.log('2-onCanceled');
@@ -127,6 +150,7 @@ export default class Parser {
       this.stream.removeListener('readable', tryRead);
       this.stream.removeListener('error', errorListener);
       this.stream.removeListener('end', endListener);
+      this.stream.removeListener('timeout', timeoutListener);
     });
   }
 
@@ -138,6 +162,7 @@ export default class Parser {
     let tryRead: () => void;
     let errorListener: (error: Error) => void;
     let endListener: () => void;
+    let timeoutListener: () => void;
     return new Bluebird<Buffer>((resolve, reject /*, onCancel*/) => {
       tryRead = () => {
         if (howMany) {
@@ -161,23 +186,30 @@ export default class Parser {
         this.ended = true;
         return reject(new Parser.PrematureEOFError(howMany));
       };
+      timeoutListener = () => {
+        this.stream.end()
+        return reject(new Parser.TimeoutError());
+      };
       errorListener = (err) => reject(err);
       this.stream.on('readable', tryRead);
       this.stream.on('error', errorListener);
       this.stream.on('end', endListener);
+      this.stream.on('timeout', timeoutListener);
       tryRead();
       // onCancel(() => {});
     }).finally(() => {
       this.stream.removeListener('readable', tryRead);
       this.stream.removeListener('error', errorListener);
       this.stream.removeListener('end', endListener);
+      this.stream.removeListener('timeout', timeoutListener);
     });
   }
 
   public readByteFlow(howMany: number, targetStream: Duplex): Bluebird<void> {
     let tryRead: () => void;
     let errorListener: (error: Error) => void;
     let endListener: () => void;
+    let timeoutListener: () => void;
     return new Bluebird<void>((resolve, reject /*, onCancel*/) => {
       tryRead = () => {
         if (howMany) {
@@ -206,15 +238,21 @@ export default class Parser {
       errorListener = function (err) {
         return reject(err);
       };
+      timeoutListener = () => {
+        this.stream.end()
+        return reject(new Parser.TimeoutError());
+      };
       this.stream.on('readable', tryRead);
       this.stream.on('error', errorListener);
       this.stream.on('end', endListener);
+      this.stream.on('timeout', timeoutListener);
       tryRead();
       // onCancel(() => {});
     }).finally(() => {
       this.stream.removeListener('readable', tryRead);
       this.stream.removeListener('error', errorListener);
       this.stream.removeListener('end', endListener);
+      this.stream.removeListener('timeout', timeoutListener);
     });
   }