Merge branch 'dev' into sync_batch_upload

packages/lib/Synchronizer.ts

@@ -19,6 +19,7 @@ import EncryptionService from './services/EncryptionService';
 import JoplinError from './JoplinError';
 import ShareService from './services/share/ShareService';
 import TaskQueue from './TaskQueue';
+import { preUploadItems, serializeAndUploadItem } from './services/synchronizer/uploadUtils';
 const { sprintf } = require('sprintf-js');
 const { Dirnames } = require('./services/synchronizer/utils/types');
 
@@ -389,6 +390,12 @@ export default class Synchronizer {
 		// correctly so as to share/unshare the right items.
 		await Folder.updateAllShareIds();
 
+		const uploadQueue = new TaskQueue('syncUpload', this.logger());
+
+		const uploadItem = (path: string, content: any) => {
+			return this.apiCall('put', path, content);
+		};
+
 		let errorToThrow = null;
 		let syncLock = null;
 
@@ -440,6 +447,8 @@ export default class Synchronizer {
 					const result = await BaseItem.itemsThatNeedSync(syncTargetId);
 					const locals = result.items;
 
+					await preUploadItems(uploadItem, uploadQueue, result.items.filter((it: any) => result.neverSyncedItemIds.includes(it.id)));
+
 					for (let i = 0; i < locals.length; i++) {
 						if (this.cancelling()) break;
 
@@ -588,8 +597,7 @@ export default class Synchronizer {
 							let canSync = true;
 							try {
 								if (this.testingHooks_.indexOf('notesRejectedByTarget') >= 0 && local.type_ === BaseModel.TYPE_NOTE) throw new JoplinError('Testing rejectedByTarget', 'rejectedByTarget');
-								const content = await ItemClass.serializeForSync(local);
-								await this.apiCall('put', path, content);
+								await serializeAndUploadItem(uploadItem, uploadQueue, ItemClass, path, local);
 							} catch (error) {
 								if (error && error.code === 'rejectedByTarget') {
 									await handleCannotSyncItem(ItemClass, syncTargetId, local, error.message);

packages/lib/services/synchronizer/uploadUtils.ts

@@ -0,0 +1,30 @@
+import { ModelType } from '../../BaseModel';
+import BaseItem, { ItemThatNeedSync } from '../../models/BaseItem';
+import TaskQueue from '../../TaskQueue';
+
+type UploadItem = (path: string, content: any)=> Promise<any>;
+
+export async function serializeAndUploadItem(uploadItem: UploadItem, uploadQueue: TaskQueue, ItemClass: any, path: string, local: ItemThatNeedSync) {
+	if (uploadQueue && uploadQueue.taskExists(path)) {
+		return uploadQueue.taskResult(path);
+	}
+
+	const content = await ItemClass.serializeForSync(local);
+	return uploadItem(path, content);
+}
+
+export async function preUploadItems(uploadItem: UploadItem, uploadQueue: TaskQueue, items: ItemThatNeedSync[]) {
+	for (const local of items) {
+		// For resources, additional logic is necessary - in particular the blob
+		// should be uploaded before the metadata, so we can't batch process.
+		if (local.type_ === ModelType.Resource) continue;
+
+		const ItemClass = BaseItem.itemClass(local);
+		const path = BaseItem.systemPath(local);
+		uploadQueue.push(path, async () => {
+			await serializeAndUploadItem(uploadItem, null, ItemClass, path, local);
+		});
+	}
+
+	await uploadQueue.waitForAll();
+}