convert hub service and epic to saga

src/epics/hub.ts

@@ -1,23 +0,0 @@
-// SPDX-License-Identifier: MIT
-// Copyright (c) 2020 The Pybricks Authors
-
-import { AnyAction } from 'redux';
-import { Epic, combineEpics, ofType } from 'redux-observable';
-import { Subject } from 'rxjs';
-import { ignoreElements, take, tap } from 'rxjs/operators';
-import { HubChecksumMessageAction, HubMessageActionType } from '../actions/hub';
-
-const checksumSubject = new Subject<number>();
-
-const checksum: Epic = (action$) =>
-    action$.pipe(
-        ofType<AnyAction, HubChecksumMessageAction>(HubMessageActionType.Checksum),
-        tap((a) => checksumSubject.next(a.checksum)),
-        ignoreElements(),
-    );
-
-export function getChecksum(): Promise<number> {
-    return checksumSubject.pipe(take(1)).toPromise();
-}
-
-export default combineEpics(checksum);

src/epics/index.ts

@@ -4,10 +4,9 @@
 import { Epic, combineEpics } from 'redux-observable';
 import { catchError } from 'rxjs/operators';
 import ble from './ble';
-import hub from './hub';
 
 const rootEpic: Epic = (action$, store$, dependencies) =>
-    combineEpics(ble, hub)(action$, store$, dependencies).pipe(
+    combineEpics(ble)(action$, store$, dependencies).pipe(
         catchError((error, source) => {
             console.error(error);
             return source;

src/sagas/hub.test.ts

@@ -0,0 +1,97 @@
+// SPDX-License-Identifier: MIT
+// Copyright (c) 2020 The Pybricks Authors
+
+import { Ace } from 'ace-builds';
+import { mock } from 'jest-mock-extended';
+import { AsyncSaga } from '../../test';
+import { BLEDataActionType, BLEDataWriteAction, didWrite } from '../actions/ble';
+import {
+    HubMessageActionType,
+    HubRuntimeStatusMessageAction,
+    HubRuntimeStatusType,
+    checksum,
+    downloadAndRun,
+    repl,
+    stop,
+} from '../actions/hub';
+import { MpyActionType, didCompile } from '../actions/mpy';
+import hub from './hub';
+
+jest.mock('ace-builds');
+
+describe('downloadAndRun', () => {
+    test('no errors', async () => {
+        const saga = new AsyncSaga(hub);
+
+        const mockEditor = mock<Ace.EditSession>();
+        saga.setState({ editor: { current: mockEditor } });
+
+        saga.put(downloadAndRun());
+
+        // first, it tries to compile the program in the current editor
+        const compileAction = await saga.take();
+        expect(compileAction.type).toBe(MpyActionType.Compile);
+        saga.put(didCompile(new Uint8Array(30)));
+
+        // then it notifies that loading has begun
+        const loadingStatusAction = await saga.take();
+        expect(loadingStatusAction.type).toBe(HubMessageActionType.RuntimeStatus);
+        expect((loadingStatusAction as HubRuntimeStatusMessageAction).newStatus).toBe(
+            HubRuntimeStatusType.Loading,
+        );
+
+        // first message is the length
+        const writeAction = await saga.take();
+        expect(writeAction.type).toBe(BLEDataActionType.Write);
+        expect((writeAction as BLEDataWriteAction).value.length).toBe(4);
+        saga.put(didWrite((writeAction as BLEDataWriteAction).id));
+        saga.put(checksum(30));
+
+        // then the first chunk of 20 bytes
+        const writeAction2 = await saga.take();
+        expect(writeAction2.type).toBe(BLEDataActionType.Write);
+        expect((writeAction2 as BLEDataWriteAction).value.length).toBe(20);
+        saga.put(didWrite((writeAction2 as BLEDataWriteAction).id));
+        saga.put(checksum(0));
+
+        // then last chunk
+        const writeAction3 = await saga.take();
+        expect(writeAction3.type).toBe(BLEDataActionType.Write);
+        expect((writeAction3 as BLEDataWriteAction).value.length).toBe(10);
+        saga.put(didWrite((writeAction3 as BLEDataWriteAction).id));
+        saga.put(checksum(0));
+
+        // Then a status message saying that we are done
+        const loadedStatusAction = await saga.take();
+        expect(loadedStatusAction.type).toBe(HubMessageActionType.RuntimeStatus);
+        expect((loadedStatusAction as HubRuntimeStatusMessageAction).newStatus).toBe(
+            HubRuntimeStatusType.Loaded,
+        );
+
+        await saga.end();
+    });
+
+    // TODO: need to test error paths
+});
+
+test('repl', async () => {
+    const saga = new AsyncSaga(hub);
+
+    saga.put(repl());
+
+    const compileAction = await saga.take();
+    expect(compileAction.type).toBe(BLEDataActionType.Write);
+
+    await saga.end();
+});
+
+test('stop', async () => {
+    const saga = new AsyncSaga(hub);
+
+    saga.put(stop());
+
+    const compileAction = await saga.take();
+    expect(compileAction.type).toBe(BLEDataActionType.Write);
+
+    await saga.end();
+});

src/sagas/hub.ts

@@ -0,0 +1,161 @@
+// SPDX-License-Identifier: MIT
+// Copyright (c) 2020 The Pybricks Authors
+
+import { Ace } from 'ace-builds';
+import { Channel } from 'redux-saga';
+import {
+    RaceEffect,
+    TakeEffect,
+    actionChannel,
+    put,
+    race,
+    select,
+    take,
+    takeEvery,
+} from 'redux-saga/effects';
+import { Action } from '../actions';
+import {
+    BLEDataActionType,
+    BLEDataDidFailToWriteAction,
+    BLEDataDidWriteAction,
+    BLEDataWriteAction,
+    write,
+} from '../actions/ble';
+import {
+    HubActionType,
+    HubChecksumMessageAction,
+    HubDownloadAndRunAction,
+    HubMessageActionType,
+    HubReplAction,
+    HubRuntimeStatusType,
+    HubStopAction,
+    updateStatus,
+} from '../actions/hub';
+import {
+    MpyActionType,
+    MpyDidCompileAction,
+    MpyDidFailToCompileAction,
+    compile,
+} from '../actions/mpy';
+import { RootState } from '../reducers';
+import { xor8 } from '../utils/math';
+
+const downloadChunkSize = 100;
+
+function waitForWrite(id: number): RaceEffect<TakeEffect> {
+    return race([
+        take((a: Action) => a.type === BLEDataActionType.DidWrite && a.id === id),
+        take((a: Action) => a.type === BLEDataActionType.DidFailToWrite && a.id === id),
+    ]);
+}
+
+function* downloadAndRun(_action: HubDownloadAndRunAction): Generator {
+    const editor = (yield select(
+        (s: RootState) => s.editor.current,
+    )) as Ace.EditSession | null;
+
+    // istanbul ignore next: it is a bug to dispatch this action with no current editor
+    if (editor === null) {
+        console.error('downloadAndRun: No current editor');
+        return;
+    }
+
+    const script = editor.getValue();
+    yield put(compile(script, ['-mno-unicode']));
+    const [mpy, mpyFail] = (yield race([
+        take(MpyActionType.DidCompile),
+        take(MpyActionType.DidFailToCompile),
+    ])) as [MpyDidCompileAction, MpyDidFailToCompileAction];
+
+    if (mpyFail) {
+        return;
+    }
+
+    // let everyone know the runtime is busy loading the program
+    yield put(updateStatus(HubRuntimeStatusType.Loading));
+
+    const checksumChannel = (yield actionChannel(
+        HubMessageActionType.Checksum,
+    )) as Channel<HubChecksumMessageAction>;
+
+    // first send payload size as big-endian 32-bit integer
+    const sizeBuf = new Uint8Array(4);
+    const sizeView = new DataView(sizeBuf.buffer);
+    sizeView.setUint32(0, mpy.data.byteLength, true);
+    const writeAction = (yield put(write(sizeBuf))) as BLEDataWriteAction;
+    const [, didFailToWrite] = (yield waitForWrite(writeAction.id)) as [
+        BLEDataDidWriteAction,
+        BLEDataDidFailToWriteAction,
+    ];
+
+    if (didFailToWrite) {
+        yield put(updateStatus(HubRuntimeStatusType.Error));
+        return;
+    }
+
+    const checksumAction = (yield take(checksumChannel)) as HubChecksumMessageAction;
+    if (checksumAction.checksum !== (0xff ^ xor8(sizeBuf))) {
+        console.error(
+            `bad checksum ${checksumAction.checksum} vs ${0xff ^ xor8(sizeBuf)}`,
+        );
+        yield put(updateStatus(HubRuntimeStatusType.Error));
+        return;
+    }
+
+    // Then send payload in 100 byte chunks waiting for checksum after
+    // each chunk
+    for (let i = 0; i < mpy.data.byteLength; i += downloadChunkSize) {
+        // need to subscribe to checksum before writing to prevent race condition
+        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) {
+            const writeAction = (yield put(
+                write(chunk.slice(j, j + 20)),
+            )) as BLEDataWriteAction;
+            const [, didFailToWrite] = (yield waitForWrite(writeAction.id)) as [
+                BLEDataDidWriteAction,
+                BLEDataDidFailToWriteAction,
+            ];
+
+            if (didFailToWrite) {
+                yield put(updateStatus(HubRuntimeStatusType.Error));
+                return;
+            }
+            // TODO: dispatch progress
+        }
+        const checksumAction = (yield take(
+            checksumChannel,
+        )) as HubChecksumMessageAction;
+        if (checksumAction.checksum !== (0xff ^ xor8(chunk))) {
+            console.error(
+                `bad checksum ${checksumAction.checksum} vs ${0xff ^ xor8(chunk)}`,
+            );
+            yield put(updateStatus(HubRuntimeStatusType.Error));
+            return;
+        }
+    }
+
+    // let everyone know the runtime is done loading the program
+    yield put(updateStatus(HubRuntimeStatusType.Loaded));
+}
+
+// SPACE, SPACE, SPACE, SPACE
+const startReplCommand = new Uint8Array([0x20, 0x20, 0x20, 0x20]);
+
+function* startRepl(_action: HubReplAction): Generator {
+    yield put(write(startReplCommand));
+}
+
+// CTRL+C, CTRL+C, CTRL+D
+const stopCommand = new Uint8Array([0x03, 0x03, 0x04]);
+
+function* stop(_action: HubStopAction): Generator {
+    yield put(write(stopCommand));
+}
+
+export default function* (): Generator {
+    yield takeEvery(HubActionType.DownloadAndRun, downloadAndRun);
+    yield takeEvery(HubActionType.Repl, startRepl);
+    yield takeEvery(HubActionType.Stop, stop);
+}

src/sagas/index.ts

@@ -5,11 +5,20 @@ import { all } from 'redux-saga/effects';
 import editor from './editor';
 import errorLog from './error-log';
 import flashFirmware from './flash-firmare';
+import hub from './hub';
 import bootloader from './lwp3-bootloader';
 import mpy from './mpy';
 import terminal from './terminal';
 
 /* istanbul ignore next */
 export default function* (): Generator {
-    yield all([bootloader(), editor(), errorLog(), flashFirmware(), mpy(), terminal()]);
+    yield all([
+        bootloader(),
+        editor(),
+        errorLog(),
+        flashFirmware(),
+        hub(),
+        mpy(),
+        terminal(),
+    ]);
 }

src/services/hub.ts

@@ -1,143 +0,0 @@
-// SPDX-License-Identifier: MIT
-// Copyright (c) 2020 The Pybricks Authors
-
-import { EventEmitter } from 'events';
-import { Action, Dispatch } from '../actions';
-import {
-    BLEDataAction,
-    BLEDataActionType,
-    BLEDataDidFailToWriteAction,
-    BLEDataDidWriteAction,
-    write,
-} from '../actions/ble';
-import { HubActionType, HubRuntimeStatusType, updateStatus } from '../actions/hub';
-import {
-    MpyActionType,
-    MpyDidCompileAction,
-    MpyDidFailToCompileAction,
-    compile,
-} from '../actions/mpy';
-import { getChecksum } from '../epics/hub';
-import { RootState } from '../reducers';
-import { combineServices } from '.';
-
-// TODO: this file needs to be converted to a saga
-
-const downloadChunkSize = 100;
-
-const compiler = new EventEmitter();
-
-function didCompile(action: Action): void {
-    if (action.type === MpyActionType.DidCompile) {
-        compiler.emit('didCompile', action);
-    }
-    if (action.type === MpyActionType.DidFailToCompile) {
-        compiler.emit('didFailToCompile', action);
-    }
-}
-
-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,
-    state: RootState,
-): Promise<void> {
-    if (action.type !== HubActionType.DownloadAndRun) {
-        return;
-    }
-
-    const script = state.editor.current?.getValue();
-    // istanbul ignore next: it should not be possible to trigger this action without a current editor
-    if (script === undefined) {
-        console.log('no current editor');
-        return;
-    }
-
-    dispatch(compile(script, ['-mno-unicode']));
-    const mpy = await new Promise<MpyDidCompileAction>((resolve, reject): void => {
-        compiler.on('didCompile', (a: MpyDidCompileAction): void => resolve(a));
-        compiler.on('didFailToCompile', (a: MpyDidFailToCompileAction) =>
-            reject(new Error(a.err)),
-        );
-    });
-    compiler.removeAllListeners();
-
-    // let everyone know the runtime is busy loading the program
-    dispatch(updateStatus(HubRuntimeStatusType.Loading));
-
-    // TODO: might need to flush checksum queue here
-
-    // first send payload size as big-endian 32-bit integer
-    const checksum = getChecksum();
-    const sizeBuf = new Uint8Array(4);
-    const sizeView = new DataView(sizeBuf.buffer);
-    sizeView.setUint32(0, mpy.data.byteLength, true);
-    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);
-
-    // Then send payload in 100 byte chunks waiting for checksum after
-    // each chunk
-    for (let i = 0; i < mpy.data.byteLength; i += downloadChunkSize) {
-        // need to subscribe to checksum before writing to prevent race condition
-        const checksum = getChecksum();
-        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
-    }
-
-    // let everyone know the runtime is done loading the program
-    dispatch(updateStatus(HubRuntimeStatusType.Loaded));
-}
-
-// SPACE, SPACE, SPACE, SPACE
-const startReplCommand = new Uint8Array([0x20, 0x20, 0x20, 0x20]);
-
-function startRepl(action: Action, dispatch: Dispatch): void {
-    if (action.type !== HubActionType.Repl) {
-        return;
-    }
-    dispatch(write(startReplCommand));
-}
-
-// CTRL+C, CTRL+C, CTRL+D
-const stopCommand = new Uint8Array([0x03, 0x03, 0x04]);
-
-function stop(action: Action, dispatch: Dispatch): void {
-    if (action.type !== HubActionType.Stop) {
-        return;
-    }
-    dispatch(write(stopCommand));
-}
-
-export default combineServices(didCompile, didWrite, downloadAndRun, startRepl, stop);

src/services/index.ts

@@ -5,7 +5,6 @@ import { Middleware } from 'redux';
 import { Action, Dispatch } from '../actions';
 import { RootState } from '../reducers';
 import ble from './ble';
-import hub from './hub';
 import bootloader from './lwp3-bootloader';
 
 type Service = (
@@ -37,7 +36,7 @@ export function combineServices(...services: Service[]): Service {
     };
 }
 
-const rootService = combineServices(ble, bootloader, hub);
+const rootService = combineServices(ble, bootloader);
 
 const serviceMiddleware: Middleware = (store) => (next) => (action): unknown => {
     runService(rootService, action, store.dispatch, store.getState());