Bidirectional Streaming

ทั้งสองฝั่งพูดพร้อมกัน

Bidirectional streaming เปิด stream เดียวที่ ทั้ง client และ server ส่ง message ได้ อิสระและพร้อมกัน ไม่มีฝั่งไหนต้องรออีกฝั่ง message ไหลไปทางไหนก็ได้เมื่อมีอะไรจะพูด

นี่คือรูปแบบสำหรับ session แบบโต้ตอบจริง ๆ: chat, state ของ multiplayer game, การแก้เอกสารร่วมกัน หรือ protocol ไหนก็ตามที่ request กับ response ของตัวเองสลับกันไปมาแทนที่จะเป็นจังหวะ lock-step

sequenceDiagram
  participant C as Client
  participant S as Server
  C->>S: join(room: "general")
  S-->>C: "Sam joined"
  C->>S: "hello everyone"
  S-->>C: "Alex: hi Sam"
  C->>S: "how's the deploy?"
  S-->>C: "Alex: green"
  Note over C,S: แต่ละฝั่งส่งเมื่อไรก็ได้; ลำดับเป็นแบบ per-direction

Bidirectional streaming: message อิสระ สลับกันไปมา

contract

ทั้ง request และ response มี keyword stream:

syntax = "proto3";
package chat.v1;

message ChatMessage {
  string from = 1;
  string text = 2;
  int64  ts   = 3;
}

service Chat {
  // Both sides stream ChatMessages independently.
  rpc Chat(stream ChatMessage) returns (stream ChatMessage);
}

ทั้งสองฝั่งรัน send loop และ receive loop

จุดต่างสำคัญจาก stream ทางเดียว: แต่ละฝั่งมักอ่านและเขียน พร้อมกัน การอ่านและเขียนบน stream เดียวจาก goroutine/thread/callback แยกกันนั้นปลอดภัยและเป็นเรื่องปกติ

stream, _ := client.Chat(ctx)

// receive loop (its own goroutine)
go func() {
    for {
        msg, err := stream.Recv()
        if err == io.EOF { return }
        if err != nil { log.Println(err); return }
        fmt.Printf("%s: %s\n", msg.From, msg.Text)
    }
}()

// send loop (main goroutine)
for _, text := range outgoing {
    stream.Send(&chatv1.ChatMessage{From: "me", Text: text})
}
stream.CloseSend() // done sending; server may still send

def outgoing():
    for text in messages:
        yield chat_pb2.ChatMessage(**{"from": "me", "text": text})

responses = client.Chat(outgoing())  # sending and receiving run concurrently
for msg in responses:
    print(f"{msg.from}: {msg.text}")

const call = client.chat();
call.on('data', (msg: ChatMessage) => console.log(`${msg.from}: ${msg.text}`));
call.on('end', () => console.log('server closed its side'));

for (const text of outgoing) call.write({ from: 'me', text });
call.end(); // done sending; the server may keep sending

server implementation ก็ทำแบบเดียวกัน: วน Recv แล้วเรียก Send เมื่อมีอะไรจะส่ง — บ่อยครั้งคือกระจาย message ที่รับมาออกไปให้ client อื่นที่ต่ออยู่

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

ลำดับเป็นแบบ per-direction message ที่ client ส่งไปถึง server ตามลำดับ; message ที่ server ส่งไปถึง client ตามลำดับ แต่สองทิศทางเป็นอิสระต่อกัน — ไม่มีลำดับรวมระหว่าง message ของ client กับของ server
half-close เป็นแบบข้างเดียว CloseSend (client) หรือการ return (server) จบการส่งของ ฝั่งนั้น ส่วนอีกฝั่ง stream ต่อได้ call จบสมบูรณ์ก็ต่อเมื่อทั้งสองทิศปิด
อ่านและเขียนพร้อมกัน อย่า block receive loop ไว้หลัง send loop ไม่งั้น flow-control window เต็มแล้ว deadlock ได้ (บทถัดไป)
นี่ไม่ใช่ message broker bidi stream เป็นแบบ point-to-point ระหว่าง client 1 ตัวกับ server 1 ตัว การกระจายไปหา client จำนวนมาก (chat room) เป็น application logic ที่อยู่ข้างบน มักมี pub/sub รองรับ