fix(downloads): make path/saveAs work when connected remotely (#3634)

- saveAs uses a stream internally and pipes it to the local file;
- path throws an error when called on a remote browser.

src/client/browser.ts

@@ -30,6 +30,7 @@ export class Browser extends ChannelOwner<channels.BrowserChannel, channels.Brow
   private _isClosedOrClosing = false;
   private _closedPromise: Promise<void>;
   readonly _browserType: BrowserType;
+  _isRemote = false;
 
   static from(browser: channels.BrowserChannel): Browser {
     return (browser as any)._object;

src/client/browserType.ts

@@ -146,6 +146,7 @@ export class BrowserType extends ChannelOwner<channels.BrowserTypeChannel, chann
         ws.addEventListener('open', async () => {
           const browser = (await connection.waitForObjectWithKnownName('connectedBrowser')) as Browser;
           browser._logger = logger;
+          browser._isRemote = true;
           const closeListener = () => {
             // Emulate all pages, contexts and the browser closing upon disconnect.
             for (const context of browser.contexts()) {

src/client/download.ts

@@ -18,14 +18,21 @@ import * as channels from '../protocol/channels';
 import { ChannelOwner } from './channelOwner';
 import { Readable } from 'stream';
 import { Stream } from './stream';
+import { Browser } from './browser';
+import { BrowserContext } from './browserContext';
+import * as fs from 'fs';
+import { mkdirIfNeeded } from '../utils/utils';
 
 export class Download extends ChannelOwner<channels.DownloadChannel, channels.DownloadInitializer> {
+  private _browser: Browser | undefined;
+
   static from(download: channels.DownloadChannel): Download {
     return (download as any)._object;
   }
 
   constructor(parent: ChannelOwner, type: string, guid: string, initializer: channels.DownloadInitializer) {
     super(parent, type, guid, initializer);
+    this._browser = (parent as BrowserContext)._browser;
   }
 
   url(): string {
@@ -37,12 +44,26 @@ export class Download extends ChannelOwner<channels.DownloadChannel, channels.Do
   }
 
   async path(): Promise<string | null> {
+    if (this._browser && this._browser._isRemote)
+      throw new Error(`Path is not available when using browserType.connect(). Use download.saveAs() to save a local copy.`);
     return (await this._channel.path()).value || null;
   }
 
   async saveAs(path: string): Promise<void> {
     return this._wrapApiCall('download.saveAs', async () => {
-      await this._channel.saveAs({ path });
+      if (!this._browser || !this._browser._isRemote) {
+        await this._channel.saveAs({ path });
+        return;
+      }
+
+      const result = await this._channel.saveAsStream();
+      const stream = Stream.from(result.stream);
+      await mkdirIfNeeded(path);
+      await new Promise((resolve, reject) => {
+        stream.stream().pipe(fs.createWriteStream(path))
+            .on('finish' as any, resolve)
+            .on('error' as any, reject);
+      });
     });
   }
 

src/client/stream.ts

@@ -47,4 +47,10 @@ class StreamImpl extends Readable {
     else
       this.push(null);
   }
+
+  _destroy(error: Error | null, callback: (error: Error | null) => void): void {
+    // Stream might be destroyed after the connection was closed.
+    this._channel.close().catch(e => null);
+    super._destroy(error, callback);
+  }
 }

src/dispatchers/downloadDispatcher.ts

@@ -18,6 +18,9 @@ import { Download } from '../server/download';
 import * as channels from '../protocol/channels';
 import { Dispatcher, DispatcherScope } from './dispatcher';
 import { StreamDispatcher } from './streamDispatcher';
+import * as fs from 'fs';
+import * as util from 'util';
+import { mkdirIfNeeded } from '../utils/utils';
 
 export class DownloadDispatcher extends Dispatcher<Download, channels.DownloadInitializer> implements channels.DownloadChannel {
   constructor(scope: DispatcherScope, download: Download) {
@@ -28,20 +31,63 @@ export class DownloadDispatcher extends Dispatcher<Download, channels.DownloadIn
   }
 
   async path(): Promise<channels.DownloadPathResult> {
-    const path = await this._object.path();
+    const path = await this._object.localPath();
     return { value: path || undefined };
   }
 
-  async saveAs(params: channels.DownloadSaveAsParams): Promise<void> {
+  async saveAs(params: channels.DownloadSaveAsParams): Promise<channels.DownloadSaveAsResult> {
-    await this._object.saveAs(params.path);
+    return await new Promise((resolve, reject) => {
+      this._object.saveAs(async (localPath, error) => {
+        if (error !== undefined) {
+          reject(error);
+          return;
+        }
+
+        try {
+          await mkdirIfNeeded(params.path);
+          await util.promisify(fs.copyFile)(localPath, params.path);
+          resolve();
+        } catch (e) {
+          reject(e);
+        }
+      });
+    });
+  }
+
+  async saveAsStream(): Promise<channels.DownloadSaveAsStreamResult> {
+    return await new Promise((resolve, reject) => {
+      this._object.saveAs(async (localPath, error) => {
+        if (error !== undefined) {
+          reject(error);
+          return;
+        }
+
+        try {
+          const readable = fs.createReadStream(localPath);
+          await new Promise(f => readable.on('readable', f));
+          const stream = new StreamDispatcher(this._scope, readable);
+          // Resolve with a stream, so that client starts saving the data.
+          resolve({ stream });
+          // Block the download until the stream is consumed.
+          await new Promise<void>(resolve => {
+            readable.on('close', resolve);
+            readable.on('end', resolve);
+            readable.on('error', resolve);
+          });
+        } catch (e) {
+          reject(e);
+        }
+      });
+    });
   }
 
   async stream(): Promise<channels.DownloadStreamResult> {
-    const stream = await this._object.createReadStream();
+    const fileName = await this._object.localPath();
-    if (!stream)
+    if (!fileName)
       return {};
-    await new Promise(f => stream.on('readable', f));
+    const readable = fs.createReadStream(fileName);
-    return { stream: new StreamDispatcher(this._scope, stream) };
+    await new Promise(f => readable.on('readable', f));
+    return { stream: new StreamDispatcher(this._scope, readable) };
   }
 
   async failure(): Promise<channels.DownloadFailureResult> {

src/dispatchers/streamDispatcher.ts

@@ -27,4 +27,8 @@ export class StreamDispatcher extends Dispatcher<stream.Readable, channels.Strea
     const buffer = this._object.read(Math.min(this._object.readableLength, params.size || this._object.readableLength));
     return { binary: buffer ? buffer.toString('base64') : '' };
   }
+
+  async close() {
+    this._object.destroy();
+  }
 }

src/protocol/channels.ts

@@ -2104,6 +2104,7 @@ export type DownloadInitializer = {
 export interface DownloadChannel extends Channel {
   path(params?: DownloadPathParams): Promise<DownloadPathResult>;
   saveAs(params: DownloadSaveAsParams): Promise<DownloadSaveAsResult>;
+  saveAsStream(params?: DownloadSaveAsStreamParams): Promise<DownloadSaveAsStreamResult>;
   failure(params?: DownloadFailureParams): Promise<DownloadFailureResult>;
   stream(params?: DownloadStreamParams): Promise<DownloadStreamResult>;
   delete(params?: DownloadDeleteParams): Promise<DownloadDeleteResult>;
@@ -2120,6 +2121,11 @@ export type DownloadSaveAsOptions = {
 
 };
 export type DownloadSaveAsResult = void;
+export type DownloadSaveAsStreamParams = {};
+export type DownloadSaveAsStreamOptions = {};
+export type DownloadSaveAsStreamResult = {
+  stream: StreamChannel,
+};
 export type DownloadFailureParams = {};
 export type DownloadFailureOptions = {};
 export type DownloadFailureResult = {
@@ -2138,6 +2144,7 @@ export type DownloadDeleteResult = void;
 export type StreamInitializer = {};
 export interface StreamChannel extends Channel {
   read(params: StreamReadParams): Promise<StreamReadResult>;
+  close(params?: StreamCloseParams): Promise<StreamCloseResult>;
 }
 export type StreamReadParams = {
   size?: number,
@@ -2148,6 +2155,9 @@ export type StreamReadOptions = {
 export type StreamReadResult = {
   binary: Binary,
 };
+export type StreamCloseParams = {};
+export type StreamCloseOptions = {};
+export type StreamCloseResult = void;
 
 // ----------- CDPSession -----------
 export type CDPSessionInitializer = {};

src/protocol/protocol.yml

@@ -1769,10 +1769,16 @@ Download:
       returns:
         value: string?
 
+    # Blocks path/failure/delete/context.close until saved to the local |path|.
     saveAs:
       parameters:
         path: string
 
+    # Blocks path/failure/delete/context.close until the stream is closed.
+    saveAsStream:
+      returns:
+        stream: Stream
+
     failure:
       returns:
         error: string?
@@ -1796,6 +1802,7 @@ Stream:
       returns:
         binary: binary
 
+    close:
 
 
 CDPSession:

src/protocol/validator.ts

@@ -804,12 +804,14 @@ export function createScheme(tChannel: (name: string) => Validator): Scheme {
   scheme.DownloadSaveAsParams = tObject({
     path: tString,
   });
+  scheme.DownloadSaveAsStreamParams = tOptional(tObject({}));
   scheme.DownloadFailureParams = tOptional(tObject({}));
   scheme.DownloadStreamParams = tOptional(tObject({}));
   scheme.DownloadDeleteParams = tOptional(tObject({}));
   scheme.StreamReadParams = tObject({
     size: tOptional(tNumber),
   });
+  scheme.StreamCloseParams = tOptional(tObject({}));
   scheme.CDPSessionSendParams = tObject({
     method: tString,
     params: tOptional(tAny),

src/server/download.ts

@@ -18,15 +18,16 @@ import * as path from 'path';
 import * as fs from 'fs';
 import * as util from 'util';
 import { Page } from './page';
-import { Readable } from 'stream';
+import { assert } from '../utils/utils';
-import { assert, mkdirIfNeeded } from '../utils/utils';
+
+type SaveCallback = (localPath: string, error?: string) => Promise<void>;
 
 export class Download {
   private _downloadsPath: string;
   private _uuid: string;
   private _finishedCallback: () => void;
   private _finishedPromise: Promise<void>;
-  private _saveAsRequests: { fulfill: () => void; reject: (error?: any) => void; path: string }[] = [];
+  private _saveCallbacks: SaveCallback[] = [];
   private _finished: boolean = false;
   private _page: Page;
   private _acceptDownloads: boolean;
@@ -63,7 +64,7 @@ export class Download {
     return this._suggestedFilename!;
   }
 
-  async path(): Promise<string | null> {
+  async localPath(): Promise<string | null> {
     if (!this._acceptDownloads)
       throw new Error('Pass { acceptDownloads: true } when you are creating your browser context.');
     const fileName = path.join(this._downloadsPath, this._uuid);
@@ -73,7 +74,7 @@ export class Download {
     return fileName;
   }
 
-  async saveAs(path: string) {
+  saveAs(saveCallback: SaveCallback) {
     if (!this._acceptDownloads)
       throw new Error('Pass { acceptDownloads: true } when you are creating your browser context.');
     if (this._deleted)
@@ -82,17 +83,10 @@ export class Download {
       throw new Error('Download not found on disk. Check download.failure() for details.');
 
     if (this._finished) {
-      await this._saveAs(path);
+      saveCallback(path.join(this._downloadsPath, this._uuid));
       return;
     }
-
+    this._saveCallbacks.push(saveCallback);
-    return new Promise((fulfill, reject) => this._saveAsRequests.push({fulfill, reject, path}));
-  }
-
-  async _saveAs(downloadPath: string) {
-    const fileName = path.join(this._downloadsPath, this._uuid);
-    await mkdirIfNeeded(downloadPath);
-    await util.promisify(fs.copyFile)(fileName, downloadPath);
   }
 
   async failure(): Promise<string | null> {
@@ -102,15 +96,10 @@ export class Download {
     return this._failure;
   }
 
-  async createReadStream(): Promise<Readable | null> {
-    const fileName = await this.path();
-    return fileName ? fs.createReadStream(fileName) : null;
-  }
-
   async delete(): Promise<void> {
     if (!this._acceptDownloads)
       return;
-    const fileName = await this.path();
+    const fileName = await this.localPath();
     if (this._deleted)
       return;
     this._deleted = true;
@@ -123,18 +112,14 @@ export class Download {
     this._failure = error || null;
 
     if (error) {
-      for (const { reject } of this._saveAsRequests)
+      for (const callback of this._saveCallbacks)
-        reject(error);
+        callback('', error);
     } else {
-      for (const { fulfill, reject, path } of this._saveAsRequests) {
+      const fullPath = path.join(this._downloadsPath, this._uuid);
-        try {
+      for (const callback of this._saveCallbacks)
-          await this._saveAs(path);
+        await callback(fullPath);
-          fulfill();
-        } catch (err) {
-          reject(err);
-        }
-      }
     }
+    this._saveCallbacks = [];
 
     this._finishedCallback();
   }

test/browsertype-connect-subprocess.spec.ts

@@ -1,46 +0,0 @@
-/**
- * Copyright Microsoft Corporation. All rights reserved.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-import { options } from './playwright.fixtures';
-import './remoteServer.fixture';
-import utils from './utils';
-
-it.skip(options.WIRE).slow()('should connect to server from another process', async({ browserType, remoteServer }) => {
-  const browser = await browserType.connect({ wsEndpoint: remoteServer.wsEndpoint() });
-  const page = await browser.newPage();
-  expect(await page.evaluate('2 + 3')).toBe(5);
-  await browser.close();
-});
-
-it.skip(options.WIRE).fail(true).slow()('should respect selectors in another process', async({ playwright, browserType, remoteServer }) => {
-  const mycss = () => ({
-    create(root, target) {},
-    query(root, selector) {
-      return root.querySelector(selector);
-    },
-    queryAll(root: HTMLElement, selector: string) {
-      return Array.from(root.querySelectorAll(selector));
-    }
-  });
-  await utils.registerEngine(playwright, 'mycss', mycss);
-
-  const browser = await browserType.connect({ wsEndpoint: remoteServer.wsEndpoint() });
-  const page = await browser.newPage();
-  await page.setContent(`<div>hello</div>`);
-  expect(await page.innerHTML('css=div')).toBe('hello');
-  expect(await page.innerHTML('mycss=div')).toBe('hello');
-  await browser.close();
-});

test/browsertype-connect.spec.ts

@@ -17,6 +17,7 @@
 
 import { options } from './playwright.fixtures';
 import utils from './utils';
+import './remoteServer.fixture';
 
 it.skip(options.WIRE).slow()('should be able to reconnect to a browser', async({browserType, defaultBrowserOptions, server}) => {
   const browserServer = await browserType.launchServer(defaultBrowserOptions);
@@ -37,6 +38,13 @@ it.skip(options.WIRE).slow()('should be able to reconnect to a browser', async({
   await browserServer.close();
 });
 
+it.skip(options.WIRE)('should connect to a remote server', async({ browserType, remoteServer }) => {
+  const browser = await browserType.connect({ wsEndpoint: remoteServer.wsEndpoint() });
+  const page = await browser.newPage();
+  expect(await page.evaluate('2 + 3')).toBe(5);
+  await browser.close();
+});
+
 it.skip(options.WIRE).fail(options.CHROMIUM && WIN).slow()('should handle exceptions during connect', async({browserType, defaultBrowserOptions}) => {
   const browserServer = await browserType.launchServer(defaultBrowserOptions);
   const __testHookBeforeCreateBrowser = () => { throw new Error('Dummy') };
@@ -87,3 +95,23 @@ it.skip(options.WIRE)('should respect selectors', async({playwright, browserType
   expect(await page.innerHTML('mycss=div')).toBe('hello');
   await browserServer.close();
 });
+
+it.skip(options.WIRE).fail(true)('should respect selectors when connected remotely', async({ playwright, browserType, remoteServer }) => {
+  const mycss = () => ({
+    create(root, target) {},
+    query(root, selector) {
+      return root.querySelector(selector);
+    },
+    queryAll(root: HTMLElement, selector: string) {
+      return Array.from(root.querySelectorAll(selector));
+    }
+  });
+  await utils.registerEngine(playwright, 'mycss', mycss);
+
+  const browser = await browserType.connect({ wsEndpoint: remoteServer.wsEndpoint() });
+  const page = await browser.newPage();
+  await page.setContent(`<div>hello</div>`);
+  expect(await page.innerHTML('css=div')).toBe('hello');
+  expect(await page.innerHTML('mycss=div')).toBe('hello');
+  await browser.close();
+});

test/download.spec.ts

@@ -15,6 +15,7 @@
  */
 
 import { options } from './playwright.fixtures';
+import './remoteServer.fixture';
 
 import fs from 'fs';
 import path from 'path';
@@ -143,6 +144,23 @@ it('should create subdirectories when saving to non-existent user-specified path
   await page.close();
 });
 
+it.skip(options.WIRE)('should save when connected remotely', async({tmpDir, server, browserType, remoteServer}) => {
+  const browser = await browserType.connect({ wsEndpoint: remoteServer.wsEndpoint() });
+  const page = await browser.newPage({ acceptDownloads: true });
+  await page.setContent(`<a href="${server.PREFIX}/download">download</a>`);
+  const [ download ] = await Promise.all([
+    page.waitForEvent('download'),
+    page.click('a')
+  ]);
+  const nestedPath = path.join(tmpDir, "these", "are", "directories", "download.txt");
+  await download.saveAs(nestedPath);
+  expect(fs.existsSync(nestedPath)).toBeTruthy();
+  expect(fs.readFileSync(nestedPath).toString()).toBe('Hello world');
+  const error = await download.path().catch(e => e);
+  expect(error.message).toContain('Path is not available when using browserType.connect(). Use download.saveAs() to save a local copy.');
+  await browser.close();
+});
+
 it('should error when saving with downloads disabled', async({tmpDir, browser, server}) => {
   const page = await browser.newPage({ acceptDownloads: false });
   await page.setContent(`<a href="${server.PREFIX}/download">download</a>`);
@@ -170,6 +188,21 @@ it('should error when saving after deletion', async({tmpDir, browser, server}) =
   await page.close();
 });
 
+it.skip(options.WIRE)('should error when saving after deletion when connected remotely', async({tmpDir, server, browserType, remoteServer}) => {
+  const browser = await browserType.connect({ wsEndpoint: remoteServer.wsEndpoint() });
+  const page = await browser.newPage({ acceptDownloads: true });
+  await page.setContent(`<a href="${server.PREFIX}/download">download</a>`);
+  const [ download ] = await Promise.all([
+    page.waitForEvent('download'),
+    page.click('a')
+  ]);
+  const userPath = path.join(tmpDir, "download.txt");
+  await download.delete();
+  const { message } = await download.saveAs(userPath).catch(e => e);
+  expect(message).toContain('Download already deleted. Save before deleting.');
+  await browser.close();
+});
+
 it('should report non-navigation downloads', async({browser, server}) => {
   // Mac WebKit embedder does not download in this case, although Safari does.
   server.setRoute('/download', (req, res) => {

test/remoteServer.fixture.ts

@@ -109,7 +109,7 @@ class RemoteServer {
     return await this._exitPromise;
   }
 
-  async _close() {
+  async close() {
     if (!this._didExit)
       this._child.kill();
     return await this.childExitCode();
@@ -120,12 +120,12 @@ registerFixture('remoteServer', async ({browserType, defaultBrowserOptions}, tes
   const remoteServer = new RemoteServer();
   await remoteServer._start(browserType, defaultBrowserOptions);
   await test(remoteServer);
-  await remoteServer._close();
+  await remoteServer.close();
 });
 
 registerFixture('stallingRemoteServer', async ({browserType, defaultBrowserOptions}, test) => {
   const remoteServer = new RemoteServer();
   await remoteServer._start(browserType, defaultBrowserOptions, { stallOnClose: true });
   await test(remoteServer);
-  await remoteServer._close();
+  await remoteServer.close();
 });