fix(android): leaking adb socket connections (#4730)

android-types-internal.d.ts

@@ -52,6 +52,7 @@ export interface AndroidDevice<BrowserContextOptions, BrowserContext, Page> exte
 
 export interface AndroidSocket extends EventEmitter {
   on(event: 'data', handler: (data: Buffer) => void): this;
+  on(event: 'close', handler: () => void): this;
   write(data: Buffer): Promise<void>
   close(): Promise<void>
 }

src/client/android.ts

@@ -262,6 +262,7 @@ export class AndroidSocket extends ChannelOwner<channels.AndroidSocketChannel, c
   constructor(parent: ChannelOwner, type: string, guid: string, initializer: channels.AndroidSocketInitializer) {
     super(parent, type, guid, initializer);
     this._channel.on('data', ({ data }) => this.emit(Events.AndroidSocket.Data, Buffer.from(data, 'base64')));
+    this._channel.on('close', () => this.emit(Events.AndroidSocket.Close));
   }
 
   async write(data: Buffer): Promise<void> {

src/client/events.ts

@@ -22,7 +22,8 @@ export const Events = {
   },
 
   AndroidSocket: {
-    Data: 'data'
+    Data: 'data',
+    Close: 'close'
   },
 
   AndroidWebView: {

src/dispatchers/androidDispatcher.ts

@@ -177,6 +177,10 @@ export class AndroidSocketDispatcher extends Dispatcher<SocketBackend, channels.
   constructor(scope: DispatcherScope, socket: SocketBackend) {
     super(scope, socket, 'AndroidSocket', {}, true);
     socket.on(Events.AndroidSocket.Data, (data: Buffer) => this._dispatchEvent('data', { data: data.toString('base64') }));
+    socket.on(Events.AndroidSocket.Close, () => {
+      this._dispatchEvent('close');
+      this._dispose();
+    });
   }
 
   async write(params: channels.AndroidSocketWriteParams, metadata?: channels.Metadata): Promise<void> {

src/protocol/channels.ts

@@ -2425,12 +2425,14 @@ export type AndroidSetDefaultTimeoutNoReplyResult = void;
 export type AndroidSocketInitializer = {};
 export interface AndroidSocketChannel extends Channel {
   on(event: 'data', callback: (params: AndroidSocketDataEvent) => void): this;
+  on(event: 'close', callback: (params: AndroidSocketCloseEvent) => void): this;
   write(params: AndroidSocketWriteParams, metadata?: Metadata): Promise<AndroidSocketWriteResult>;
   close(params?: AndroidSocketCloseParams, metadata?: Metadata): Promise<AndroidSocketCloseResult>;
 }
 export type AndroidSocketDataEvent = {
   data: Binary,
 };
+export type AndroidSocketCloseEvent = {};
 export type AndroidSocketWriteParams = {
   data: Binary,
 };

src/protocol/protocol.yml

@@ -2097,6 +2097,7 @@ AndroidSocket:
     data:
       parameters:
         data: binary
+    close:
 
 AndroidDevice:
   type: interface

src/server/android/android.ts

@@ -171,7 +171,7 @@ export class AndroidDevice extends EventEmitter {
       await this.installApk(await readFileAsync(require.resolve(`../../../bin/${file}`)));
 
     debug('pw:android')('Starting the new driver');
-    this.shell(`am instrument -w com.microsoft.playwright.androiddriver.test/androidx.test.runner.AndroidJUnitRunner`);
+    this.shell('am instrument -w com.microsoft.playwright.androiddriver.test/androidx.test.runner.AndroidJUnitRunner');
     const socket = await this._waitForLocalAbstract('playwright_android_driver_socket');
     const transport = new Transport(socket, socket, socket, 'be');
     transport.onmessage = message => {
@@ -285,6 +285,7 @@ export class AndroidDevice extends EventEmitter {
     await installSocket.write(content);
     const success = await new Promise(f => installSocket.on('data', f));
     debug('pw:android')('Written driver bytes: ' + success);
+    await installSocket.close();
   }
 
   async push(content: Buffer, path: string, mode = 0o644): Promise<void> {

src/server/android/backendAdb.ts

@@ -68,11 +68,15 @@ async function runCommand(command: string, serial?: string): Promise<Buffer> {
   await socket.write(encodeMessage(command));
   const status = await socket.read(4);
   assert(status.toString() === 'OKAY', status.toString());
+  let commandOutput: Buffer;
   if (!command.startsWith('shell:')) {
     const remainingLength = parseInt((await socket.read(4)).toString(), 16);
-    return (await socket.read(remainingLength));
+    commandOutput = await socket.read(remainingLength);
+  } else {
+    commandOutput = await socket.readAll();
   }
-  return (await socket.readAll());
+  await socket.close();
+  return commandOutput;
 }
 
 async function open(command: string, serial?: string): Promise<BufferedSocketWrapper> {
@@ -122,6 +126,7 @@ class BufferedSocketWrapper extends EventEmitter implements SocketBackend {
       this._isClosed = true;
       if (this._notifyReader)
         this._notifyReader();
+      this.close();
       this.emit('close');
     });
     this._socket.on('error', error => this.emit('error', error));
@@ -134,6 +139,8 @@ class BufferedSocketWrapper extends EventEmitter implements SocketBackend {
   }
 
   async close() {
+    if (this._isClosed)
+      return;
     debug('pw:adb')('Close ' + this._command);
     this._socket.destroy();
   }

test/android/android.fixtures.ts

@@ -29,7 +29,7 @@ fixtures.device.init(async ({ playwright }, runTest) => {
   await device.shell('am force-stop com.android.chrome');
   device.setDefaultTimeout(120000);
   await runTest(device);
-  device.close();
+  await device.close();
 });
 
 export const folio = fixtures.build();

test/android/device.spec.ts

@@ -31,6 +31,9 @@ if (process.env.PW_ANDROID_TESTS) {
     await socket.write(Buffer.from('321\n'));
     const output = await new Promise(resolve => socket.on('data', resolve));
     expect(output.toString()).toBe('321\n');
+    const closedPromise = new Promise(resolve => socket.on('close', resolve));
+    await socket.close();
+    await closedPromise;
   });
 
   it('androidDevice.screenshot', async function({ device, testInfo }) {