improve serialization of BLE writes

src/actions/ble.ts

@@ -2,6 +2,8 @@
 // Copyright (c) 2020 The Pybricks Authors
 
 import { Action } from 'redux';
+import { assert } from '../utils';
+import { createCountFunc } from '../utils/iter';
 
 /**
  * Bluetooth low energy connection action types.
@@ -62,19 +64,48 @@ export enum BLEDataActionType {
     /**
      * Write data.
      */
-    Write = 'ble.data.write',
+    Write = 'ble.data.action.write',
+    /**
+     * Writing completed successfully.
+     */
+    DidWrite = 'ble.data.didWrite',
+    /**
+     * Writing failed.
+     */
+    DidFailToWrite = 'ble.data.action.didFailToWrite',
     /**
      * Notify that data was received.
      */
-    Notify = 'ble.data.receive',
+    Notify = 'ble.data.action.receive',
 }
 
-export interface BLEDataWriteAction extends Action<BLEDataActionType.Write> {
+const nextId = createCountFunc();
+
+export type BLEDataWriteAction = Action<BLEDataActionType.Write> & {
+    id: number;
     value: Uint8Array;
-}
+};
 
 export function write(value: Uint8Array): BLEDataWriteAction {
-    return { type: BLEDataActionType.Write, value };
+    assert(value.length <= 20, 'value can be at most 20 bytes');
+    return { type: BLEDataActionType.Write, id: nextId(), value };
+}
+
+export type BLEDataDidWriteAction = Action<BLEDataActionType.DidWrite> & {
+    id: number;
+};
+
+export function didWrite(id: number): BLEDataDidWriteAction {
+    return { type: BLEDataActionType.DidWrite, id };
+}
+
+export type BLEDataDidFailToWriteAction = Action<BLEDataActionType.DidFailToWrite> & {
+    id: number;
+    err: Error;
+};
+
+export function didFailToWrite(id: number, err: Error): BLEDataDidFailToWriteAction {
+    return { type: BLEDataActionType.DidFailToWrite, id, err };
 }
 
 export interface BLEDataNotifyAction extends Action<BLEDataActionType.Notify> {
@@ -86,7 +117,11 @@ export function notify(value: DataView): BLEDataNotifyAction {
 }
 
 /** Common type for low-level BLE data actions. */
-export type BLEDataAction = BLEDataWriteAction | BLEDataNotifyAction;
+export type BLEDataAction =
+    | BLEDataWriteAction
+    | BLEDataDidWriteAction
+    | BLEDataDidFailToWriteAction
+    | BLEDataNotifyAction;
 
 /**
  * High-level BLE actions.

src/epics/ble.ts

@@ -4,7 +4,7 @@
 import { AnyAction } from 'redux';
 import { Epic, combineEpics, ofType } from 'redux-observable';
 import { map } from 'rxjs/operators';
-import { BLEDataAction, BLEDataActionType } from '../actions/ble';
+import { BLEDataActionType, BLEDataNotifyAction } from '../actions/ble';
 import { HubRuntimeStatusType, checksum, updateStatus } from '../actions/hub';
 import { sendData } from '../actions/terminal';
 import { RootState } from '../reducers';
@@ -14,7 +14,7 @@ const decoder = new TextDecoder();
 
 const rxUartData: Epic<AnyAction, AnyAction, RootState> = (action$, state$) =>
     action$.pipe(
-        ofType<AnyAction, BLEDataAction>(BLEDataActionType.Notify),
+        ofType<AnyAction, BLEDataNotifyAction>(BLEDataActionType.Notify),
         map((a) => {
             if (
                 state$.value.hub.runtime === HubRuntimeState.Loading &&

src/sagas/terminal.test.ts

@@ -3,7 +3,12 @@
 
 import { END, MulticastChannel, Saga, Task, runSaga, stdChannel } from 'redux-saga';
 import { Action } from '../actions';
-import { BLEDataActionType, BLEDataWriteAction } from '../actions/ble';
+import {
+    BLEDataActionType,
+    BLEDataWriteAction,
+    didFailToWrite,
+    didWrite,
+} from '../actions/ble';
 import {
     TerminalActionType,
     TerminalSetDataSourceAction,
@@ -23,11 +28,19 @@ class AsyncSaga {
         this.takers = [];
         this.channel = stdChannel();
         this.task = runSaga(
-            { channel: this.channel, dispatch: this.dispatch.bind(this) },
+            {
+                channel: this.channel,
+                dispatch: this.dispatch.bind(this),
+                onError: (e) => fail(e),
+            },
             saga,
         );
     }
 
+    public numPending(): number {
+        return this.dispatches.length;
+    }
+
     public put(action: Action): void {
         this.channel.put(action);
     }
@@ -60,11 +73,11 @@ class AsyncSaga {
         this.task.cancel();
         await this.task.toPromise();
         if (this.dispatches.some((x) => x.type !== END.type)) {
-            fail(`unhandled dispatches remain: ${this.dispatches}`);
+            fail(`unhandled dispatches remain: ${JSON.stringify(this.dispatches)}`);
         }
     }
 
-    private dispatch(action: Action | END): void {
+    private dispatch(action: Action | END): Action | END {
         const taker = this.takers.shift();
         if (taker === undefined) {
             // if there are no takers waiting, the queue the action
@@ -73,6 +86,7 @@ class AsyncSaga {
             // otherwise complete the promise
             taker.put(action);
         }
+        return action;
     }
 }
 
@@ -98,20 +112,111 @@ test('Terminal data source responds to send data actions', async () => {
     await saga.end();
 });
 
-test('Terminal data source responds to receive data actions', async () => {
-    const saga = new AsyncSaga(terminal);
+describe('Terminal data source responds to receive data actions', () => {
+    // ASCII/UTF-8 encoding of 'test1234'
+    const expected = new Uint8Array([0x74, 0x65, 0x73, 0x74, 0x31, 0x32, 0x33, 0x34]);
 
-    // set data source is always first action so we have to take it
-    const dataSourceAction = await saga.take();
-    expect(dataSourceAction.type).toBe(TerminalActionType.SetDataSource);
+    test('basic function works', async () => {
+        const saga = new AsyncSaga(terminal);
 
-    saga.put(receiveData('test1234'));
+        // set data source is always first action so we have to take it
+        const dataSourceAction = await saga.take();
+        expect(dataSourceAction.type).toBe(TerminalActionType.SetDataSource);
 
-    const action = await saga.take();
-    expect(action.type).toBe(BLEDataActionType.Write);
-    expect((action as BLEDataWriteAction).value).toEqual(
-        new Uint8Array([0x74, 0x65, 0x73, 0x74, 0x31, 0x32, 0x33, 0x34]),
-    );
+        saga.put(receiveData('test1234'));
 
-    await saga.end();
+        const action = await saga.take();
+        expect(action.type).toBe(BLEDataActionType.Write);
+        expect((action as BLEDataWriteAction).value).toEqual(expected);
+
+        await saga.end();
+    });
+
+    test('messages are queued until previous has completed', async () => {
+        const saga = new AsyncSaga(terminal);
+
+        // set data source is always first action so we have to take it
+        const dataSourceAction = await saga.take();
+        expect(dataSourceAction.type).toBe(TerminalActionType.SetDataSource);
+
+        saga.put(receiveData('test1234'));
+        saga.put(receiveData('test1234'));
+
+        // second message is queued until didWrite or didFailToWrite
+        expect(saga.numPending()).toBe(1);
+
+        const action = await saga.take();
+        expect(action.type).toBe(BLEDataActionType.Write);
+        expect((action as BLEDataWriteAction).value).toEqual(expected);
+
+        // second message is queued until didWrite or didFailToWrite
+        expect(saga.numPending()).toBe(0);
+
+        saga.put(didWrite((action as BLEDataWriteAction).id));
+
+        const action2 = await saga.take();
+        expect(action2.type).toBe(BLEDataActionType.Write);
+        expect((action2 as BLEDataWriteAction).value).toEqual(expected);
+
+        saga.put(didWrite((action2 as BLEDataWriteAction).id));
+
+        await saga.end();
+    });
+
+    test('messages are queued until previous has failed', async () => {
+        const saga = new AsyncSaga(terminal);
+
+        // set data source is always first action so we have to take it
+        const dataSourceAction = await saga.take();
+        expect(dataSourceAction.type).toBe(TerminalActionType.SetDataSource);
+
+        saga.put(receiveData('test1234'));
+        saga.put(receiveData('test1234'));
+
+        // second message is queued until didWrite or didFailToWrite
+        expect(saga.numPending()).toBe(1);
+
+        const action = await saga.take();
+        expect(action.type).toBe(BLEDataActionType.Write);
+        expect((action as BLEDataWriteAction).value).toEqual(expected);
+
+        // second message is queued until didWrite or didFailToWrite
+        expect(saga.numPending()).toBe(0);
+
+        saga.put(
+            didFailToWrite((action as BLEDataWriteAction).id, new Error('test error')),
+        );
+
+        const action2 = await saga.take();
+        expect(action2.type).toBe(BLEDataActionType.Write);
+        expect((action2 as BLEDataWriteAction).value).toEqual(expected);
+
+        saga.put(didWrite((action2 as BLEDataWriteAction).id));
+
+        await saga.end();
+    });
+
+    test('long messages are split', async () => {
+        const saga = new AsyncSaga(terminal);
+
+        // set data source is always first action so we have to take it
+        const dataSourceAction = await saga.take();
+        expect(dataSourceAction.type).toBe(TerminalActionType.SetDataSource);
+
+        saga.put(receiveData('012345678901234567890123456789'));
+
+        const action = await saga.take();
+        expect(action.type).toBe(BLEDataActionType.Write);
+        expect((action as BLEDataWriteAction).value.length).toEqual(20);
+
+        saga.put(didWrite((action as BLEDataWriteAction).id));
+
+        const action2 = await saga.take();
+        expect(action2.type).toBe(BLEDataActionType.Write);
+        expect((action2 as BLEDataWriteAction).value.length).toEqual(10);
+
+        saga.put(didWrite((action2 as BLEDataWriteAction).id));
+
+        await saga.end();
+    });
 });

src/sagas/terminal.ts

@@ -1,22 +1,44 @@
 // SPDX-License-Identifier: MIT
 // Copyright (c) 2020 The Pybricks Authors
 
-import { put, takeEvery } from 'redux-saga/effects';
+import { Channel, buffers } from 'redux-saga';
+import { actionChannel, fork, put, take, takeEvery } from 'redux-saga/effects';
 import PushStream from 'zen-push';
-import { write } from '../actions/ble';
+import { Action } from '../actions';
+import { BLEDataActionType, BLEDataWriteAction, write } from '../actions/ble';
 import {
     TerminalActionType,
     TerminalDataReceiveDataAction,
-    TerminalDataSendDataAction,
     setDataSource,
 } from '../actions/terminal';
 
 const encoder = new TextEncoder();
 const terminalDataSource = new PushStream<string>();
 
-function* receiveTerminalData(action: TerminalDataSendDataAction): Generator {
-    // stdin gets piped to BLE connection
-    yield put(write(encoder.encode(action.value)));
+function* receiveTerminalData(): Generator {
+    const channel = (yield actionChannel(
+        TerminalActionType.ReceivedData,
+        buffers.expanding(),
+    )) as Channel<TerminalDataReceiveDataAction>;
+    while (true) {
+        // wait for input from terminal
+        const action = (yield take(channel)) as TerminalDataReceiveDataAction;
+
+        // stdin gets piped to BLE connection
+        const data = encoder.encode(action.value);
+        for (let i = 0; i < data.length; i += 20) {
+            const { id } = (yield put(
+                write(data.slice(i, i + 20)),
+            )) as BLEDataWriteAction;
+
+            yield take(
+                (a: Action) =>
+                    (a.type === BLEDataActionType.DidWrite ||
+                        a.type === BLEDataActionType.DidFailToWrite) &&
+                    a.id === id,
+            );
+        }
+    }
 }
 
 function sendTerminalData(action: TerminalDataReceiveDataAction): void {
@@ -25,7 +47,7 @@ function sendTerminalData(action: TerminalDataReceiveDataAction): void {
 }
 
 export default function* (): Generator {
-    yield takeEvery(TerminalActionType.ReceivedData, receiveTerminalData);
+    yield fork(receiveTerminalData);
     yield takeEvery(TerminalActionType.SendData, sendTerminalData);
     yield put(setDataSource(terminalDataSource.observable));
 }

src/services/ble.ts

@@ -9,6 +9,8 @@ import {
     connect as connectAction,
     didConnect,
     didDisconnect,
+    didFailToWrite,
+    didWrite,
     disconnect as disconnectAction,
     notify,
 } from '../actions/ble';
@@ -28,7 +30,6 @@ const pybricksServiceUUID = 'c5f50001-8280-46da-89f4-6d8051e4aeef';
 const bleNusServiceUUID = '6e400001-b5a3-f393-e0a9-e50e24dcca9e';
 const bleNusCharRXUUID = '6e400002-b5a3-f393-e0a9-e50e24dcca9e';
 const bleNusCharTXUUID = '6e400003-b5a3-f393-e0a9-e50e24dcca9e';
-const bleNusMaxSize = 20;
 
 let device: BluetoothDevice | undefined;
 let rxChar: PolyfillBluetoothRemoteGATTCharacteristic | undefined;
@@ -116,13 +117,15 @@ function disconnect(action: Action): void {
     device?.gatt?.disconnect();
 }
 
-async function write(action: Action): Promise<void> {
+async function write(action: Action, dispatch: Dispatch): Promise<void> {
     if (action.type !== BLEDataActionType.Write) {
         return;
     }
-    const value = action.value.buffer;
-    for (let i = 0; i < value.byteLength; i += bleNusMaxSize) {
-        await rxChar?.xWriteValueWithoutResponse(value.slice(i, i + bleNusMaxSize));
+    try {
+        await rxChar?.xWriteValueWithoutResponse(action.value.buffer);
+        dispatch(didWrite(action.id));
+    } catch (err) {
+        dispatch(didFailToWrite(action.id, err));
     }
 }
 

src/services/error-log.ts

@@ -2,6 +2,7 @@
 // Copyright (c) 2020 The Pybricks Authors
 
 import { Action } from '../actions';
+import { BLEDataActionType } from '../actions/ble';
 import {
     BootloaderConnectionActionType,
     BootloaderConnectionFailureReason,
@@ -14,6 +15,9 @@ import { combineServices } from '.';
  */
 function consoleLog(action: Action): void {
     switch (action.type) {
+        case BLEDataActionType.DidFailToWrite:
+            console.error(action.err);
+            break;
         case BootloaderConnectionActionType.DidFailToConnect:
             if (action.reason === BootloaderConnectionFailureReason.Unknown) {
                 console.error(action.err);

src/services/hub.ts

@@ -3,7 +3,13 @@
 
 import { EventEmitter } from 'events';
 import { Action, Dispatch } from '../actions';
-import { write } from '../actions/ble';
+import {
+    BLEDataAction,
+    BLEDataActionType,
+    BLEDataDidFailToWriteAction,
+    BLEDataDidWriteAction,
+    write,
+} from '../actions/ble';
 import { HubActionType, HubRuntimeStatusType, updateStatus } from '../actions/hub';
 import {
     MpyActionType,
@@ -30,6 +36,17 @@ function didCompile(action: Action): void {
     }
 }
 
+const writer = new EventEmitter();
+
+function didWrite(action: Action): void {
+    if (action.type === BLEDataActionType.DidWrite) {
+        writer.emit('didWrite', action);
+    }
+    if (action.type === BLEDataActionType.DidFailToWrite) {
+        writer.emit('didFailToWrite', action);
+    }
+}
+
 async function downloadAndRun(
     action: Action,
     dispatch: Dispatch,
@@ -53,6 +70,7 @@ async function downloadAndRun(
             reject(new Error(a.err)),
         );
     });
+    compiler.removeAllListeners();
 
     // let everyone know the runtime is busy loading the program
     dispatch(updateStatus(HubRuntimeStatusType.Loading));
@@ -64,7 +82,13 @@ async function downloadAndRun(
     const sizeBuf = new Uint8Array(4);
     const sizeView = new DataView(sizeBuf.buffer);
     sizeView.setUint32(0, mpy.data.byteLength, true);
-    await dispatch(write(sizeBuf));
+    dispatch(write(sizeBuf));
+    await new Promise<BLEDataAction>((resolve, reject): void => {
+        writer.on('didWrite', (a: BLEDataDidWriteAction): void => resolve(a));
+        writer.on('didFailToWrite', (a: BLEDataDidFailToWriteAction) => reject(a.err));
+    });
+    writer.removeAllListeners();
+
     // TODO: verify checksum
     console.log(await checksum);
 
@@ -73,7 +97,20 @@ async function downloadAndRun(
     for (let i = 0; i < mpy.data.byteLength; i += downloadChunkSize) {
         // need to subscribe to checksum before writing to prevent race condition
         const checksum = getChecksum();
-        await dispatch(write(mpy.data.slice(i, i + downloadChunkSize)));
+        const chunk = mpy.data.slice(i, i + downloadChunkSize);
+
+        // we can actually only write 20 bytes at a time
+        for (let j = 0; j < chunk.length; j += 20) {
+            dispatch(write(chunk.slice(j, j + 20)));
+            await new Promise<BLEDataAction>((resolve, reject): void => {
+                writer.on('didWrite', (a: BLEDataDidWriteAction): void => resolve(a));
+                writer.on('didFailToWrite', (a: BLEDataDidFailToWriteAction) =>
+                    reject(a.err),
+                );
+            });
+            writer.removeAllListeners();
+        }
+
         // TODO: verify checksum
         console.log(await checksum);
         // TODO: dispatch progress
@@ -103,4 +140,4 @@ function stop(action: Action, dispatch: Dispatch): void {
     dispatch(write(stopCommand));
 }
 
-export default combineServices(didCompile, downloadAndRun, startRepl, stop);
+export default combineServices(didCompile, didWrite, downloadAndRun, startRepl, stop);