android-livesync-tool.ts

import * as net from "net";
import * as path from "path";
import * as crypto from "crypto";

const PROTOCOL_VERSION_LENGTH_SIZE = 1;
const PROTOCOL_OPERATION_LENGTH_SIZE = 1;
const SIZE_BYTE_LENGTH = 1;
const DELETE_FILE_OPERATION = 7;
const CREATE_FILE_OPERATION = 8;
const DO_SYNC_OPERATION = 9;
const ERROR_REPORT = 1;
const OPERATION_END_REPORT = 2;
const OPERATION_END_NO_REFRESH_REPORT_CODE = 3;
const REPORT_LENGTH = 1;
const DO_REFRESH_LENGTH = 1;
const DO_REFRESH = 1;
const SKIP_REFRESH = 0;
const SYNC_OPERATION_TIMEOUT = 60000;
const TRY_CONNECT_TIMEOUT = 30000;
const DEFAULT_LOCAL_HOST_ADDRESS = "127.0.0.1";

export class AndroidLivesyncTool implements IAndroidLivesyncTool {
	private operationPromises: IDictionary<any>;
	private socketError: string | Error;
	private socketConnection: IDuplexSocket;
	private configuration: IAndroidLivesyncToolConfiguration;
	private pendingConnectionData: {
		connectionTimer?: NodeJS.Timer,
		socketTimer?: NodeJS.Timer,
		rejectHandler?: Function,
		socket?: IDuplexSocket
	} = null;

	constructor(private $androidProcessService: Mobile.IAndroidProcessService,
		private $errors: IErrors,
		private $fs: IFileSystem,
		private $logger: ILogger,
		private $mobileHelper: Mobile.IMobileHelper,
		private $processService: IProcessService) {
			this.operationPromises = Object.create(null);
			this.socketError = null;
			this.socketConnection = null;
			this.$processService.attachToProcessExitSignals(this, this.dispose);
	}

	public async connect(configuration: IAndroidLivesyncToolConfiguration): Promise<void> {
		if (!configuration.appIdentifier) {
			this.$errors.fail(`You need to provide "appIdentifier" as a configuration property!`);
		}

		if (!configuration.appPlatformsPath) {
			this.$errors.fail(`You need to provide "baseDir" as a configuration property!`);
		}

		if (this.socketConnection) {
			this.$errors.fail("Socket connection already exists.");
		}

		if (!configuration.localHostAddress) {
			configuration.localHostAddress = DEFAULT_LOCAL_HOST_ADDRESS;
		}

		this.configuration = configuration;
		this.socketError = null;

		const port = await this.$androidProcessService.forwardFreeTcpToAbstractPort({
			deviceIdentifier: configuration.deviceIdentifier,
			appIdentifier: configuration.appIdentifier,
			abstractPort: `localabstract:${configuration.appIdentifier}-livesync`
		});

		const connectionResult = await this.connectEventuallyUntilTimeout(this.createSocket.bind(this, port), TRY_CONNECT_TIMEOUT);
		this.handleConnection(connectionResult);
	}

	public async sendFile(filePath: string): Promise<void> {
		await this.sendFileHeader(filePath);
		await this.sendFileContent(filePath);
	}

	public async sendFiles(filePaths: string[]) {
		for (const filePath of filePaths) {
			if (this.$fs.getLsStats(filePath).isFile()) {
				if (!this.$fs.exists(filePath)) {
					this.$errors.fail(`${filePath} doesn't exist.`);
				}

				await this.sendFile(filePath);
			}
		}
	}

	public sendDirectory(directoryPath: string) {
		const list = this.$fs.enumerateFilesInDirectorySync(directoryPath);
		return this.sendFiles(list);
	}

	public removeFile(filePath: string): Promise<boolean> {
		return new Promise((resolve: Function, reject: Function) => {
			this.verifyActiveConnection(reject);
			const filePathData = this.getFilePathData(filePath);
			const headerBuffer = Buffer.alloc(PROTOCOL_OPERATION_LENGTH_SIZE +
				SIZE_BYTE_LENGTH +
				filePathData.filePathLengthSize +
				filePathData.filePathLengthBytes);

			let offset = 0;
			offset += headerBuffer.write(DELETE_FILE_OPERATION.toString(), offset, PROTOCOL_OPERATION_LENGTH_SIZE);
			offset = headerBuffer.writeInt8(filePathData.filePathLengthSize, offset);
			offset += headerBuffer.write(filePathData.filePathLengthString, offset, filePathData.filePathLengthSize);
			headerBuffer.write(filePathData.relativeFilePath, offset, filePathData.filePathLengthBytes);
			const hash = crypto.createHash("md5").update(headerBuffer).digest();

			this.socketConnection.write(headerBuffer);
			this.socketConnection.write(hash, () => {
				resolve(true);
			});
		});
	}

	public removeFiles(files: string[]) {
		return Promise.all(files.map(file => this.removeFile(file)));
	}

	public generateOperationIdentifier(): string {
		return crypto.randomBytes(16).toString("hex");
	}

	public isOperationInProgress(operationId: string): boolean {
		return !!this.operationPromises[operationId];
	}

	public sendDoSyncOperation(doRefresh = true , timeout?: number, operationId?: string): Promise<IAndroidLivesyncSyncOperationResult> {
		const id = operationId || this.generateOperationIdentifier();
		const operationPromise: Promise<IAndroidLivesyncSyncOperationResult> = new Promise((resolve: Function, reject: Function) => {
			this.verifyActiveConnection(reject);
			const message = `${DO_SYNC_OPERATION}${id}`;
			const headerBuffer = Buffer.alloc(Buffer.byteLength(message) + DO_REFRESH_LENGTH);
			const socketId = this.socketConnection.uid;
			const doRefreshCode = doRefresh ? DO_REFRESH : SKIP_REFRESH;
			const offset = headerBuffer.write(message);

			headerBuffer.writeUInt8(doRefreshCode, offset);
			const hash = crypto.createHash("md5").update(headerBuffer).digest();

			this.socketConnection.write(headerBuffer);
			this.socketConnection.write(hash);

			timeout = timeout || SYNC_OPERATION_TIMEOUT;
			const timeoutId = setTimeout(() => {
				if (this.isOperationInProgress(id)) {
					this.handleSocketError(socketId, "Sync operation is taking too long");
				}
			}, timeout);

			this.operationPromises[id] = {
				resolve,
				reject,
				socketId,
				timeoutId
			};
		});

		return operationPromise;
	}

	public end(error?: Error) {
		if (this.socketConnection) {
			const socketUid = this.socketConnection.uid;
			const socket = this.socketConnection;
			error = error || this.getErrorWithMessage("Socket connection ended before sync operation is complete.");
			//remove listeners and delete this.socketConnection
			this.cleanState(socketUid);
			//call end of the connection (close and error callbacks won't be called - listeners removed)
			socket.end();
			//reject all pending sync requests and clear timeouts
			this.rejectPendingSyncOperations(socketUid, error);
		}
	}

	private sendFileHeader(filePath: string): Promise<void> {
		return new Promise((resolve, reject) => {
			let error;
			this.verifyActiveConnection(reject);
			const filePathData = this.getFilePathData(filePath);
			const stats = this.$fs.getFsStats(filePathData.filePath);
			const fileContentLengthBytes = stats.size;
			const fileContentLengthString = fileContentLengthBytes.toString();
			const fileContentLengthSize = Buffer.byteLength(fileContentLengthString);
			const headerBuffer = Buffer.alloc(PROTOCOL_OPERATION_LENGTH_SIZE +
					SIZE_BYTE_LENGTH +
					filePathData.filePathLengthSize +
					filePathData.filePathLengthBytes +
					SIZE_BYTE_LENGTH +
					fileContentLengthSize);

			if (filePathData.filePathLengthSize > 255) {
				error = this.getErrorWithMessage("File name size is longer that 255 digits.");
			} else if (fileContentLengthSize > 255) {
				error = this.getErrorWithMessage("File name size is longer that 255 digits.");
			}

			if (error) {
				reject(error);
			}

			let offset = 0;
			offset += headerBuffer.write(CREATE_FILE_OPERATION.toString(), offset, PROTOCOL_OPERATION_LENGTH_SIZE);
			offset = headerBuffer.writeUInt8(filePathData.filePathLengthSize, offset);
			offset += headerBuffer.write(filePathData.filePathLengthString, offset, filePathData.filePathLengthSize);
			offset += headerBuffer.write(filePathData.relativeFilePath, offset, filePathData.filePathLengthBytes);
			offset = headerBuffer.writeUInt8(fileContentLengthSize, offset);
			headerBuffer.write(fileContentLengthString, offset, fileContentLengthSize);
			const hash = crypto.createHash("md5").update(headerBuffer).digest();

			this.socketConnection.write(headerBuffer);
			this.socketConnection.write(hash);
			resolve();
		});
	}

	private sendFileContent(filePath: string): Promise<boolean> {
		return new Promise((resolve, reject) => {
			this.verifyActiveConnection(reject);
			const fileStream = this.$fs.createReadStream(filePath);
			const fileHash = crypto.createHash("md5");

			fileStream
				.on("data", (chunk: string | Buffer) => {
					fileHash.update(chunk);
					if (this.socketConnection) {
						this.socketConnection.write(chunk);
					} else {
						const error = this.checkConnectionStatus();
						//TODO Destroy method added in node 8.0.0.
						//when we deprecate node 6.x uncomment the line below
						//fileStream.destroy(error);
						reject(error);
					}
				})
				.on("end", () => {
					if (this.socketConnection) {
						this.socketConnection.write(fileHash.digest(), () => {
							resolve(true);
						});
					} else {
						const error = this.checkConnectionStatus();
						reject(error);
					}
				})
				.on("error", (error: Error) => {
					reject(error);
				});
		});
	}

	private createSocket(port: number): IDuplexSocket {
		const socket = new net.Socket();
		socket.connect(port, this.configuration.localHostAddress);
		return socket;
	}

	private checkConnectionStatus() {
		if (this.socketConnection === null) {
			const defaultError = this.getErrorWithMessage("No socket connection available.");
			const error = this.socketError || defaultError;

			return error;
		}
	}

	private verifyActiveConnection(rejectHandler?: any) {
		const error = this.checkConnectionStatus();
		if (error && rejectHandler) {
			rejectHandler(error);
		}

		if (error && !rejectHandler) {
			this.$errors.failWithoutHelp(error.toString());
		}
	}

	private handleConnection({ socket, data }: { socket: IDuplexSocket, data: NodeBuffer | string }) {
		this.socketConnection = socket;
		this.socketConnection.uid = this.generateOperationIdentifier();

		const versionLength = (<NodeBuffer>data).readUInt8(0);
		const versionBuffer = data.slice(PROTOCOL_VERSION_LENGTH_SIZE, versionLength + PROTOCOL_VERSION_LENGTH_SIZE);
		const appIdentifierBuffer = data.slice(versionLength + PROTOCOL_VERSION_LENGTH_SIZE, data.length);

		const protocolVersion = versionBuffer.toString();
		const appIdentifier = appIdentifierBuffer.toString();
		this.$logger.trace(`Handle socket connection for app identifier: ${appIdentifier} with protocol version: ${protocolVersion}.`);

		this.socketConnection.on("data", (connectionData: NodeBuffer) => this.handleData(socket.uid, connectionData));
		this.socketConnection.on("close", (hasError: boolean) => this.handleSocketClose(socket.uid, hasError));
		this.socketConnection.on("error", (err: Error) => {
			const error = new Error(`Socket Error:\n${err}`);
			if (this.configuration.errorHandler) {
				this.configuration.errorHandler(error);
			} else {
				this.handleSocketError(socket.uid, error.message);
			}
		});
	}

	private connectEventuallyUntilTimeout(factory: () => IDuplexSocket, timeout: number): Promise<{socket: IDuplexSocket, data: NodeBuffer | string}> {
		return new Promise((resolve, reject) => {
			let lastKnownError: Error | string,
				isConnected = false;

			const connectionTimer = setTimeout(() => {
				if (!isConnected) {
					isConnected = true;
					reject(lastKnownError || { message: "Socket connection timeouted." });
					this.pendingConnectionData = null;
				}
			}, timeout);

			this.pendingConnectionData = {
				connectionTimer,
				rejectHandler: reject
			};

			const tryConnect = () => {
				const socket = factory();

				const tryConnectAfterTimeout = (error: Error) => {
					if (isConnected) {
						this.pendingConnectionData = null;
						return;
					}

					if (typeof (error) === "boolean" && error) {
						error = new Error("Socket closed due to error");
					}

					lastKnownError = error;
					socket.removeAllListeners();
					this.pendingConnectionData.socketTimer = setTimeout(tryConnect, 1000);
				};

				this.pendingConnectionData.socket = socket;

				socket.once("data", data => {
					socket.removeListener("close", tryConnectAfterTimeout);
					socket.removeListener("error", tryConnectAfterTimeout);
					isConnected = true;
					resolve({ socket, data });
				});
				socket.on("close", tryConnectAfterTimeout);
				socket.on("error", tryConnectAfterTimeout);
			};

			tryConnect();
		});
	}

	private handleData(socketId: string, data: any) {
		const reportType = data.readUInt8();
		const infoBuffer = data.slice(REPORT_LENGTH, data.length);

		if (reportType === ERROR_REPORT) {
			const errorMessage = infoBuffer.toString();
			this.handleSocketError(socketId, errorMessage);
		} else if (reportType === OPERATION_END_REPORT) {
			this.handleSyncEnd({data:infoBuffer, didRefresh: true});
		} else if (reportType === OPERATION_END_NO_REFRESH_REPORT_CODE) {
			this.handleSyncEnd({data:infoBuffer, didRefresh: false});
		}
	}

	private handleSyncEnd({data, didRefresh}: {data: any, didRefresh: boolean}) {
		const operationId = data.toString();
		const promiseHandler = this.operationPromises[operationId];

		if (promiseHandler) {
			clearTimeout(promiseHandler.timeoutId);
			promiseHandler.resolve({operationId, didRefresh});
			delete this.operationPromises[operationId];
		}
	}

	private handleSocketClose(socketId: string, hasError: boolean) {
		const errorMessage = "Socket closed from server before operation end.";
		this.handleSocketError(socketId, errorMessage);
	}

	private handleSocketError(socketId: string, errorMessage: string) {
		const error = this.getErrorWithMessage(errorMessage);
		if (this.socketConnection && this.socketConnection.uid === socketId) {
			this.socketError = error;
			this.end(error);
		} else {
			this.rejectPendingSyncOperations(socketId, error);
		}
	}

	private cleanState(socketId: string) {
		if (this.socketConnection && this.socketConnection.uid === socketId) {
			this.socketConnection.removeAllListeners();
			this.socketConnection = null;
		}
	}

	private rejectPendingSyncOperations(socketId: string, error: Error) {
		_.keys(this.operationPromises)
			.forEach(operationId => {
				const operationPromise = this.operationPromises[operationId];
				if (operationPromise.socketId === socketId) {
					clearTimeout(operationPromise.timeoutId);
					operationPromise.reject(error);
					delete this.operationPromises[operationId];
				}
			});
	}

	private getErrorWithMessage(errorMessage: string) {
		const error = new Error(errorMessage);
		error.message = errorMessage;

		return error;
	}

	private getFilePathData(filePath: string): { relativeFilePath: string, filePathLengthBytes: number, filePathLengthString: string, filePathLengthSize: number, filePath: string } {
		const relativeFilePath = this.resolveRelativePath(filePath);
		const filePathLengthBytes = Buffer.byteLength(relativeFilePath);
		const filePathLengthString = filePathLengthBytes.toString();
		const filePathLengthSize = Buffer.byteLength(filePathLengthString);

		return {
			relativeFilePath,
			filePathLengthBytes,
			filePathLengthString,
			filePathLengthSize,
			filePath
		};
	}

	private resolveRelativePath(filePath: string): string {
		const relativeFilePath = path.relative(this.configuration.appPlatformsPath, filePath);

		return this.$mobileHelper.buildDevicePath(relativeFilePath);
	}

	private dispose(): void {
		if (this.pendingConnectionData) {
			clearTimeout(this.pendingConnectionData.connectionTimer);

			if (this.pendingConnectionData.socketTimer) {
				clearTimeout(this.pendingConnectionData.socketTimer);
			}
			if (this.pendingConnectionData.socket) {
				this.pendingConnectionData.socket.removeAllListeners();
			}

			this.pendingConnectionData.rejectHandler("LiveSync aborted.");
		}
		this.end();

		_.keys(this.operationPromises)
			.forEach(operationId => {
				const operationPromise = this.operationPromises[operationId];
				clearTimeout(operationPromise.timeoutId);
			});
	}
}
$injector.register("androidLivesyncTool", AndroidLivesyncTool);