fix: Update error handling in push/pull file methods for real devices (…

…appium#1140)

lib/commands/file-movement.js

@@ -14,8 +14,9 @@ const CONTAINER_PATH_PATTERN = new RegExp(`^${CONTAINER_PATH_MARKER}([^/]+)/(.*)
 const CONTAINER_TYPE_SEPARATOR = ':';
 const IFUSE_CONTAINER_DOCUMENTS = 'documents';
 const CONTAINER_DOCUMENTS_PATH = 'Documents';
-const IO_TIMEOUT = 30000;
+const IO_TIMEOUT = 60000;
 const OBJECT_NOT_FOUND_ERROR_MESSAGE = 'OBJECT_NOT_FOUND';
+const MAX_PULL_CHUNK_SIZE = 5;
 
 let commands = iosCommands.file;
 
@@ -73,11 +74,14 @@ async function createService (udid, remotePath) {
 
 async function pullFileFromRealDevice (service, relativePath) {
   const stream = await service.createReadStream(relativePath, { autoDestroy: true });
-  const closeEvent = new B((resolve) => stream.on('close', resolve)).timeout(IO_TIMEOUT);
+  const pullPromise = new B((resolve, reject) => {
+    stream.on('close', resolve);
+    stream.on('error', reject);
+  });
   const buffer = [];
   stream.on('data', (data) => buffer.push(data));
   try {
-    await closeEvent;
+    await pullPromise.timeout(IO_TIMEOUT);
   } catch (e) {
     throw new Error(`Couldn't pull the file '${relativePath}' ` +
       `within the given timeout ${IO_TIMEOUT}ms. Original error: ${e.message}`);
@@ -90,24 +94,44 @@ async function pullFolderFromRealDevice (service, relativePath) {
   try {
     const folderPath = path.join(tmpFolder, relativePath);
     await mkdirp(folderPath);
-    const promises = [];
+    const pullPromises = [];
+    const waitForPullChunks = async () => {
+      if (_.isEmpty(pullPromises)) {
+        return;
+      }
+
+      try {
+        await B.all(pullPromises).timeout(IO_TIMEOUT);
+      } catch (e) {
+        throw new Error(`Couldn't pull all items in the folder '${relativePath}' ` +
+          `within the given timeout ${IO_TIMEOUT}ms. Original error: ${e.message}`);
+      }
+    };
     await service.walkDir(relativePath, true, async (itemPath, isDir) => {
       const pathOnServer = path.join(tmpFolder, itemPath);
       if (isDir) {
         await fs.mkdir(pathOnServer);
-      } else {
-        const readStream = await service.createReadStream(itemPath, {autoDestroy: true });
-        const writeStream = fs.createWriteStream(pathOnServer, {autoClose: true});
-        promises.push(new B((resolve) => writeStream.on('close', resolve)));
-        readStream.pipe(writeStream);
+        return;
+      }
+
+      const readStream = await service.createReadStream(itemPath, {autoDestroy: true});
+      const writeStream = fs.createWriteStream(pathOnServer, {autoClose: true});
+      pullPromises.push(new B((resolve, reject) => {
+        writeStream.on('close', resolve);
+        const onStreamingError = (e) => {
+          readStream.unpipe(writeStream);
+          reject(e);
+        };
+        writeStream.on('error', onStreamingError);
+        readStream.on('error', onStreamingError);
+      }));
+      readStream.pipe(writeStream);
+      if (pullPromises.length % MAX_PULL_CHUNK_SIZE === 0) {
+        await waitForPullChunks();
       }
     });
-    try {
-      await B.all(promises).timeout(IO_TIMEOUT);
-    } catch (e) {
-      throw new Error(`Couldn't pull all items in the folder '${relativePath}' ` +
-        `within the given timeout ${IO_TIMEOUT}ms. Original error: ${e.message}`);
-    }
+    // Wait for the rest of the chunks
+    await waitForPullChunks();
     return Buffer.from(await zip.toInMemoryZip(folderPath)).toString('base64');
   } finally {
     await fs.rimraf(tmpFolder);
@@ -229,12 +253,15 @@ async function pushFileToRealDevice (device, remotePath, base64Data) {
   const {service, relativePath} = await createService(device.udid, remotePath);
   try {
     await mkdirpDevice(service, path.dirname(relativePath));
-    const stream = await service.createWriteStream(relativePath, { autoClose: true });
+    const stream = await service.createWriteStream(relativePath, {autoDestroy: true});
+    const pushPromise = new B((resolve, reject) => {
+      stream.on('error', reject);
+      stream.on('close', resolve);
+    });
     stream.write(Buffer.from(base64Data, 'base64'));
-    const closeEvent = new B((resolve) => stream.on('close', resolve)).timeout(IO_TIMEOUT);
-    stream.destroy();
+    stream.end();
     try {
-      await closeEvent;
+      await pushPromise.timeout(IO_TIMEOUT);
     } catch (e) {
       throw new Error(`Could not push the file within the given timeout ${IO_TIMEOUT}ms. ` +
         `Original error: ${e.message}`);