move ble epic to terminal saga

this is the last epic so removing redux-observable as well

package.json

@@ -33,7 +33,6 @@
         "react-splitter-layout": "^4.0.0",
         "redux": "^4.0.5",
         "redux-logger": "^3.0.6",
-        "redux-observable": "^1.2.0",
         "redux-saga": "^1.1.3",
         "typescript": "~3.9.5",
         "xterm": "^4.6.0",

src/epics/ble.ts

@@ -1,43 +0,0 @@
-// SPDX-License-Identifier: MIT
-// Copyright (c) 2020 The Pybricks Authors
-
-import { AnyAction } from 'redux';
-import { Epic, combineEpics, ofType } from 'redux-observable';
-import { map } from 'rxjs/operators';
-import { BLEDataActionType, BLEDataNotifyAction } from '../actions/ble';
-import { HubRuntimeStatusType, checksum, updateStatus } from '../actions/hub';
-import { sendData } from '../actions/terminal';
-import { RootState } from '../reducers';
-import { HubRuntimeState } from '../reducers/hub';
-
-const decoder = new TextDecoder();
-
-const rxUartData: Epic<AnyAction, AnyAction, RootState> = (action$, state$) =>
-    action$.pipe(
-        ofType<AnyAction, BLEDataNotifyAction>(BLEDataActionType.Notify),
-        map((a) => {
-            if (
-                state$.value.hub.runtime === HubRuntimeState.Loading &&
-                a.value.buffer.byteLength === 1
-            ) {
-                const view = new DataView(a.value.buffer);
-                return checksum(view.getUint8(0));
-            } else {
-                const value = decoder.decode(a.value.buffer);
-                // FIXME: sometimes we get ERROR and IDLE in same message except
-                // last E is cut off
-                if (value.match(/>>>> IDLE/)) {
-                    return updateStatus(HubRuntimeStatusType.Idle);
-                }
-                if (value.match(/>>>> ERROR/)) {
-                    return updateStatus(HubRuntimeStatusType.Error);
-                }
-                if (value.match(/>>>> RUNNING/)) {
-                    return updateStatus(HubRuntimeStatusType.Running);
-                }
-                return sendData(value);
-            }
-        }),
-    );
-
-export default combineEpics(rxUartData);

src/epics/index.ts

@@ -1,16 +0,0 @@
-// SPDX-License-Identifier: MIT
-// Copyright (c) 2020 The Pybricks Authors
-
-import { Epic, combineEpics } from 'redux-observable';
-import { catchError } from 'rxjs/operators';
-import ble from './ble';
-
-const rootEpic: Epic = (action$, store$, dependencies) =>
-    combineEpics(ble)(action$, store$, dependencies).pipe(
-        catchError((error, source) => {
-            console.error(error);
-            return source;
-        }),
-    );
-
-export default rootEpic;

src/index.tsx

@@ -8,20 +8,17 @@ import ReactDOM from 'react-dom';
 import { Provider } from 'react-redux';
 import { applyMiddleware, createStore } from 'redux';
 import { createLogger } from 'redux-logger';
-import { createEpicMiddleware } from 'redux-observable';
 import createSagaMiddleware from 'redux-saga';
 import './index.scss';
 // import { success, update } from './actions/service-worker';
 import App from './components/App';
 import NotificationStack from './components/NotificationStack';
-import rootEpic from './epics';
 import rootReducer from './reducers';
 import rootSaga from './sagas';
 import * as serviceWorker from './serviceWorker';
 import serviceMiddleware from './services';
 
 const sagaMiddleware = createSagaMiddleware();
-const epicMiddleware = createEpicMiddleware();
 // TODO: add runtime option or filter - logger affects firmware flash performance
 const loggerMiddleware = createLogger({ predicate: () => false });
 
@@ -32,16 +29,10 @@ const i18n = new I18nManager({
 
 const store = createStore(
     rootReducer,
-    applyMiddleware(
-        sagaMiddleware,
-        epicMiddleware,
-        serviceMiddleware,
-        loggerMiddleware,
-    ),
+    applyMiddleware(sagaMiddleware, serviceMiddleware, loggerMiddleware),
 );
 
 sagaMiddleware.run(rootSaga);
-epicMiddleware.run(rootEpic);
 
 ReactDOM.render(
     <React.StrictMode>

src/sagas/terminal.test.ts

@@ -8,15 +8,313 @@ import {
     BLEDataWriteAction,
     didFailToWrite,
     didWrite,
+    notify,
 } from '../actions/ble';
+import {
+    HubChecksumMessageAction,
+    HubMessageActionType,
+    HubRuntimeStatusMessageAction,
+    HubRuntimeStatusType,
+} from '../actions/hub';
 import {
     TerminalActionType,
+    TerminalDataSendDataAction,
     TerminalSetDataSourceAction,
     receiveData,
     sendData,
 } from '../actions/terminal';
+import { HubRuntimeState } from '../reducers/hub';
 import terminal from './terminal';
 
+describe('Data receiver filters out hub status', () => {
+    test('normal message - no status', async () => {
+        const saga = new AsyncSaga(terminal);
+
+        const dataSourceAction = await saga.take();
+        expect(dataSourceAction.type).toBe(TerminalActionType.SetDataSource);
+
+        // sending ASCII space character
+        saga.setState({ hub: { runtime: HubRuntimeState.Unknown } });
+        saga.put(notify(new DataView(new Uint8Array([0x20]).buffer)));
+
+        const action = await saga.take();
+        expect(action.type).toBe(TerminalActionType.SendData);
+        expect((action as TerminalDataSendDataAction).value).toBe(' ');
+
+        await saga.end();
+    });
+
+    test('checksum message', async () => {
+        const saga = new AsyncSaga(terminal);
+
+        const dataSourceAction = await saga.take();
+        expect(dataSourceAction.type).toBe(TerminalActionType.SetDataSource);
+
+        saga.setState({ hub: { runtime: HubRuntimeState.Loading } });
+        saga.put(notify(new DataView(new Uint8Array([0xaa]).buffer)));
+
+        const action = await saga.take();
+        expect(action.type).toBe(HubMessageActionType.Checksum);
+        expect((action as HubChecksumMessageAction).checksum).toBe(0xaa);
+
+        await saga.end();
+    });
+
+    test('idle message', async () => {
+        const saga = new AsyncSaga(terminal);
+
+        const dataSourceAction = await saga.take();
+        expect(dataSourceAction.type).toBe(TerminalActionType.SetDataSource);
+
+        // '>>>> IDLE'
+        saga.setState({ hub: { runtime: HubRuntimeState.Unknown } });
+        saga.put(
+            notify(
+                new DataView(
+                    new Uint8Array([
+                        0x3e,
+                        0x3e,
+                        0x3e,
+                        0x3e,
+                        0x20,
+                        0x49,
+                        0x44,
+                        0x4c,
+                        0x45,
+                    ]).buffer,
+                ),
+            ),
+        );
+
+        const action = await saga.take();
+        expect(action.type).toBe(HubMessageActionType.RuntimeStatus);
+        expect((action as HubRuntimeStatusMessageAction).newStatus).toBe(
+            HubRuntimeStatusType.Idle,
+        );
+
+        await saga.end();
+    });
+
+    test('idle message with extra text', async () => {
+        const saga = new AsyncSaga(terminal);
+
+        const dataSourceAction = await saga.take();
+        expect(dataSourceAction.type).toBe(TerminalActionType.SetDataSource);
+
+        // '0>>>> IDLE1'
+        saga.setState({ hub: { runtime: HubRuntimeState.Unknown } });
+        saga.put(
+            notify(
+                new DataView(
+                    new Uint8Array([
+                        0x30,
+                        0x3e,
+                        0x3e,
+                        0x3e,
+                        0x3e,
+                        0x20,
+                        0x49,
+                        0x44,
+                        0x4c,
+                        0x45,
+                        0x31,
+                    ]).buffer,
+                ),
+            ),
+        );
+
+        // this should get split into '0', idle status, '1'
+
+        const action1 = await saga.take();
+        expect(action1.type).toBe(TerminalActionType.SendData);
+        expect((action1 as TerminalDataSendDataAction).value).toBe('0');
+
+        const action2 = await saga.take();
+        expect(action2.type).toBe(HubMessageActionType.RuntimeStatus);
+        expect((action2 as HubRuntimeStatusMessageAction).newStatus).toBe(
+            HubRuntimeStatusType.Idle,
+        );
+
+        const action3 = await saga.take();
+        expect(action3.type).toBe(TerminalActionType.SendData);
+        expect((action3 as TerminalDataSendDataAction).value).toBe('1');
+
+        await saga.end();
+    });
+
+    test('error message', async () => {
+        const saga = new AsyncSaga(terminal);
+
+        const dataSourceAction = await saga.take();
+        expect(dataSourceAction.type).toBe(TerminalActionType.SetDataSource);
+
+        // '>>>> ERROR'
+        saga.setState({ hub: { runtime: HubRuntimeState.Unknown } });
+        saga.put(
+            notify(
+                new DataView(
+                    new Uint8Array([
+                        0x3e,
+                        0x3e,
+                        0x3e,
+                        0x3e,
+                        0x20,
+                        0x45,
+                        0x52,
+                        0x52,
+                        0x4f,
+                        0x52,
+                    ]).buffer,
+                ),
+            ),
+        );
+
+        const action = await saga.take();
+        expect(action.type).toBe(HubMessageActionType.RuntimeStatus);
+        expect((action as HubRuntimeStatusMessageAction).newStatus).toBe(
+            HubRuntimeStatusType.Error,
+        );
+
+        await saga.end();
+    });
+
+    test('error message with extra text', async () => {
+        const saga = new AsyncSaga(terminal);
+
+        const dataSourceAction = await saga.take();
+        expect(dataSourceAction.type).toBe(TerminalActionType.SetDataSource);
+
+        // '0>>>> ERROR1'
+        saga.setState({ hub: { runtime: HubRuntimeState.Unknown } });
+        saga.put(
+            notify(
+                new DataView(
+                    new Uint8Array([
+                        0x30,
+                        0x3e,
+                        0x3e,
+                        0x3e,
+                        0x3e,
+                        0x20,
+                        0x45,
+                        0x52,
+                        0x52,
+                        0x4f,
+                        0x52,
+                        0x31,
+                    ]).buffer,
+                ),
+            ),
+        );
+
+        // this should get split into '0', error status, '1'
+
+        const action1 = await saga.take();
+        expect(action1.type).toBe(TerminalActionType.SendData);
+        expect((action1 as TerminalDataSendDataAction).value).toBe('0');
+
+        const action2 = await saga.take();
+        expect(action2.type).toBe(HubMessageActionType.RuntimeStatus);
+        expect((action2 as HubRuntimeStatusMessageAction).newStatus).toBe(
+            HubRuntimeStatusType.Error,
+        );
+
+        const action3 = await saga.take();
+        expect(action3.type).toBe(TerminalActionType.SendData);
+        expect((action3 as TerminalDataSendDataAction).value).toBe('1');
+
+        await saga.end();
+    });
+
+    test('running message', async () => {
+        const saga = new AsyncSaga(terminal);
+
+        const dataSourceAction = await saga.take();
+        expect(dataSourceAction.type).toBe(TerminalActionType.SetDataSource);
+
+        // '>>>> ERROR'
+        saga.setState({ hub: { runtime: HubRuntimeState.Unknown } });
+        saga.put(
+            notify(
+                new DataView(
+                    new Uint8Array([
+                        0x3e,
+                        0x3e,
+                        0x3e,
+                        0x3e,
+                        0x20,
+                        0x52,
+                        0x55,
+                        0x4e,
+                        0x4e,
+                        0x49,
+                        0x4e,
+                        0x47,
+                    ]).buffer,
+                ),
+            ),
+        );
+
+        const action = await saga.take();
+        expect(action.type).toBe(HubMessageActionType.RuntimeStatus);
+        expect((action as HubRuntimeStatusMessageAction).newStatus).toBe(
+            HubRuntimeStatusType.Running,
+        );
+
+        await saga.end();
+    });
+
+    test('running message with extra text', async () => {
+        const saga = new AsyncSaga(terminal);
+
+        const dataSourceAction = await saga.take();
+        expect(dataSourceAction.type).toBe(TerminalActionType.SetDataSource);
+
+        // '0>>>> RUNNING1'
+        saga.setState({ hub: { runtime: HubRuntimeState.Unknown } });
+        saga.put(
+            notify(
+                new DataView(
+                    new Uint8Array([
+                        0x30,
+                        0x3e,
+                        0x3e,
+                        0x3e,
+                        0x3e,
+                        0x20,
+                        0x52,
+                        0x55,
+                        0x4e,
+                        0x4e,
+                        0x49,
+                        0x4e,
+                        0x47,
+                        0x31,
+                    ]).buffer,
+                ),
+            ),
+        );
+
+        // this should get split into '0', running status, '1'
+
+        const action1 = await saga.take();
+        expect(action1.type).toBe(TerminalActionType.SendData);
+        expect((action1 as TerminalDataSendDataAction).value).toBe('0');
+
+        const action2 = await saga.take();
+        expect(action2.type).toBe(HubMessageActionType.RuntimeStatus);
+        expect((action2 as HubRuntimeStatusMessageAction).newStatus).toBe(
+            HubRuntimeStatusType.Running,
+        );
+
+        const action3 = await saga.take();
+        expect(action3.type).toBe(TerminalActionType.SendData);
+        expect((action3 as TerminalDataSendDataAction).value).toBe('1');
+
+        await saga.end();
+    });
+});
+
 test('Terminal data source responds to send data actions', async () => {
     const saga = new AsyncSaga(terminal);
 

src/sagas/terminal.ts

@@ -8,21 +8,91 @@ import {
     fork,
     put,
     race,
+    select,
     take,
     takeEvery,
 } from 'redux-saga/effects';
 import PushStream from 'zen-push';
 import { Action } from '../actions';
-import { BLEDataActionType, BLEDataWriteAction, write } from '../actions/ble';
+import {
+    BLEDataActionType,
+    BLEDataNotifyAction,
+    BLEDataWriteAction,
+    write,
+} from '../actions/ble';
+import { HubRuntimeStatusType, checksum, updateStatus } from '../actions/hub';
 import {
     TerminalActionType,
     TerminalDataReceiveDataAction,
+    sendData,
     setDataSource,
 } from '../actions/terminal';
+import { RootState } from '../reducers';
+import { HubRuntimeState } from '../reducers/hub';
 
 const encoder = new TextEncoder();
+const decoder = new TextDecoder();
 const terminalDataSource = new PushStream<string>();
 
+function* handleMatch(
+    match: RegExpMatchArray | null,
+    status: HubRuntimeStatusType,
+): Generator<unknown, boolean> {
+    if (!match) {
+        return false;
+    }
+
+    if (match[1]) {
+        yield put(sendData(match[1]));
+    }
+
+    yield put(updateStatus(status));
+
+    if (match[2]) {
+        yield put(sendData(match[2]));
+    }
+
+    return true;
+}
+
+function* receiveUartData(action: BLEDataNotifyAction): Generator {
+    const hubState = (yield select((s: RootState) => s.hub.runtime)) as HubRuntimeState;
+
+    if (hubState === HubRuntimeState.Loading && action.value.buffer.byteLength === 1) {
+        const view = new DataView(action.value.buffer);
+        yield put(checksum(view.getUint8(0)));
+        return;
+    }
+
+    const value = decoder.decode(action.value.buffer);
+
+    if (
+        yield* handleMatch(value.match(/(.*)>>>> IDLE(.*)/), HubRuntimeStatusType.Idle)
+    ) {
+        return;
+    }
+
+    if (
+        yield* handleMatch(
+            value.match(/(.*)>>>> ERROR(.*)/),
+            HubRuntimeStatusType.Error,
+        )
+    ) {
+        return;
+    }
+
+    if (
+        yield* handleMatch(
+            value.match(/(.*)>>>> RUNNING(.*)/),
+            HubRuntimeStatusType.Running,
+        )
+    ) {
+        return;
+    }
+
+    yield put(sendData(value));
+}
+
 function* receiveTerminalData(): Generator {
     const channel = (yield actionChannel(TerminalActionType.ReceivedData)) as Channel<
         TerminalDataReceiveDataAction
@@ -70,6 +140,7 @@ function sendTerminalData(action: TerminalDataReceiveDataAction): void {
 }
 
 export default function* (): Generator {
+    yield takeEvery(BLEDataActionType.Notify, receiveUartData);
     yield fork(receiveTerminalData);
     yield takeEvery(TerminalActionType.SendData, sendTerminalData);
     yield put(setDataSource(terminalDataSource.observable));

yarn.lock

@@ -9740,11 +9740,6 @@ redux-logger@^3.0.6:
   dependencies:
     deep-diff "^0.3.5"
 
-redux-observable@^1.2.0:
-  version "1.2.0"
-  resolved "https://registry.yarnpkg.com/redux-observable/-/redux-observable-1.2.0.tgz#ff51b6c6be2598e9b5e89fc36639186bb0e669c7"
-  integrity sha512-yeR90RP2WzZzCxxnQPlh2uFzyfFLsfXu8ROh53jGDPXVqj71uNDMmvi/YKQkd9ofiVoO4OYb1snbowO49tCEMg==
-
 redux-saga@^1.1.3:
   version "1.1.3"
   resolved "https://registry.yarnpkg.com/redux-saga/-/redux-saga-1.1.3.tgz#9f3e6aebd3c994bbc0f6901a625f9a42b51d1112"