# 通知設計書 14-ActionCable::Channel::Streams#stream_from

## 概要

本ドキュメントは、ActionCableのストリーミング購読機能について、設計情報を記載するものである。

### 本通知の処理概要

ActionCable::Channel::Streams#stream_fromは、ActionCableチャンネル内でブロードキャストの購読を開始するためのメソッドである。指定した名前のブロードキャストチャンネルからのメッセージをリアルタイムで受信し、接続中のWebSocketクライアントに転送する。これにより、サーバーからクライアントへのプッシュ通知が可能になる。

**業務上の目的・背景**：WebSocketクライアントが特定のブロードキャストチャンネルを購読し、そのチャンネルに送信されたメッセージをリアルタイムで受信するために使用される。チャットルーム、通知フィード、ライブ更新など、複数のクライアントが同じデータストリームを共有する必要があるアプリケーションで必須の機能である。

**通知の送信タイミング**：クライアントがチャンネルを購読（subscribe）した際に、subscribedコールバック内でstream_fromが呼び出される。以降、そのブロードキャストチャンネルへのメッセージはすべてクライアントに転送される。

**通知の受信者**：stream_fromを呼び出したチャンネルインスタンスに紐づくWebSocketクライアント。

**通知内容の概要**：ブロードキャストされたメッセージ（通常はJSONデコードされたオブジェクト）がそのままクライアントに転送される。

**期待されるアクション**：クライアント側のJavaScriptでreceivedコールバックが発火し、メッセージに応じたUI更新等の処理が実行される。

## 通知種別

WebSocket通知（ストリーミング購読）

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | リアルタイムストリーミング（pubsub購読） |
| 優先度 | 高（WebSocket直接転送） |
| リトライ | なし（接続切断時は自動再購読） |

### 送信先決定ロジック

1. チャンネルのsubscribedコールバック内で`stream_from "channel_name"`を呼び出す
2. pubsubアダプターの`subscribe`メソッドで指定チャンネルを購読
3. 以降、そのチャンネルへのbroadcastはすべてこの接続に転送される

## 通知テンプレート

### WebSocket通知の場合

| 項目 | 内容 |
|-----|------|
| デコーディング | JSON（デフォルト）またはカスタムcoder |
| メッセージ形式 | ブロードキャストされたメッセージをそのまま転送 |
| フィルタリング | コールバックブロック指定時はカスタム処理可能 |

### メッセージ受信例

```javascript
// クライアント側
channel.received = function(data) {
  console.log("Received:", data);
  // { "message": "Hello", "user": "Alice" }
};
```

### 添付ファイル

| ファイル名 | 形式 | 条件 | 説明 |
|----------|------|------|------|
| N/A | - | - | WebSocket通知には添付ファイルの概念なし |

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| broadcasting | 購読するブロードキャストチャンネル名 | 呼び出し元で指定 | Yes |
| callback | カスタムメッセージハンドラ | ブロックまたは引数 | No |
| coder | デコーダー | デフォルトnil（JSONデコード無効） | No |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| チャンネル購読 | subscribed | 購読成功時 | クライアントのsubscribe実行 |
| アクション実行 | perform | 任意 | チャンネルアクション内でのstream_from呼び出し |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| unsubscribed? | チャンネルが購読解除されている場合はstream_fromを無視 |
| WebSocket切断 | 接続が切断された場合は自動的にストリーム停止 |

## 処理フロー

### 購読フロー

```mermaid
flowchart TD
    A[チャンネル購読: subscribed] --> B[stream_from 呼び出し]
    B --> C{unsubscribed?}
    C -->|Yes| D[処理スキップ]
    C -->|No| E[broadcasting文字列化]
    E --> F[defer_subscription_confirmation!]
    F --> G[stream_handler生成]
    G --> H[streams hashに登録]
    H --> I[event_loop.post]
    I --> J[pubsub.subscribe]
    J --> K{購読成功?}
    K -->|Yes| L[ensure_confirmation_sent]
    K -->|No| M[エラー処理]
    L --> N[ログ出力: is streaming from]
    N --> O[ストリーム開始]
    O --> P{broadcast受信}
    P -->|メッセージ到着| Q[worker_pool.async_invoke]
    Q --> R[stream_handler.call]
    R --> S[transmit to client]
    S --> P
```

## データベース参照・更新仕様

### 参照テーブル一覧

| テーブル名 | 用途 | 備考 |
|-----------|------|------|
| なし | - | stream_fromメソッド自体はDBを参照しない |

### 更新テーブル一覧

| テーブル名 | 操作 | 概要 |
|-----------|------|------|
| なし | - | stream_fromメソッド自体はDBを更新しない |

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| pubsub購読エラー | pubsubバックエンド接続失敗 | 接続を確認、リトライ |
| ハンドラ実行エラー | callbackブロック内で例外 | rescue_fromで例外処理 |
| 接続切断 | WebSocket接続が切れた | 自動的にストリーム解除 |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | なし（pubsubアダプターに依存） |
| リトライ間隔 | N/A |
| リトライ対象エラー | N/A |

## 配信設定

### レート制限

| 項目 | 内容 |
|-----|------|
| 1分あたり上限 | 設定なし（リアルタイムストリーミング） |
| 1日あたり上限 | 設定なし |

### 配信時間帯

制限なし（24時間365日）

## セキュリティ考慮事項

- ブロードキャストチャンネル名にユーザーIDなどを含め、他ユーザーのストリームにアクセスできないようにする
- subscribedコールバック内で適切な認証・認可チェックを行う
- 接続時のOrigin検証を有効にする
- reject呼び出しで不正な購読を拒否する

## 備考

- stream_fromはsubscribedコールバック以外でも呼び出し可能だが、通常はsubscribedで使用する
- callbackブロックを指定すると、デフォルトの転送動作を上書きできる
- stop_stream_fromで個別のストリームを停止、stop_all_streamsで全ストリームを停止可能
- on_unsubscribeコールバックで自動的にstop_all_streamsが呼ばれる

---

## コードリーディングガイド

本通知を理解するために参照すべきファイルと、推奨する読み解き順序を以下に示す。

### 推奨読解順序

#### Step 1: データ構造を理解する

ストリームの管理構造を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | streams.rb | `actioncable/lib/action_cable/channel/streams.rb` | Streamsモジュール全体、@_streams hash |

**読解のコツ**: @_streamsはbroadcasting名をキー、handlerを値とするHashで、購読中のストリームを管理している。

#### Step 2: エントリーポイントを理解する

stream_fromメソッドの実装を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | streams.rb | `actioncable/lib/action_cable/channel/streams.rb` | stream_fromメソッド |

**主要処理フロー**:
1. **行90-91**: `stream_from(broadcasting, callback = nil, coder: nil, &block)` - エントリーポイント
2. **行91**: `return if unsubscribed?` - 購読解除済みなら何もしない
3. **行93**: `broadcasting = String(broadcasting)` - 文字列化
4. **行96**: `defer_subscription_confirmation!` - 確認を遅延
5. **行100**: `handler = worker_pool_stream_handler(...)` - ハンドラ生成
6. **行101**: `streams[broadcasting] = handler` - ストリームを登録
7. **行103-108**: `event_loop.post`内でpubsub.subscribeを実行

#### Step 3: ストリームハンドラを理解する

メッセージ受信時の処理を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | streams.rb | `actioncable/lib/action_cable/channel/streams.rb` | worker_pool_stream_handler、stream_handler、default_stream_handler |

**主要処理フロー**:
- **行163-168**: `worker_pool_stream_handler` - ワーカープールで非同期実行するラッパー
- **行175-181**: `stream_handler` - ユーザーハンドラまたはデフォルトハンドラを選択
- **行191-194**: `default_stream_handler` - JSON デコードしてtransmit

#### Step 4: ストリーム停止を理解する

ストリーム解除の処理を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | streams.rb | `actioncable/lib/action_cable/channel/streams.rb` | stop_stream_from、stop_all_streams |

**主要処理フロー**:
- **行123-129**: `stop_stream_from` - 指定ブロードキャストの購読解除
- **行137-142**: `stop_all_streams` - 全ストリームの購読解除
- **行81**: `on_unsubscribe :stop_all_streams` - 購読解除時に自動呼び出し

### プログラム呼び出し階層図

```
stream_from(broadcasting, callback, coder:, &block)
    │
    ├─ return if unsubscribed? (行91)
    │
    ├─ String(broadcasting) (行93)
    │
    ├─ defer_subscription_confirmation! (行96)
    │
    ├─ worker_pool_stream_handler (行100)
    │      │
    │      └─ stream_handler (行164)
    │             │
    │             ├─ user_handler指定あり
    │             │      └─ stream_decoder(user_handler, coder:) (行177)
    │             │
    │             └─ user_handler指定なし
    │                    └─ default_stream_handler (行179)
    │                           └─ stream_transmitter(stream_decoder()) (行193)
    │
    ├─ streams[broadcasting] = handler (行101)
    │
    └─ connection.server.event_loop.post (行103-108)
           │
           └─ pubsub.subscribe(broadcasting, handler, success_callback)
                  │
                  ├─ Redis SUBSCRIBE (Redisアダプター)
                  ├─ PG LISTEN (PostgreSQLアダプター)
                  └─ 直接登録 (Inlineアダプター)
                         │
                         └─ success_callback
                                ├─ ensure_confirmation_sent (行105)
                                └─ logger.info (行106)
```

### データフロー図

```
[入力]                          [処理]                              [出力]

broadcasting ─────────────▶ String() ───▶ @_streams[broadcasting]

callback/block ───────────▶ stream_handler ───▶ handler lambda

coder ────────────────────▶ stream_decoder ───▶ デコード処理

                          │
                          ▼
                   pubsub.subscribe
                          │
                          ▼
               ┌─────────────────────┐
               │   pubsub backend    │
               │ (Redis/PG/Inline)   │
               └─────────────────────┘
                          │
                  broadcast受信
                          │
                          ▼
               ┌─────────────────────┐
               │  worker_pool        │
               │  async_invoke       │
               └─────────────────────┘
                          │
                          ▼
               ┌─────────────────────┐
               │  stream_handler     │
               │  → decode → transmit│
               └─────────────────────┘
                          │
                          ▼
               ┌─────────────────────┐
               │  WebSocket Client   │
               │  (received callback)│
               └─────────────────────┘
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| streams.rb | `actioncable/lib/action_cable/channel/streams.rb` | ソース | Streamsモジュール |
| base.rb | `actioncable/lib/action_cable/channel/base.rb` | ソース | チャンネル基底クラス |
| broadcasting.rb | `actioncable/lib/action_cable/server/broadcasting.rb` | ソース | ブロードキャスト機能 |
| redis.rb | `actioncable/lib/action_cable/subscription_adapter/redis.rb` | ソース | Redis購読アダプター |
| postgresql.rb | `actioncable/lib/action_cable/subscription_adapter/postgresql.rb` | ソース | PostgreSQL購読アダプター |
| connection.rb | `actioncable/lib/action_cable/connection/base.rb` | ソース | WebSocket接続基底クラス |
