Server Streaming

1 request เข้า, หลาย response ออก

Server streaming คือรูปแบบที่ client ส่ง request เดียวแล้ว server ตอบกลับเป็น ลำดับ ของ message บน stream ที่เปิดค้างไว้ client อ่านทีละอันจนกว่า server จะบอกว่าจบ

รูปแบบนี้เหมาะกับกรณีที่คำตอบไม่ใช่ค่าเดียวแต่เป็นชุด: ดู event ตอนที่เกิด, stream result set ขนาดใหญ่ทีละ row หรือรายงาน progress ของงานที่รันนาน

sequenceDiagram
  participant C as Client
  participant S as Server
  C->>S: WatchPrices(symbol: "ACME")
  S-->>C: Price 10.20
  S-->>C: Price 10.24
  S-->>C: Price 10.19
  S-->>C: ... (จนกว่า server จะปิด)
  Note over C,S: server จบ stream, loop ของ client ก็จบ

Server streaming: 1 request, stream ของ response

contract

คุณ mark response เป็น stream ด้วย keyword stream ส่วน request ยังเป็นตัวเดียว:

syntax = "proto3";
package market.v1;

message WatchRequest {
  string symbol = 1;
}

message Price {
  string symbol = 1;
  double value  = 2;
  int64  ts     = 3;
}

service Market {
  // One request, a stream of Price responses.
  rpc WatchPrices(WatchRequest) returns (stream Price);
}

keyword stream ตัวเดียวเปลี่ยน code ที่ generate ออกมา: แทนที่จะ return Price ตัวเดียว server ได้ stream ไว้ส่งหลายอัน และ client ได้ iterator ไว้อ่านหลายอัน

Server: ส่งใน loop

server implementation รับ request แล้วเรียก “send” ซ้ำ ๆ เมื่อ method return stream ก็ถูกปิด

func (s *server) WatchPrices(req *marketv1.WatchRequest, stream marketv1.Market_WatchPricesServer) error {
    for i := 0; i < 5; i++ {
        price := &marketv1.Price{Symbol: req.Symbol, Value: 10 + float64(i)*0.01}
        if err := stream.Send(price); err != nil {
            return err // client went away
        }
        time.Sleep(time.Second)
    }
    return nil // returning closes the stream
}

def WatchPrices(self, request, context):
    for i in range(5):
        yield market_pb2.Price(symbol=request.symbol, value=10 + i * 0.01)
        time.sleep(1)
    # the generator ending closes the stream

function watchPrices(call: ServerWritableStream<WatchRequest, Price>) {
  let i = 0;
  const timer = setInterval(() => {
    call.write({ symbol: call.request.symbol, value: 10 + i * 0.01 });
    if (++i === 5) {
      clearInterval(timer);
      call.end(); // closes the stream
    }
  }, 1000);
}

Client: อ่านจนจบ

client เรียก call เดียวแล้ววน iterate response loop จบเมื่อ server ปิด stream

stream, _ := client.WatchPrices(ctx, &marketv1.WatchRequest{Symbol: "ACME"})
for {
    price, err := stream.Recv()
    if err == io.EOF {
        break // server closed the stream
    }
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println(price.Value)
}

for price in client.WatchPrices(market_pb2.WatchRequest(symbol="ACME")):
    print(price.value)
# loop ends when the server closes the stream

const call = client.watchPrices({ symbol: 'ACME' });
call.on('data', (price: Price) => console.log(price.value));
call.on('end', () => console.log('server closed the stream'));
call.on('error', (err) => console.error(err));

หมายเหตุด้านการออกแบบ

message มาตามลำดับ ภายใน stream เดียว gRPC รับประกันว่า client อ่าน message เรียงตามลำดับที่ server ส่งเป๊ะ
ตั้ง deadline server stream รันได้นาน client ก็ยังควรจำกัดด้วย deadline หรือ cancel เมื่อไม่สนใจแล้ว (อยู่ในบท deadline)
error มาได้กลาง stream หลังจากส่ง message ดี ๆ ไปหลายอัน server ก็ยังจบด้วย error status ได้ handle branch ที่เป็น error ของ receive loop ด้วย ไม่ใช่แค่ happy path
server ควรเคารพการ cancel ถ้า client หลุด Send จะ return error — หยุดผลิตแทนที่จะวนไม่รู้จบ