双方向ストリーミングの実装

var _ = Service("logger", func() {
    Method("monitor", func() {
        StreamingPayload(LogFilter)
        StreamingResult(LogEntry)
        HTTP(func() {
            GET("/logs/monitor")
            Response(StatusOK)
        })
        GRPC(func() {})
    })
})

// クライアントが満たすべきインターフェース
type MonitorClientStream interface {
    // "LogFilter"のインスタンスをストリーミング
    Send(*LogFilter) error
    // ストリーム内の次の結果を返す
    Recv() (*LogEntry, error)
    // ストリームを終了
    Close() error
}

func monitorLogs(client logger.Client, initialFilter *LogFilter) error {
    stream, err := client.Monitor(context.Background())
    if err != nil {
        return fmt.Errorf("モニターストリームの開始に失敗: %w", err)
    }
    defer stream.Close()

    // ログ受信を処理するゴルーチンを開始
    go func() {
        for {
            logEntry, err := stream.Recv()
            if err == io.EOF {
                return
            }
            if err != nil {
                log.Printf("ログエントリの受信エラー: %v", err)
                return
            }
            processLogEntry(logEntry)
        }
    }()

    // 初期フィルターを送信
    if err := stream.Send(initialFilter); err != nil {
        return fmt.Errorf("初期フィルターの送信に失敗: %w", err)
    }

    // 条件に基づいてフィルターを更新
    for {
        newFilter := waitForFilterUpdate()
        if err := stream.Send(newFilter); err != nil {
            return fmt.Errorf("フィルターの更新に失敗: %w", err)
        }
    }
}

// サーバーが満たすべきインターフェース
type MonitorServerStream interface {
    // "LogEntry"のインスタンスをストリーミング
    Send(*LogEntry) error
    // ストリーム内の次のフィルターを返す
    Recv() (*LogFilter, error)
    // ストリームを終了
    Close() error
}

func (s *loggerSvc) Monitor(ctx context.Context, stream logger.MonitorServerStream) error {
    // フィルター更新を処理するゴルーチンを開始
    filterCh := make(chan *LogFilter, 1)
    go func() {
        defer close(filterCh)
        for {
            filter, err := stream.Recv()
            if err == io.EOF {
                return
            }
            if err != nil {
                log.Printf("フィルター更新の受信エラー: %v", err)
                return
            }
            filterCh <- filter
        }
    }()

    // ログの処理とフィルターの適用のメインループ
    var currentFilter *LogFilter
    for {
        select {
        case filter, ok := <-filterCh:
            if !ok {
                // チャネルが閉じられた、処理を停止
                return nil
            }
            currentFilter = filter
        case <-ctx.Done():
            // コンテキストがキャンセルされた、処理を停止
            return ctx.Err()
        default:
            if currentFilter != nil {
                logEntry := s.getNextMatchingLog(currentFilter)
                if err := stream.Send(logEntry); err != nil {
                    return fmt.Errorf("ログエントリの送信エラー: %w", err)
                }
            }
        }
    }
}