高度な使用方法

SerialSession は意図的に小さな公開面に絞られています。応用パターンの大半は、receive$ と send$ の上に普通の RxJS オペレータを組み合わせることで表現できます。API の全体像は先にSerialSession の概要とクイックスタートを読み、本ページは概要で省いた行フレーミング・派生ストリーム・リカバリのレシピに絞ります。ライフサイクルとエラーは state$ の state.status narrowing と errors$ の error.is() を優先してください — v3 への移行 を参照。

本ページは Issue #228 で列挙したパターンに対応します。lines$ は SerialSession の組み込みとして用意されています。isConnected$ は v3.x で非推奨です — state$ narrowing を優先してください。sendLine・readUntil・waitForState などは、引き続きコア API の上に組み立てるパターンです（専用の追加 export はありません）。USB OTG シリアルコンソールの実例として CHIRIMEN PiZeroWebSerialConsole があります。同アプリの読み書きループを SerialSession で書き直すときも、ここでのレシピがそのまま使えます。

行単位のフレーミング（組み込み lines$ と receive$ 上のカスタム分割）

通常は 組み込みの lines$ で十分です（改行区切りの1行が都度1件 emit）。receive$ は従来どおり、デコーダが返す 生チャンクをそのまま流します。組み込み lines$ とは異なる区切り文字や正規化が必要なときだけ scan などでフレーミングします。

import { filter, map, scan } from 'rxjs';
import { createSerialSession } from '@gurezo/web-serial-rxjs';

const session = createSerialSession({ baudRate: 115200 });
session.connect$().subscribe();

// カスタムフレーミング（組み込み `lines$` では足りない場合のみ）
const customLines$ = session.receive$.pipe(
  scan(
    (acc, chunk) => {
      const combined = acc.buffer + chunk;
      const parts = combined.split('\n');
      return { buffer: parts.pop() ?? '', lines: parts };
    },
    { buffer: '', lines: [] as string[] },
  ),
  filter((s) => s.lines.length > 0),
  map((s) => s.lines),
);

customLines$
  .subscribe((lines) => lines.forEach((line) => console.log('行:', line)));
Copy

組み込みシェルでよく使う \r\n も、既定の lines$ 側で扱います。上のパターンは、独自の分割ルールが必要なとき専用です。

プロンプトや改行なしのデータ: lines$ は改行が揃うまで emit しません。デバイスが 改行（\n / \r\n）なしでプロンプトや行の途中だけを送る場合は、lines$ を待たず receive$ でバッファを積み上げる（下の「readUntil パターン」）方が向いています。

接続中フラグ（state$ narrowing）

ボタンの有効／無効など「接続済みかどうか」だけが欲しい場合も、state$ を canonical API として使います。state.status === SerialSessionStatus.Connected で narrowing すれば、TypeScript が connected state の型情報を保持します。boolean だけ必要な UI では state$ から derive してください。

import { distinctUntilChanged, map } from 'rxjs';
import { SerialSessionStatus } from '@gurezo/web-serial-rxjs';

const isConnected$ = session.state$.pipe(
  map((state) => state.status === SerialSessionStatus.Connected),
  distinctUntilChanged(),
);

isConnected$.subscribe((isOpen) => {
  // 操作の有効化など
});
Copy

isConnected$ は v3.x で非推奨です。多段階の UI では下記の state$ 駆動の UI のように state$ 全体を使う方が分かりやすいです。

RxJS pipeline 内で portInfo など connected 専用フィールドにアクセスする場合は、filter() と isConnectedSessionState を組み合わせてください。inline の status 比較では TypeScript の narrowing は行われません。

import { filter } from 'rxjs';
import { isConnectedSessionState } from '@gurezo/web-serial-rxjs';

session.state$
  .pipe(filter(isConnectedSessionState))
  .subscribe((state) => {
    console.log(state.portInfo);
  });
Copy

1 行送信（sendLine / sendLine$ 相当）

対話シェルでは CRLF 終端の 1 行を期待することが多いです。ライブラリに API を増やさず、send$ の薄いラッパーにします。

const sendLine = (line: string) => session.send$(`${line}\r\n`);

sendLine('ls -al').subscribe({
  error: (error) => console.error('送信失敗:', error),
});
Copy

相手が LF のみを期待する UART プロトコルなら \n に置き換えます。いずれも文字列は同様に UTF-8 エンコードされます。

順序保証のある送信

send$ は内部 FIFO キューで直列化済みなので、並行購読でも呼び出し順に配送されます。

import { from, concatMap } from 'rxjs';

const commands = ['help\n', 'status\n', 'version\n'];
from(commands)
  .pipe(concatMap((cmd) => session.send$(cmd)))
  .subscribe({
    error: (error) => console.error('コマンド失敗:', error),
  });
Copy

readUntil パターン（readUntil$ / プロンプト待ち）

receive$ はチャンク単位であり、メッセージ単位ではありません。readUntil はバッファに蓄積したテキストが述語（区切り・正規表現・プロンプト）を満たすまで待ちます。receive$ はホットで過去チャンクを後から購読しただけでは再現されないため、相手がすぐ応答するなら send$ する前に receive$ 側の待ち受けを開始してください。

import { firstValueFrom, scan, filter, map, take, timeout } from 'rxjs';

async function readUntil(
  predicate: (buffer: string) => boolean,
  options: { timeoutMs?: number } = {},
): Promise<string> {
  const timeoutMs = options.timeoutMs ?? 5000;
  const match$ = session.receive$.pipe(
    scan((buffer, chunk) => buffer + chunk, ''),
    filter(predicate),
    map((buffer) => buffer),
    take(1),
    timeout(timeoutMs),
  );
  return firstValueFrom(match$);
}

const sendLine = (line: string) => session.send$(`${line}\r\n`);

// 例: ログインプロンプトを待ってから資格情報を送る（説明用）
await readUntil((buf) => /login:\s*$/im.test(buf));
await firstValueFrom(sendLine('pi'));
await readUntil((buf) => /password:\s*$/im.test(buf));
await firstValueFrom(sendLine('raspberry'));
Copy

コマンド送信＋プロンプト待ちも同じ蓄積パイプラインです。先に購読（firstValueFrom で Promise を取得）してから送ります。

async function query(cmd: string, prompt = /device>\s$/): Promise<string> {
  const response$ = session.receive$.pipe(
    scan((buffer, chunk) => buffer + chunk, ''),
    filter((buffer) => prompt.test(buffer)),
    map((buffer) => buffer),
    take(1),
    timeout(5000),
  );
  const responsePromise = firstValueFrom(response$);
  await firstValueFrom(session.send$(cmd));
  return responsePromise;
}
Copy

waitForState

connect$ や disconnect$ のあと、特定のライフサイクル status（例: SerialSessionStatus.Connected や SerialSessionStatus.Idle）が立つまで await したい場合があります。state$ に filter・take(1)・必要なら timeout を載せます。

import { filter, take, firstValueFrom, timeout } from 'rxjs';
import { SerialSessionStatus } from '@gurezo/web-serial-rxjs';

async function waitForState(
  target: (typeof SerialSessionStatus)[keyof typeof SerialSessionStatus],
  options: { timeoutMs?: number } = {},
): Promise<void> {
  const timeoutMs = options.timeoutMs ?? 30_000;
  await firstValueFrom(
    session.state$.pipe(
      filter((s) => s.status === target),
      take(1),
      timeout(timeoutMs),
    ),
  );
}

// 例: connect$ 成功後はすでに 'connected' だが、他の非同期処理との整合や
// タイムアウトを明示したいときに使う
await firstValueFrom(session.connect$());
await waitForState(SerialSessionStatus.Connected, { timeoutMs: 5000 });
Copy

state$ 駆動の UI

真偽値を自分で追うのではなく、UI 遷移は state$ で駆動します。

import { SerialSessionStatus } from '@gurezo/web-serial-rxjs';

session.state$.subscribe((state) => {
  switch (state.status) {
    case SerialSessionStatus.Idle:
      showConnectButton();
      break;
    case SerialSessionStatus.Connecting:
    case SerialSessionStatus.Disconnecting:
      showSpinner();
      break;
    case SerialSessionStatus.Connected:
      showSendUi(state.portInfo);
      break;
    case SerialSessionStatus.Error:
      showErrorBanner(state.error);
      break;
    case SerialSessionStatus.Unsupported:
      showUnsupportedBanner();
      break;
  }
});
Copy

一元化されたエラーハンドリング

errors$ が主エラーチャネルです。connect$().subscribe({ error }) で受け取るのは errors$ に流れるものと同一の SerialError インスタンスです。

import { SerialErrorCode } from '@gurezo/web-serial-rxjs';

session.errors$.subscribe((error) => {
  if (error.is(SerialErrorCode.READ_FAILED)) {
    // 致命的エラー — session はすでに 'error' 状態でポートもテアダウン済み
    console.error(error.context.cause);
    session.disconnect$().subscribe();
  }
});
Copy

致命的エラー時の再接続

致命的エラーは state$ を { status: 'error', error } に遷移させるので、再接続ポリシーは素直に書けます。

import { filter, concatMap } from 'rxjs';
import { SerialSessionStatus } from '@gurezo/web-serial-rxjs';

session.state$
  .pipe(
    filter((state) => state.status === SerialSessionStatus.Error),
    concatMap(() => session.disconnect$()),
    concatMap(() => session.connect$()),
  )
  .subscribe({
    error: (error) => console.error('再接続失敗:', error),
  });
Copy

フレームワーク統合

各 example は典型的な統合例を示しています。

• Angular: ReplaySubject<SerialSession> を switchMap で展開し、state$ / receive$ / errors$ を service から公開
• Vue 3: 同じストリームを ref にミラーする composable
• React: session を ref に保持しつつ、ストリームを useState にミラーする hook
• Svelte: derived store でセッションをラップ
• Vanilla JS/TS: そのまま subscribe