การสตรีมฝั่งเซิร์ฟเวอร์และไคลเอนต์

ชนิดการสตรีมของ gRPC

การสตรีมจากเซิร์ฟเวอร์ใน .proto

service EventService {
    rpc ListEvents (ListEventsRequest) returns (stream Event);
}

การนำการสตรีมจากเซิร์ฟเวอร์ไปใช้งาน

func (s *server) ListEvents(req *pb.ListEventsRequest, stream pb.EventService_ListEventsServer) error {
    for _, event := range s.events {
        if err := stream.Send(event); err != nil { return err }
    }
    return nil
}

ไคลเอนต์รับสตรีมจากเซิร์ฟเวอร์

stream, err := client.ListEvents(ctx, &pb.ListEventsRequest{})
for {
    event, err := stream.Recv()
    if err == io.EOF { break }
    if err != nil { log.Fatal(err) }
    fmt.Println(event)
}

การสตรีมจากไคลเอนต์ใน .proto

service UploadService {
    rpc Upload (stream Chunk) returns (UploadResponse);
}

การนำการสตรีมจากไคลเอนต์ไปใช้งาน

stream, _ := client.Upload(ctx)
for _, chunk := range chunks {
    stream.Send(chunk)
}
resp, err := stream.CloseAndRecv()

เซิร์ฟเวอร์รับสตรีมจากไคลเอนต์

func (s *server) Upload(stream pb.UploadService_UploadServer) error {
    for {
        chunk, err := stream.Recv()
        if err == io.EOF {
            return stream.SendAndClose(&pb.UploadResponse{Received: total})
        }
        if err != nil { return err }
        total += len(chunk.Data)
    }
}

การสตรีมสองทิศทาง

service ChatService {
    rpc Chat (stream Message) returns (stream Message);
}

การนำเซิร์ฟเวอร์แบบสองทิศทางไปใช้งาน

func (s *server) Chat(stream pb.ChatService_ChatServer) error {
    for {
        msg, err := stream.Recv()
        if err == io.EOF { return nil }
        if err != nil { return err }
        stream.Send(&pb.Message{Text: "echo: " + msg.Text})
    }
}

การควบคุมการไหล

การยกเลิกบริบท

ตรวจสอบความเข้าใจอย่างรวดเร็ว

สรุป: การสตรีม gRPC