Consumer (read) protocol — plan & proposal

외부 클라이언트가 foundation Source Bus에서 한 source의 Sample 스트림을 받아오는 계약 설계안. 기존 ingest(쓰기) 계약과 in-process subscribe를 거의 대칭으로 변형 — foundation에 새로 추가할 것을 계획부터 정리했습니다.

1. 한눈에: 쓰기(있음) ↔ 읽기(제안)는 대칭

핵심: 메시지(Sample/StreamStateChanged)도, 버스 기능(subscribe 팬아웃)도 이미 있습니다. 빠진 건 "server-streaming RPC 한 개 + 그걸 subscribe에 잇는 thin servicer"뿐. 새로 만드는 양이 작고, ingest의 거울이라 컨벤션도 정해져 있습니다.

2. 참고자료 — 이미 있는 것 (재사용/거울의 기준)

service SourceIngestService {
  rpc Connect(stream ClientEvent) returns (stream RecorderCommand);
  rpc StreamSamples(stream SampleBatch) returns (StreamSamplesResponse);
}
// 둘 다 ingest 방향(클라→서버). 서버는 sink.
// server→client 로 Sample 흘리는 RPC는 없음.

message Sample {
  int64 timestamp_us = 1;
  uint64 sequence = 2;
  oneof payload { TactilePayload tactile=10; RobotPayload robot=11; MarkerPayload marker=12; }
}
message StreamStateChanged {
  string session_id=1; string source_id=2; StreamState state=3; string detail=4;
}
// StreamState: UNSPECIFIED|STARTED|STOPPED|FAILED

def subscribe(self, source_id: str)
    -> AsyncIterator[SourceSample | SourceEvent]
# bounded(256) per-subscriber 큐, 다중 구독 지원,
# 한 소스당 broadcast 런루프. 외부 호출 불가(같은 프로세스만).

# grpc_source._stream: 이미 pb2.Sample ↔ SourceSample 변환 코드 존재
# (ingest 수신측). 읽기측은 이 매핑을 역방향으로 재사용.
# preview: subscribe() 이터레이터를 HTTP(MJPEG)로 어댑트한 선례
#   → 우리는 같은 이터레이터를 gRPC server-stream으로 어댑트.

3. 제안 — 새 read 계약 foundation에 추가

// 별도 서비스 권장(ingest는 sink, read는 source — 방향이 반대).
service SourceReadService {
  rpc Subscribe(SubscribeRequest) returns (stream SubscribeEvent);
}

message SubscribeRequest {
  string source_id = 1;        // 받을 소스 (open 선행 필요 — Q4)
  string session_id = 2;       // 선택: 세션 범위 한정
}
// Q4: source가 open 안 됐으면 SourceNotFoundError(=StreamStateChanged FAILED).
//     open 트리거 방식(별도 단계 vs Subscribe-implies-open)은 합의 항목.

message SubscribeEvent {
  oneof event {
    Sample sample = 1;              // ← 재사용
    StreamStateChanged state = 2;   // ← 재사용 (FAILED/STOPPED)
  }
}

# 새 GrpcService — ingest 서버와 동일하게 register()로 붙음
class GrpcSourceReadServer(pb2_grpc.SourceReadServiceServicer, GrpcService):
  async def Subscribe(self, req, ctx):
    async for item in self._session.subscribe(req.source_id):
      if isinstance(item, SourceEvent):
        yield SubscribeEvent(state=_to_state(item)); continue
      yield SubscribeEvent(sample=_to_pb_sample(item))
# _to_pb_sample = grpc_source._stream 매핑의 역방향(SourceSample→pb2.Sample)
# register(): add_SourceReadServiceServicer_to_server(self, server)

4. SDK 쪽 (우리, foundation 이후)

class SourceConsumer:   # SourceProducer의 대칭 (이름 이미 예약됨)
  @classmethod
  def connect(loop_addr, ...) -> SourceConsumer ...
  def subscribe(source_id: str) -> Iterator[RobotFrame | TactileFrame | MarkerFrame]:
      # Subscribe RPC 열고, SubscribeEvent.sample → 도메인 frame 디코딩,
      # StreamStateChanged(FAILED) → 도메인 예외. proto_mapping 역방향 추가.

5. 설계 결정 — 코드로 검증 + 합의 결과

6. 계획 단계

참고: module/foundation/.../source_ingest.proto, inbound/grpc_source.py(변환), outbound/in_memory_source_session.py(subscribe 팬아웃), collect preview(subscribe→HTTP 어댑트 선례). 계약은 foundation 소유 — 이 문서는 우리(SDK) 측 제안/계획이며 합의 후 foundation이 proto를 추가합니다. 상세 요청서: docs/foundation_request_consumer_rpc.md (PR #2).