クライアントサイドストリーミングの実装

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

// クライアントが満たすべきインターフェース
type UploadClientStream interface {
    // "LogEntry"のインスタンスをストリーミング
    Send(*LogEntry) error
    // ストリームを終了
    Close() error
}

func uploadLogEntries(client *logger.Client, logEntries []*LogEntry) error {
    stream, err := client.Upload(context.Background())
    if err != nil {
        return fmt.Errorf("アップロードストリームの開始に失敗: %w", err)
    }

    for _, logEntry := range logEntries {
        if err := stream.Send(logEntry); err != nil {
            return fmt.Errorf("ログエントリの送信に失敗: %w", err)
        }
    }

    if err := stream.Close(); err != nil {
        return fmt.Errorf("ストリームの終了に失敗: %w", err)
    }

    return nil
}

// サーバーがストリームを受信するために使用するインターフェース
type UploadServerStream interface {
    // Recvはストリーム内の次のペイロードを返します
    Recv() (*LogEntry, error)
    // Closeはストリームを終了します
    Close() error
}

func (s *loggerSvc) Upload(ctx context.Context, stream logger.UploadServerStream) error {
    for {
        logEntry, err := stream.Recv()
        if err == io.EOF {
            // ストリームが終了
            return nil
        }
        if err != nil {
            return fmt.Errorf("ログエントリの受信エラー: %w", err)
        }

        // 受信したログエントリを処理
        if err := s.processLogEntry(logEntry); err != nil {
            return fmt.Errorf("ログエントリの処理エラー: %w", err)
        }
    }
}