From 34184785c80a5a7d08e851966aad2b8ff708005f Mon Sep 17 00:00:00 2001 From: David Lechner Date: Wed, 10 Jun 2020 20:43:17 -0500 Subject: [PATCH] convert hub service and epic to saga --- src/epics/hub.ts | 23 ------ src/epics/index.ts | 3 +- src/sagas/hub.test.ts | 97 +++++++++++++++++++++++++ src/sagas/hub.ts | 161 ++++++++++++++++++++++++++++++++++++++++++ src/sagas/index.ts | 11 ++- src/services/hub.ts | 143 ------------------------------------- src/services/index.ts | 3 +- 7 files changed, 270 insertions(+), 171 deletions(-) delete mode 100644 src/epics/hub.ts create mode 100644 src/sagas/hub.test.ts create mode 100644 src/sagas/hub.ts delete mode 100644 src/services/hub.ts diff --git a/src/epics/hub.ts b/src/epics/hub.ts deleted file mode 100644 index 758d2928..00000000 --- a/src/epics/hub.ts +++ /dev/null @@ -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(); - -const checksum: Epic = (action$) => - action$.pipe( - ofType(HubMessageActionType.Checksum), - tap((a) => checksumSubject.next(a.checksum)), - ignoreElements(), - ); - -export function getChecksum(): Promise { - return checksumSubject.pipe(take(1)).toPromise(); -} - -export default combineEpics(checksum); diff --git a/src/epics/index.ts b/src/epics/index.ts index 03ae0e07..ab38ab56 100644 --- a/src/epics/index.ts +++ b/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; diff --git a/src/sagas/hub.test.ts b/src/sagas/hub.test.ts new file mode 100644 index 00000000..59a6d27f --- /dev/null +++ b/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(); + 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(); +}); diff --git a/src/sagas/hub.ts b/src/sagas/hub.ts new file mode 100644 index 00000000..833ba7f6 --- /dev/null +++ b/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 { + 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; + + // 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); +} diff --git a/src/sagas/index.ts b/src/sagas/index.ts index 7c11a143..46a79d15 100644 --- a/src/sagas/index.ts +++ b/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(), + ]); } diff --git a/src/services/hub.ts b/src/services/hub.ts deleted file mode 100644 index 287e047d..00000000 --- a/src/services/hub.ts +++ /dev/null @@ -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 { - 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((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((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((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); diff --git a/src/services/index.ts b/src/services/index.ts index 8477656b..55e58def 100644 --- a/src/services/index.ts +++ b/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());