convert terminal from epic to saga

package.json

@@ -19,6 +19,7 @@
         "@types/react-splitter-layout": "^3.0.0",
         "@types/redux-logger": "^3.0.8",
         "@types/web-bluetooth": "^0.0.6",
+        "@types/zen-push": "^0.1.1",
         "ace-builds": "^1.4.11",
         "file-saver": "^2.0.2",
         "jszip": "^3.4.0",
@@ -36,7 +37,8 @@
         "redux-saga": "^1.1.3",
         "typescript": "~3.9.5",
         "xterm": "^4.6.0",
-        "xterm-addon-fit": "^0.4.0"
+        "xterm-addon-fit": "^0.4.0",
+        "zen-push": "^0.2.1"
     },
     "scripts": {
         "prepare": "mkdir -p public/static/js && cp node_modules/@pybricks/mpy-cross-v5/build/mpy-cross.wasm public/static/js/",

src/actions/terminal.ts

@@ -1,27 +1,53 @@
 // SPDX-License-Identifier: MIT
 // Copyright (c) 2020 The Pybricks Authors
 
-import { Action } from 'redux';
+import { Action, Observable } from 'redux';
 
-export enum TerminalDataActionType {
+export enum TerminalActionType {
+    /**
+     * Set the current data source.
+     */
+    SetDataSource = 'terminal.action.setDataSource',
     /**
      * Send data.
      */
-    SendData = 'terminal.data.send',
+    SendData = 'terminal.action.sendData',
     /**
      * Data was received.
      */
-    ReceivedData = 'terminal.data.receive',
+    ReceivedData = 'terminal.action.receiveData',
 }
 
-export interface TerminalDataAction extends Action<TerminalDataActionType> {
+export interface TerminalSetDataSourceAction
+    extends Action<TerminalActionType.SetDataSource> {
+    dataSource: Observable<string>;
+}
+
+export function setDataSource(
+    dataSource: Observable<string>,
+): TerminalSetDataSourceAction {
+    return { type: TerminalActionType.SetDataSource, dataSource };
+}
+
+export interface TerminalDataSendDataAction
+    extends Action<TerminalActionType.SendData> {
     value: string;
 }
 
-export function sendData(data: string): TerminalDataAction {
-    return { type: TerminalDataActionType.SendData, value: data };
+export function sendData(data: string): TerminalDataSendDataAction {
+    return { type: TerminalActionType.SendData, value: data };
 }
 
-export function receiveData(data: string): TerminalDataAction {
-    return { type: TerminalDataActionType.ReceivedData, value: data };
+export interface TerminalDataReceiveDataAction
+    extends Action<TerminalActionType.ReceivedData> {
+    value: string;
 }
+
+export function receiveData(data: string): TerminalDataReceiveDataAction {
+    return { type: TerminalActionType.ReceivedData, value: data };
+}
+
+export type TerminalDataAction =
+    | TerminalSetDataSourceAction
+    | TerminalDataSendDataAction
+    | TerminalDataReceiveDataAction;

src/components/Terminal.tsx

@@ -4,26 +4,30 @@
 import { ResizeSensor } from '@blueprintjs/core';
 import React from 'react';
 import { connect } from 'react-redux';
-import { Subscription } from 'rxjs';
+import { Observable, Unsubscribe } from 'redux';
 import { Terminal as XTerm } from 'xterm';
 import { FitAddon } from 'xterm-addon-fit';
 import { Dispatch } from '../actions';
 import { receiveData } from '../actions/terminal';
-import { terminalOutput } from '../epics/terminal';
+import { RootState } from '../reducers';
 
 import 'xterm/css/xterm.css';
 
+interface StateProps {
+    dataSource: Observable<string> | null;
+}
+
 interface DispatchProps {
     onData: (data: string) => void;
 }
 
-type TerminalProps = DispatchProps;
+type TerminalProps = StateProps & DispatchProps;
 
 class Terminal extends React.Component<TerminalProps> {
     private xterm: XTerm;
     private fitAddon: FitAddon;
     private terminalRef: React.RefObject<HTMLDivElement>;
-    private subscription?: Subscription;
+    private subscription?: { unsubscribe: Unsubscribe };
 
     constructor(props: TerminalProps) {
         super(props);
@@ -52,7 +56,9 @@ class Terminal extends React.Component<TerminalProps> {
         }
         this.xterm.open(this.terminalRef.current);
         this.fitAddon.fit();
-        this.subscription = terminalOutput.subscribe((v) => this.xterm.write(v));
+        this.subscription = this.props.dataSource?.subscribe({
+            next: (d) => this.xterm.write(d),
+        });
     }
 
     componentWillUnmount(): void {
@@ -73,10 +79,14 @@ class Terminal extends React.Component<TerminalProps> {
     }
 }
 
+const mapStateToProps = (state: RootState): StateProps => ({
+    dataSource: state.terminal.dataSource,
+});
+
 const mapDispatchToProps = (dispatch: Dispatch): DispatchProps => ({
     onData: (d): void => {
         dispatch(receiveData(d));
     },
 });
 
-export default connect(null, mapDispatchToProps)(Terminal);
+export default connect(mapStateToProps, mapDispatchToProps)(Terminal);

src/epics/index.ts

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

src/epics/terminal.ts

@@ -1,32 +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, map, tap } from 'rxjs/operators';
-import { write } from '../actions/ble';
-import { TerminalDataAction, TerminalDataActionType } from '../actions/terminal';
-
-const encoder = new TextEncoder();
-
-const terminalOutputSubject = new Subject<string>();
-export const terminalOutput = terminalOutputSubject.asObservable();
-
-// When terminal has focus this receives the input. All input is sent over BLE connection.
-const receiveTerminalData: Epic = (action$) =>
-    action$.pipe(
-        ofType<AnyAction, TerminalDataAction>(TerminalDataActionType.ReceivedData),
-        map((a) => write(encoder.encode(a.value))),
-    );
-
-// Request to write to the terminal are handled by an observable rather than
-// using redux state.
-const sendTerminalData: Epic = (action$) =>
-    action$.pipe(
-        ofType<AnyAction, TerminalDataAction>(TerminalDataActionType.SendData),
-        tap((a) => terminalOutputSubject.next(a.value)),
-        ignoreElements(),
-    );
-
-export default combineEpics(receiveTerminalData, sendTerminalData);

src/reducers/index.ts

@@ -8,6 +8,7 @@ import editor, { EditorState } from './editor';
 import hub, { HubState } from './hub';
 import notification, { NotificationState } from './notification';
 import status, { StatusState } from './status';
+import terminal, { TerminalState } from './terminal';
 
 /**
  * Root state for redux store.
@@ -19,6 +20,15 @@ export interface RootState {
     readonly hub: HubState;
     readonly notification: NotificationState;
     readonly status: StatusState;
+    readonly terminal: TerminalState;
 }
 
-export default combineReducers({ bootloader, ble, editor, hub, notification, status });
+export default combineReducers({
+    bootloader,
+    ble,
+    editor,
+    hub,
+    notification,
+    status,
+    terminal,
+});

src/reducers/terminal.ts

@@ -0,0 +1,24 @@
+// SPDX-License-Identifier: MIT
+// Copyright (c) 2020 The Pybricks Authors
+
+import { Reducer } from 'react';
+import { Observable, combineReducers } from 'redux';
+import { Action } from '../actions';
+import { TerminalActionType } from '../actions/terminal';
+
+type DataSource = Observable<string> | null;
+
+const dataSource: Reducer<DataSource, Action> = (state = null, action) => {
+    switch (action.type) {
+        case TerminalActionType.SetDataSource:
+            return action.dataSource;
+        default:
+            return state;
+    }
+};
+
+export interface TerminalState {
+    readonly dataSource: DataSource;
+}
+
+export default combineReducers({ dataSource });

src/sagas/index.ts

@@ -6,8 +6,9 @@ import editor from './editor';
 import flashFirmware from './flash-firmare';
 import bootloader from './lwp3-bootloader';
 import mpy from './mpy';
+import terminal from './terminal';
 
 /* istanbul ignore next */
 export default function* (): Generator {
-    yield all([bootloader(), editor(), flashFirmware(), mpy()]);
+    yield all([bootloader(), editor(), flashFirmware(), mpy(), terminal()]);
 }

src/sagas/terminal.test.ts

@@ -0,0 +1,117 @@
+// SPDX-License-Identifier: MIT
+// Copyright (c) 2020 The Pybricks Authors
+
+import { END, MulticastChannel, Saga, Task, runSaga, stdChannel } from 'redux-saga';
+import { Action } from '../actions';
+import { BLEDataActionType, BLEDataWriteAction } from '../actions/ble';
+import {
+    TerminalActionType,
+    TerminalSetDataSourceAction,
+    receiveData,
+    sendData,
+} from '../actions/terminal';
+import terminal from './terminal';
+
+class AsyncSaga {
+    private dispatches: (Action | END)[];
+    private takers: { put: (action: Action | END) => void }[];
+    private channel: MulticastChannel<Action>;
+    private task: Task;
+
+    public constructor(saga: Saga) {
+        this.dispatches = [];
+        this.takers = [];
+        this.channel = stdChannel();
+        this.task = runSaga(
+            { channel: this.channel, dispatch: this.dispatch.bind(this) },
+            saga,
+        );
+    }
+
+    public put(action: Action): void {
+        this.channel.put(action);
+    }
+
+    public take(): Promise<Action> {
+        const next = this.dispatches.shift();
+        if (next === undefined) {
+            // if there are no dispatches queued, then queue the taker to be
+            // completed later
+            return new Promise((resolve, reject) => {
+                this.takers.push({
+                    put: (a: Action | END): void => {
+                        if (a.type === END.type) {
+                            reject();
+                        } else {
+                            resolve(a);
+                        }
+                    },
+                });
+            });
+        }
+        // otherwise complete immediately
+        if (next.type === END.type) {
+            return Promise.reject();
+        }
+        return Promise.resolve(next);
+    }
+
+    public async end(): Promise<void> {
+        this.task.cancel();
+        await this.task.toPromise();
+        if (this.dispatches.some((x) => x.type !== END.type)) {
+            fail(`unhandled dispatches remain: ${this.dispatches}`);
+        }
+    }
+
+    private dispatch(action: Action | END): void {
+        const taker = this.takers.shift();
+        if (taker === undefined) {
+            // if there are no takers waiting, the queue the action
+            this.dispatches.push(action);
+        } else {
+            // otherwise complete the promise
+            taker.put(action);
+        }
+    }
+}
+
+test('Terminal data source responds to send data actions', async () => {
+    const saga = new AsyncSaga(terminal);
+
+    const dataSourceAction = await saga.take();
+    expect(dataSourceAction.type).toBe(TerminalActionType.SetDataSource);
+
+    const dataSource = (dataSourceAction as TerminalSetDataSourceAction).dataSource;
+    const data = new Array<string>();
+    dataSource.subscribe({ next: (v) => data.push(v) });
+
+    saga.put(sendData('1'));
+    saga.put(sendData('2'));
+    saga.put(sendData('3'));
+
+    expect(data.length).toBe(3);
+    expect(data[0]).toBe('1');
+    expect(data[1]).toBe('2');
+    expect(data[2]).toBe('3');
+
+    await saga.end();
+});
+
+test('Terminal data source responds to receive data actions', 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'));
+
+    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]),
+    );
+
+    await saga.end();
+});

src/sagas/terminal.ts

@@ -0,0 +1,31 @@
+// SPDX-License-Identifier: MIT
+// Copyright (c) 2020 The Pybricks Authors
+
+import { put, takeEvery } from 'redux-saga/effects';
+import PushStream from 'zen-push';
+import { 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 sendTerminalData(action: TerminalDataReceiveDataAction): void {
+    // This is used to provide a data source for the Terminal component
+    terminalDataSource.next(action.value);
+}
+
+export default function* (): Generator {
+    yield takeEvery(TerminalActionType.ReceivedData, receiveTerminalData);
+    yield takeEvery(TerminalActionType.SendData, sendTerminalData);
+    yield put(setDataSource(terminalDataSource.observable));
+}

yarn.lock

@@ -1836,6 +1836,18 @@
   dependencies:
     "@types/yargs-parser" "*"
 
+"@types/zen-observable@*":
+  version "0.8.0"
+  resolved "https://registry.yarnpkg.com/@types/zen-observable/-/zen-observable-0.8.0.tgz#8b63ab7f1aa5321248aad5ac890a485656dcea4d"
+  integrity sha512-te5lMAWii1uEJ4FwLjzdlbw3+n0FZNOvFXHxQDKeT0dilh7HOzdMzV2TrJVUzq8ep7J4Na8OUYPRLSQkJHAlrg==
+
+"@types/zen-push@^0.1.1":
+  version "0.1.1"
+  resolved "https://registry.yarnpkg.com/@types/zen-push/-/zen-push-0.1.1.tgz#a058a080d4488e4fa88492b2bb0c8447b5d272c0"
+  integrity sha512-lc6wO0vtv3LFNXj1U8cDrtB5hajXzXCWYO0cztPf5xzHKbciVPf5H5uccVLgApE1KlXb4z0X1bjvYkLwhXIQfA==
+  dependencies:
+    "@types/zen-observable" "*"
+
 "@typescript-eslint/eslint-plugin@^2.10.0", "@typescript-eslint/eslint-plugin@^2.34.0":
   version "2.34.0"
   resolved "https://registry.yarnpkg.com/@typescript-eslint/eslint-plugin/-/eslint-plugin-2.34.0.tgz#6f8ce8a46c7dea4a6f1d171d2bb8fbae6dac2be9"
@@ -12081,3 +12093,15 @@ yargs@^13.3.0, yargs@^13.3.2:
     which-module "^2.0.0"
     y18n "^4.0.0"
     yargs-parser "^13.1.2"
+
+zen-observable@^0.7.0:
+  version "0.7.1"
+  resolved "https://registry.yarnpkg.com/zen-observable/-/zen-observable-0.7.1.tgz#f84075c0ee085594d3566e1d6454207f126411b3"
+  integrity sha512-OI6VMSe0yeqaouIXtedC+F55Sr6r9ppS7+wTbSexkYdHbdt4ctTuPNXP/rwm7GTVI63YBc+EBT0b0tl7YnJLRg==
+
+zen-push@^0.2.1:
+  version "0.2.1"
+  resolved "https://registry.yarnpkg.com/zen-push/-/zen-push-0.2.1.tgz#ddc33b90f66f9a84237d5f1893970f6be60c3c28"
+  integrity sha512-Qv4qvc8ZIue51B/0zmeIMxpIGDVhz4GhJALBvnKs/FRa2T7jy4Ori9wFwaHVt0zWV7MIFglKAHbgnVxVTw7U1w==
+  dependencies:
+    zen-observable "^0.7.0"