feat: add brew app connection

- initialize tcp communication with brew app
- WIP value editor sync

Signed-off-by: pakintada@gmail.com <Pakin>

src/lib/core/adb/adb.ts

@@ -1,14 +1,21 @@
 import { Adb, AdbDaemonTransport, encodeUtf8 } from '@yume-chan/adb';
 import AdbWebCredentialStore from '@yume-chan/adb-credential-web';
 import {
+	AdbDaemonWebUsbDevice,
 	AdbDaemonWebUsbDeviceManager,
-	AdbDaemonWebUsbDeviceObserver,
-	type AdbDaemonWebUsbDevice
+	AdbDaemonWebUsbDeviceObserver
 } from '@yume-chan/adb-daemon-webusb';
 import { AdbInstance } from '../../../routes/state.svelte';
 import { deviceCredentialManager } from './deviceCredManager';
 import { Consumable, MaybeConsumable, ReadableStream } from '@yume-chan/stream-extra';
 import { AdbScrcpyClient } from '@yume-chan/adb-scrcpy';
+import { addNotification } from '../stores/noti';
+import { handleAdbPayload } from '../handlers/adbPayloadHandler';
+import { adbWriter } from '../stores/adbWriter';
+import { WritableStream } from '@yume-chan/stream-extra';
+import { env } from '$env/dynamic/public';
+
+let syncConnection: any = null;
 
 export async function connnectViaWebUSB() {
 	const device = await AdbDaemonWebUsbDeviceManager.BROWSER?.requestDevice();
@@ -27,13 +34,21 @@ export async function connnectViaWebUSB() {
 			});
 
 			const adb = new Adb(transport);
-			saveAdbInstance(adb);
+			await saveAdbInstance(adb);
+			await connectToAndroidServer();
 
 			// save device info
 			await deviceCredentialManager.saveDeviceInfo(device);
 		} catch (e: any) {
 			console.error('error on connect', e);
-			throw new Error(e.toString());
+
+			if (e instanceof AdbDaemonWebUsbDevice.DeviceBusyError) {
+				addNotification(
+					'ERR:Device is already in use by another program, please close the program and try again'
+				);
+			}
+
+			throw e;
 		}
 	}
 }
@@ -51,7 +66,9 @@ export async function connectDeviceByCred(
 		});
 
 		const adb = new Adb(transport);
-		saveAdbInstance(adb);
+
+		await saveAdbInstance(adb);
+		await connectToAndroidServer();
 
 		return true;
 	} catch (error) {
@@ -59,7 +76,8 @@ export async function connectDeviceByCred(
 	}
 }
 
-export function saveAdbInstance(adb: Adb | undefined) {
+export async function saveAdbInstance(adb: Adb | undefined) {
+	await cleanupSync();
 	AdbInstance.instance = adb;
 }
 
@@ -100,11 +118,12 @@ export async function executeCmd(command: string) {
 			};
 		}
 	} catch (e: any) {
-		console.log(e.message);
+		// console.log(e.message);
 		//ExactReadable ended
 		if (e.message.includes('ExactReadable ended')) {
-			console.error('connection cut off');
 			return {
+				output: '',
+				exitCode: 1,
 				error: 'ExactReadableEndedError'
 			};
 		}
@@ -119,32 +138,49 @@ export async function disconnect() {
 	if (instance) {
 		await instance.close();
 		console.log('close instance');
-		saveAdbInstance(undefined);
+		await saveAdbInstance(undefined);
 	}
 }
 
-export async function pull(filename: string) {
-	let instance = getAdbInstance();
-	if (instance) {
-		let chunkList: Uint8Array<ArrayBufferLike>[] = [];
-		let sync = await instance.sync();
-		const content = sync.read(filename);
-		let result = content.values();
-		let res;
-
-		let result_string = '';
-
-		while ((res = await result.next()) != null) {
-			// console.log(res.value);
-			if (res.value != undefined) {
-				result_string += new TextDecoder().decode(res.value);
-			}
-			if (res.done) {
-				break;
-			}
+export async function cleanupSync() {
+	if (syncConnection) {
+		try {
+			await syncConnection.dispose();
+		} catch (e) {
+			console.error('error on dispose sync', e);
 		}
+	}
 
-		return result_string;
+	syncConnection = null;
+}
+
+export async function pull(filename: string, timeoutMs: number = 5000) {
+	let instance = getAdbInstance();
+
+	await cleanupSync();
+
+	try {
+		if (instance) {
+			let chunkList: Uint8Array<ArrayBufferLike>[] = [];
+			const syncProm = instance.sync();
+			const timeoutProm = new Promise<never>((_, reject) => {
+				setTimeout(() => reject(new Error('sync timeout')), timeoutMs);
+			});
+
+			syncConnection = await Promise.race([syncProm, timeoutProm]);
+			const content = syncConnection.read(filename);
+			let result_string = '';
+
+			for await (const chunk of content) {
+				result_string += new TextDecoder().decode(chunk);
+			}
+
+			return result_string;
+		}
+	} catch (pull_error: any) {
+		console.log('pulling error', pull_error);
+	} finally {
+		await cleanupSync();
 	}
 }
 
@@ -176,6 +212,38 @@ export async function push(path: string, obj: string) {
 	}
 }
 
+// NOTE: adb reverse is not work by unavailable features support
+async function connectToAndroidServer() {
+	try {
+		let inst = getAdbInstance();
+		if (!inst) {
+			console.warn('adb instance not found');
+			return;
+		}
+		const stream = await inst.transport.connect(env.PUBLIC_BREW_CONN_PORT);
+		const writer = stream.writable.getWriter();
+		const reader = stream.readable.getReader();
+
+		adbWriter.set(writer);
+
+		(async () => {
+			try {
+				while (true) {
+					const { value, done } = await reader.read();
+					if (done) break;
+					handleAdbPayload(new TextDecoder().decode(value));
+				}
+			} catch (e) {
+				console.error('read error', e);
+			} finally {
+				adbWriter.set(null);
+			}
+		})();
+	} catch (err) {
+		console.error('Connection failed. Suspect java running or not', err);
+	}
+}
+
 // logcat stream
 
 // TODO: screen mirror

src/lib/core/brew/command.ts

@@ -0,0 +1,46 @@
+import { env } from '$env/dynamic/public';
+import * as adb from '$lib/core/adb/adb';
+
+class BrewCommandError extends Error {
+	public readonly field?: string;
+
+	constructor(message: string, field?: string) {
+		super(message);
+		this.name = 'BrewCommandError';
+		this.field = field;
+	}
+}
+
+/// Send command to brew app
+/// NOTE: app must enable flag `enable_adb_block_watch`
+async function sendCommand(type: string, params?: string[]) {
+	// check instance
+	let inst = adb.getAdbInstance();
+	if (inst) {
+		try {
+			let cmd = type + ' ' + (params?.join(' ') ?? '');
+			await adb.push(env.PUBLIC_BREW_CMD_WEB, cmd);
+		} catch (e) {
+			throw new BrewCommandError('Command failed', `${e}`);
+		}
+	} else {
+		throw new BrewCommandError('Instance lost');
+	}
+}
+
+async function sendReset() {
+	let inst = adb.getAdbInstance();
+	if (inst) {
+		try {
+			await adb.push(env.PUBLIC_BREW_CMD_WEB, '');
+			await adb.push(env.PUBLIC_BREW_CURRENT_RECIPE, '');
+			await adb.push(env.PUBLIC_BREW_WEB_STATUS, '');
+		} catch (e) {
+			throw new BrewCommandError('Reset failed', `${e}`);
+		}
+	} else {
+		throw new BrewCommandError('Instance lost');
+	}
+}
+
+export { sendCommand, sendReset, BrewCommandError };

src/lib/core/handlers/adbPayloadHandler.ts

@@ -0,0 +1,41 @@
+import { addNotification } from '../stores/noti';
+
+type AdbPayload = { type: string; payload: any };
+
+async function handleAdbPayload(raw_payload: string) {
+	console.log('get payload', raw_payload);
+	try {
+		const payload: AdbPayload = JSON.parse(raw_payload);
+		switch (payload.type) {
+			case 'log':
+				let log_level = payload.payload['level'] ?? 'INFO';
+				let log_message = payload.payload['msg'] ?? '';
+
+				if (log_message !== '') addNotification(`${log_level}`);
+				break;
+			case 'response':
+				if (payload.payload instanceof String) {
+					// single message response
+				}
+				break;
+			case 'ACK':
+				// acknowledge response from app
+				if (payload.payload !== 'OK') {
+					// abnormal
+					console.error('error from ACK', payload.payload);
+					addNotification('ERR:Request rejected');
+				}
+				break;
+			case 'error':
+				// show error to user from brew app
+				addNotification(`ERR:${payload.payload}`);
+				// send message to server if needed
+				break;
+			default:
+		}
+	} catch (error: any) {
+		// invalid format
+	}
+}
+
+export { handleAdbPayload };

src/lib/core/stores/adbWriter.ts

@@ -0,0 +1,28 @@
+import { get, writable } from 'svelte/store';
+import { addNotification } from './noti';
+
+const adbWriter: any = writable(null);
+
+async function sendToAndroid(message: any) {
+	let writer: any = get(adbWriter);
+	console.log('writer', writer);
+	if (!writer) {
+		addNotification('ERR:No active connection');
+		return;
+	}
+	try {
+		const encoder = new TextEncoder();
+		console.log(writer);
+		await writer.write(encoder.encode(JSON.stringify(message) + '\n'));
+		console.log('sent!');
+	} catch (error) {
+		console.error('write failed', error);
+	}
+}
+
+// helper function for checking if connection is ok
+function isAdbWriterAvailable() {
+	return get(adbWriter) != null;
+}
+
+export { sendToAndroid, adbWriter, isAdbWriterAvailable };