diff --git a/.vscode/launch.json b/.vscode/launch.json index d9e46586..b8120989 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -13,6 +13,18 @@ "linux": { "runtimeExecutable": "/usr/bin/chromium-browser" } + }, + { + "name": "Debug Tests", + "type": "node", + "request": "launch", + "runtimeExecutable": "${workspaceRoot}/node_modules/.bin/react-scripts", + "env": { "CI": "true" }, + "args": ["test", "--runInBand", "--no-cache", "--env=./test-env.js"], + "cwd": "${workspaceRoot}", + "protocol": "inspector", + "console": "integratedTerminal", + "internalConsoleOptions": "neverOpen" } ] } diff --git a/src/actions/ble.ts b/src/actions/ble.ts index c6d2f182..7292db58 100644 --- a/src/actions/ble.ts +++ b/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 { +const nextId = createCountFunc(); + +export type BLEDataWriteAction = Action & { + 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 & { + id: number; +}; + +export function didWrite(id: number): BLEDataDidWriteAction { + return { type: BLEDataActionType.DidWrite, id }; +} + +export type BLEDataDidFailToWriteAction = Action & { + id: number; + err: Error; +}; + +export function didFailToWrite(id: number, err: Error): BLEDataDidFailToWriteAction { + return { type: BLEDataActionType.DidFailToWrite, id, err }; } export interface BLEDataNotifyAction extends Action { @@ -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. diff --git a/src/epics/ble.ts b/src/epics/ble.ts index 7448c620..e67d9850 100644 --- a/src/epics/ble.ts +++ b/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 = (action$, state$) => action$.pipe( - ofType(BLEDataActionType.Notify), + ofType(BLEDataActionType.Notify), map((a) => { if ( state$.value.hub.runtime === HubRuntimeState.Loading && diff --git a/src/sagas/terminal.test.ts b/src/sagas/terminal.test.ts index e4702ce3..f0b4cb8e 100644 --- a/src/sagas/terminal.test.ts +++ b/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(); + }); }); diff --git a/src/sagas/terminal.ts b/src/sagas/terminal.ts index aa15f380..6e0a5ee7 100644 --- a/src/sagas/terminal.ts +++ b/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(); -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; + 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)); } diff --git a/src/services/ble.ts b/src/services/ble.ts index aba7b9ed..ff9cd1d8 100644 --- a/src/services/ble.ts +++ b/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 { +async function write(action: Action, dispatch: Dispatch): Promise { 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)); } } diff --git a/src/services/error-log.ts b/src/services/error-log.ts index 672354ab..16a16cd6 100644 --- a/src/services/error-log.ts +++ b/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); diff --git a/src/services/hub.ts b/src/services/hub.ts index d223d33a..287e047d 100644 --- a/src/services/hub.ts +++ b/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((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((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);