Throttle BLE messages from terminal

This makes the terminal more efficient by combining individual characters into a single message. Also (mostly) fixes overring the hub and locking it up.

src/sagas/terminal.test.ts

@@ -90,6 +90,10 @@ class AsyncSaga {
     }
 }
 
+function delay(ms: number): Promise<void> {
+    return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
 test('Terminal data source responds to send data actions', async () => {
     const saga = new AsyncSaga(terminal);
 
@@ -140,6 +144,7 @@ describe('Terminal data source responds to receive data actions', () => {
         expect(dataSourceAction.type).toBe(TerminalActionType.SetDataSource);
 
         saga.put(receiveData('test1234'));
+        await delay(50); // without delay, messages are combined
         saga.put(receiveData('test1234'));
 
         // second message is queued until didWrite or didFailToWrite
@@ -171,6 +176,7 @@ describe('Terminal data source responds to receive data actions', () => {
         expect(dataSourceAction.type).toBe(TerminalActionType.SetDataSource);
 
         saga.put(receiveData('test1234'));
+        await delay(50); // without delay, messages are combined
         saga.put(receiveData('test1234'));
 
         // second message is queued until didWrite or didFailToWrite
@@ -196,6 +202,25 @@ describe('Terminal data source responds to receive data actions', () => {
         await saga.end();
     });
 
+    test('small messages are combined', 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'));
+
+        const action = await saga.take();
+        expect(action.type).toBe(BLEDataActionType.Write);
+        expect((action as BLEDataWriteAction).value).toEqual(
+            new Uint8Array([...expected, ...expected]),
+        );
+
+        await saga.end();
+    });
+
     test('long messages are split', async () => {
         const saga = new AsyncSaga(terminal);
 

src/sagas/terminal.ts

@@ -2,7 +2,15 @@
 // Copyright (c) 2020 The Pybricks Authors
 
 import { Channel, buffers } from 'redux-saga';
-import { actionChannel, fork, put, take, takeEvery } from 'redux-saga/effects';
+import {
+    actionChannel,
+    delay,
+    fork,
+    put,
+    race,
+    take,
+    takeEvery,
+} from 'redux-saga/effects';
 import PushStream from 'zen-push';
 import { Action } from '../actions';
 import { BLEDataActionType, BLEDataWriteAction, write } from '../actions/ble';
@@ -23,9 +31,22 @@ function* receiveTerminalData(): Generator {
     while (true) {
         // wait for input from terminal
         const action = (yield take(channel)) as TerminalDataReceiveDataAction;
+        let value = action.value;
+
+        // Try to collect more data so that we aren't sending just one byte at time
+        while (value.length < 20) {
+            const [action, timeout] = (yield race([take(channel), delay(20)])) as [
+                TerminalDataReceiveDataAction,
+                boolean,
+            ];
+            if (timeout) {
+                break;
+            }
+            value += action.value;
+        }
 
         // stdin gets piped to BLE connection
-        const data = encoder.encode(action.value);
+        const data = encoder.encode(value);
         for (let i = 0; i < data.length; i += 20) {
             const { id } = (yield put(
                 write(data.slice(i, i + 20)),
@@ -37,6 +58,9 @@ function* receiveTerminalData(): Generator {
                         a.type === BLEDataActionType.DidFailToWrite) &&
                     a.id === id,
             );
+
+            // wait for echo so tht we don't overrun the hub with messages
+            yield race([take(BLEDataActionType.Notify), delay(100)]);
         }
     }
 }