เริ่มต้นใช้งาน gRPC-Rust - การสตรีม

1. บทนำ

ใน Codelab นี้ คุณจะได้ใช้ gRPC-Rust เพื่อสร้างไคลเอ็นต์และเซิร์ฟเวอร์ที่เป็นรากฐานของแอปพลิเคชันการกำหนดเส้นทางที่เขียนด้วย Rust

เมื่อจบบทแนะนำ คุณจะมีไคลเอ็นต์ที่เชื่อมต่อกับเซิร์ฟเวอร์ระยะไกลโดยใช้ gRPC เพื่อรับข้อมูลเกี่ยวกับฟีเจอร์ในเส้นทางของไคลเอ็นต์ สร้างข้อมูลสรุปของเส้นทางของไคลเอ็นต์ และแลกเปลี่ยนข้อมูลเส้นทาง เช่น ข้อมูลอัปเดตการจราจรกับเซิร์ฟเวอร์และไคลเอ็นต์อื่นๆ

บริการนี้กำหนดไว้ในไฟล์ Protocol Buffers ซึ่งจะใช้เพื่อสร้างโค้ด Boilerplate สำหรับไคลเอ็นต์และเซิร์ฟเวอร์เพื่อให้สื่อสารกันได้ ซึ่งจะช่วยประหยัดเวลาและความพยายามในการใช้ฟังก์ชันการทำงานดังกล่าว

โค้ดที่สร้างขึ้นนี้จะจัดการความซับซ้อนของการสื่อสารระหว่างเซิร์ฟเวอร์และไคลเอ็นต์ รวมถึงการซีเรียลไลซ์และการดีซีเรียลไลซ์ข้อมูล

สิ่งที่คุณจะได้เรียนรู้

  • วิธีใช้ Protocol Buffers เพื่อกำหนด API ของบริการ
  • วิธีสร้างไคลเอ็นต์และเซิร์ฟเวอร์ที่ใช้ gRPC จากคำจำกัดความของ Protocol Buffers โดยใช้การสร้างโค้ดอัตโนมัติ
  • ความเข้าใจเกี่ยวกับการสื่อสารแบบสตรีมมิงระหว่างไคลเอ็นต์กับเซิร์ฟเวอร์ด้วย gRPC

Codelab นี้เหมาะสำหรับนักพัฒนาแอป Rust ที่เพิ่งเริ่มใช้ gRPC หรือต้องการทบทวน gRPC หรือผู้ที่สนใจสร้างระบบแบบกระจาย ไม่จำเป็นต้องมีประสบการณ์การใช้ gRPC มาก่อน

2. ก่อนเริ่มต้น

ข้อกำหนดเบื้องต้น

ตรวจสอบว่าคุณได้ติดตั้งสิ่งต่อไปนี้แล้ว

รับโค้ด

Codelab นี้มีโครงสร้างพื้นฐานของซอร์สโค้ดของแอปพลิเคชันให้คุณกรอกข้อมูลให้สมบูรณ์ เพื่อให้คุณไม่ต้องเริ่มต้นจากศูนย์ ขั้นตอนต่อไปนี้จะแสดงวิธีทำให้แอปพลิเคชันเสร็จสมบูรณ์ รวมถึงการใช้ปลั๊กอินคอมไพเลอร์บัฟเฟอร์โปรโตคอลเพื่อสร้างโค้ด gRPC เริ่มต้น

ขั้นแรก ให้สร้างไดเรกทอรีการทำงานของ Codelab แล้วใช้คำสั่ง cd เพื่อเข้าไปในไดเรกทอรีดังกล่าว

mkdir streaming-grpc-rust-getting-started && cd streaming-grpc-rust-getting-started

ดาวน์โหลดและแตกไฟล์ Codelab

curl -sL https://github.com/grpc-ecosystem/grpc-codelabs/archive/refs/heads/2026.tar.gz \
  | tar xvz --strip-components=4 \
  grpc-codelabs-2026/codelabs/grpc-rust-streaming/start_here

หรือคุณจะดาวน์โหลดไฟล์ .zip ที่มีเฉพาะไดเรกทอรี Codelab แล้วแตกไฟล์ด้วยตนเองก็ได้

ซอร์สโค้ดที่เสร็จสมบูรณ์แล้วพร้อมให้บริการบน GitHub หากคุณไม่ต้องการพิมพ์การใช้งาน

3. กำหนดข้อความและบริการ

ขั้นตอนแรกคือการกำหนดบริการ gRPC ของแอปพลิเคชัน เมธอด RPC รวมถึงประเภทข้อความคำขอและการตอบกลับโดยใช้ Protocol Buffers บริการของคุณจะมีสิ่งต่อไปนี้

  • เมธอด RPC ที่เรียกว่า ListFeatures, RecordRoute และ RouteChat ซึ่งเซิร์ฟเวอร์จะใช้และไคลเอ็นต์จะเรียก
  • ประเภทข้อความ Point, Feature, Rectangle, RouteNote และ RouteSummary ซึ่งเป็นโครงสร้างข้อมูลที่ไคลเอ็นต์และเซิร์ฟเวอร์แลกเปลี่ยนกันเมื่อเรียกเมธอดข้างต้น

เมธอด RPC และประเภทข้อความทั้งหมดนี้จะกำหนดไว้ในไฟล์ proto/routeguide.proto ของซอร์สโค้ดที่ให้มา

Protocol Buffers หรือที่เรียกกันโดยทั่วไปว่า protobufs ดูข้อมูลเพิ่มเติมเกี่ยวกับคำศัพท์ gRPC ได้ที่ แนวคิดหลัก สถาปัตยกรรม และวงจรชีวิตของ gRPC

กำหนดประเภทข้อความ

ก่อนอื่นมากำหนดข้อความที่จะใช้โดย RPC ของเรากัน ในไฟล์ proto/routeguide.proto ของซอร์สโค้ด ให้กำหนดประเภทข้อความ Point ก่อน Point แสดงถึงคู่พิกัดละติจูด-ลองจิจูดบนแผนที่ สำหรับ Codelab นี้ ให้ใช้จำนวนเต็มสำหรับพิกัด

message Point {
  int32 latitude = 1;
  int32 longitude = 2;
}

ตัวเลข 1 และ 2 เป็นหมายเลขรหัสที่ไม่ซ้ำกันสำหรับแต่ละช่องในโครงสร้าง message

จากนั้นกำหนดประเภทข้อความ Feature Feature ใช้ช่อง string สำหรับชื่อหรือที่อยู่ทางไปรษณีย์ของสิ่งใดสิ่งหนึ่งในตำแหน่งที่ระบุโดย Point

message Feature {
  // The name or address of the feature.
  string name = 1;

  // The point where the feature is located.
  Point location = 2;
}

จากนั้นกำหนดข้อความ Rectangle ซึ่งแสดงถึงสี่เหลี่ยมผืนผ้าละติจูด-ลองจิจูดที่แสดงเป็นจุด 2 จุดที่อยู่มุมตรงข้ามกัน "lo" และ "hi"

message Rectangle {
  // One corner of the rectangle.
  Point lo = 1;

  // The other corner of the rectangle.
  Point hi = 2;
}

รวมถึงข้อความ RouteNote ซึ่งแสดงถึงข้อความที่ส่งขณะอยู่ที่จุดหนึ่งๆ

message RouteNote {
  // The location from which the message is sent.
  Point location = 1;

  // The message to be sent.
  string message = 2;
}

นอกจากนี้ เรายังต้องมีข้อความ RouteSummary ด้วย ข้อความนี้จะได้รับเป็นการตอบกลับ RPC RecordRoute ซึ่งอธิบายไว้ในส่วนถัดไป โดยจะมีจำนวนจุดแต่ละจุดที่ได้รับ จำนวนฟีเจอร์ที่ตรวจพบ และระยะทางทั้งหมดที่เดินทางเป็นผลรวมสะสมของระยะทางระหว่างแต่ละจุด

message RouteSummary {
  // The number of points received.
  int32 point_count = 1;

  // The number of known features passed while traversing the route.
  int32 feature_count = 2;

  // The distance covered in metres.
  int32 distance = 3;

  // The duration of the traversal in seconds.
  int32 elapsed_time = 4;
}

กำหนดเมธอดของบริการ

ก่อนอื่นมากำหนดบริการของเราก่อน แล้วค่อยกำหนดข้อความในภายหลัง หากต้องการกำหนดบริการ ให้ระบุบริการที่มีชื่อในไฟล์ .proto ไฟล์ proto/routeguide.proto มีโครงสร้าง service ที่ชื่อว่า RouteGuide ซึ่งกำหนดเมธอดอย่างน้อย 1 รายการที่บริการของแอปพลิเคชันมีให้

กำหนดเมธอด RPC ภายในคำจำกัดความของบริการ โดยระบุประเภทคำขอและการตอบกลับ ในส่วนนี้ของ Codelab เราจะกำหนดสิ่งต่อไปนี้

ListFeatures

รับ Feature ที่มีอยู่ใน Rectangle ที่ระบุ ระบบจะสตรีมผลลัพธ์แทนที่จะแสดงผลพร้อมกัน (เช่น ในข้อความตอบกลับที่มีช่องซ้ำ) เนื่องจากสี่เหลี่ยมผืนผ้าอาจครอบคลุมพื้นที่ขนาดใหญ่และมีฟีเจอร์จำนวนมาก

ประเภทที่เหมาะสมสำหรับ RPC นี้คือ RPC แบบ สตรีมมิงฝั่งเซิร์ฟเวอร์ โดยไคลเอ็นต์จะส่งคำขอไปยังเซิร์ฟเวอร์และรับสตรีมเพื่ออ่านลำดับข้อความกลับ ไคลเอ็นต์จะอ่านจากสตรีมที่แสดงผลจนกว่าจะไม่มีข้อความเหลือ ดังที่คุณเห็นในตัวอย่าง เราจะระบุเมธอดสตรีมมิงฝั่งเซิร์ฟเวอร์โดยวางคีย์เวิร์ด stream ไว้หน้าประเภทการตอบกลับ

rpc ListFeatures(Rectangle) returns (stream Feature) {}

RecordRoute

ยอมรับสตรีม Point ในเส้นทางที่เดินทาง และแสดงผล RouteSummary เมื่อเดินทางเสร็จสมบูรณ์

RPC แบบ สตรีมมิงฝั่งไคลเอ็นต์ ดูเหมือนจะเหมาะสมในกรณีนี้ โดยไคลเอ็นต์จะเขียนลำดับข้อความและส่งไปยังเซิร์ฟเวอร์อีกครั้งโดยใช้สตรีมที่ให้มา เมื่อไคลเอ็นต์เขียนข้อความเสร็จแล้ว ไคลเอ็นต์จะรอให้เซิร์ฟเวอร์อ่านข้อความทั้งหมดและแสดงผลการตอบกลับ คุณระบุเมธอดสตรีมมิงฝั่งไคลเอ็นต์โดยวางคีย์เวิร์ด stream ไว้หน้าประเภทคำขอ

rpc RecordRoute(stream Point) returns (RouteSummary) {}

RouteChat

ยอมรับสตรีม RouteNote ที่ส่งขณะเดินทางในเส้นทางหนึ่งๆ ขณะเดียวกันก็รับ RouteNote อื่นๆ (เช่น จากผู้ใช้รายอื่น)

นี่เป็นกรณีการใช้งานที่เหมาะสมสำหรับ การสตรีมมิงแบบสองทิศทาง RPC แบบสตรีมมิงแบบสองทิศทางจะให้ทั้ง 2 ฝ่ายส่งลำดับข้อความโดยใช้สตรีมแบบอ่าน-เขียน สตรีมทั้ง 2 รายการจะทำงานแยกกัน ดังนั้นไคลเอ็นต์และเซิร์ฟเวอร์จึงอ่านและเขียนได้ตามลำดับที่ต้องการ

เช่น เซิร์ฟเวอร์อาจรอรับข้อความทั้งหมดจากไคลเอ็นต์ก่อนที่จะเขียนการตอบกลับ หรืออาจอ่านข้อความแล้วเขียนข้อความ หรืออ่านและเขียนสลับกัน

ระบบจะเก็บลำดับข้อความในแต่ละสตรีมไว้ คุณระบุเมธอดประเภทนี้โดยวางคีย์เวิร์ด stream ไว้หน้าทั้งคำขอและการตอบกลับ

rpc RouteChat(stream RouteNote) returns (stream RouteNote) {}

4. สร้างโค้ดไคลเอ็นต์และเซิร์ฟเวอร์

เราได้ให้โค้ดที่สร้างขึ้นจากไฟล์ .proto ในไดเรกทอรี generated/ แก่คุณแล้ว ซึ่งรวมถึงการเพิ่มทั้งหมดที่คุณทำไว้ข้างต้น อย่างไรก็ตาม เราขอใช้เวลาสักครู่เพื่ออธิบายวิธีการสร้างโค้ด

ไฟล์ .proto อธิบายโครงสร้างและฟังก์ชันทั้งหมดที่ไคลเอ็นต์หรือเซิร์ฟเวอร์ใช้ เราใช้สคริปต์การสร้าง Cargo (build.rs) พร้อมกับกล่อง grpc-protobuf-build เพื่อสร้างโค้ดนี้โดยอัตโนมัติ

ใน Cargo.toml เราได้เพิ่ม grpc-protobuf-build เป็นการขึ้นต่อกันในการสร้างแล้ว

ใน build.rs เราจะกำหนดค่า grpc_protobuf_build::CodeGen เพื่อคอมไพล์ proto/routeguide.proto ลงในไดเรกทอรี generated/ บรรทัดสำคัญมีดังนี้

grpc_protobuf_build::CodeGen::new()
    .include("proto")
    .input("routeguide.proto")
    .output_dir("generated")
    .compile()
    .unwrap();

คำสั่งนี้จะเรียกการสร้างโค้ดของกล่อง grpc_protobuf_build โดยส่ง routeguide.proto ไปยังกล่องดังกล่าว เราได้ใส่โค้ดนี้ไว้ในโค้ดบางส่วนเพื่อให้ทำงานเฉพาะเมื่อมีการส่งแฟล็กฟีเจอร์เท่านั้น เพื่อให้ระบบสร้างโค้ดใหม่เมื่อคุณต้องการเท่านั้น คุณไม่จำเป็นต้องเรียกใช้โค้ดนี้ในตอนนี้ เนื่องจากเราได้สร้างโค้ดให้คุณแล้ว

cargo build --bin routeguide-server --features regenerate_proto

เมื่อคุณเรียกใช้ cargo build, build.rs จะคอมไพล์คำจำกัดความของบัฟเฟอร์โปรโตคอลลงในไดเรกทอรี generate/ ซึ่งรวมถึง

  • คำจำกัดความของโครงสร้างสำหรับประเภทข้อความ Point และ Feature
  • ลักษณะการทำงานของบริการ Tonic ที่เราจะต้องใช้สำหรับเซิร์ฟเวอร์: route_guide_server::RouteGuide
  • ประเภทไคลเอ็นต์ gRPC-Rust ที่เราจะใช้เพื่อเรียกเซิร์ฟเวอร์: route_guide_client::RouteGuideClient<T>

ดูข้อมูลเพิ่มเติมได้ที่คู่มือ protoc-gen-rust-grpc

จากนั้นเราจะใช้เมธอดของบริการในเซิร์ฟเวอร์

5. ใช้บริการ

ก่อนอื่นมาดูวิธีสร้างเซิร์ฟเวอร์ RouteGuide กัน การทำให้บริการ RouteGuide ทำงานได้นั้นมี 2 ส่วนดังนี้

  • การใช้อินเทอร์เฟซของบริการที่สร้างขึ้นจากคำจำกัดความของบริการ: การ "ทำงาน" จริงของบริการ
  • การเรียกใช้เซิร์ฟเวอร์ gRPC เพื่อรอรับคำขอจากไคลเอ็นต์และส่งคำขอไปยังการใช้งานเมธอดที่ถูกต้อง

ใน src/server/server.rs เราสามารถนำโค้ดที่สร้างขึ้นมาใช้ได้ผ่านแมโคร include_generated_proto! ของ gRPC และนำเข้าลักษณะการทำงาน RouteGuide และ Point

mod grpc_pb {
    grpc::include_generated_proto!("generated", "routeguide");
}

pub use grpc_pb::{
    route_guide_server::{RouteGuideServer, RouteGuide},
    Point, Feature, Rectangle, RouteNote, RouteSummary
};

เราสามารถเริ่มต้นด้วยการกำหนดโครงสร้างเพื่อแสดงถึงบริการของเรา โดยทำได้ใน src/server/server.rs ในตอนนี้

#[derive(Debug)]
pub struct RouteGuideService {
    features: Vec<Feature>,
}

ตอนนี้เราต้องใช้ลักษณะภาษา route_guide_server::RouteGuide จากโค้ดที่สร้างขึ้น

ใช้ RouteGuide

เราต้องใช้อินเทอร์เฟซ RouteGuide ที่สร้างขึ้น การใช้งานจะมีลักษณะดังนี้ ซึ่งอยู่ในเทมเพลตแล้ว

#[tonic::async_trait]
impl RouteGuide for RouteGuideService {
    async fn list_features(
        &self,
        request: Request<Rectangle>,
    ) -> Result<Response<ListFeaturesStream>, Status> {
        ...
    }

    async fn record_route(
        &self,
        request: Request<tonic::Streaming<Point>>,
    ) -> Result<Response<RouteSummary>, Status> {
        ...
    }

    async fn route_chat(
        &self,
        request: Request<tonic::Streaming<RouteNote>>,
    ) -> Result<Response<RouteChatStream>, Status> {
        ...
    }
}

มาดูรายละเอียดการใช้งาน RPC แต่ละรายการกัน

RPC แบบสตรีมมิงฝั่งเซิร์ฟเวอร์: ListFeatures

มาเริ่มที่ ListFeatures กัน นี่คือ RPC แบบสตรีมมิงฝั่งเซิร์ฟเวอร์ (ไคลเอ็นต์จะส่งข้อความ 1 รายการ เซิร์ฟเวอร์จะตอบกลับด้วยข้อความจำนวนมาก) ดังนั้นเราจึงต้องส่ง Feature หลายรายการกลับไปยังไคลเอ็นต์

async fn list_features(
        &self,
        request: Request<Rectangle>,
    ) -> Result<Response<ListFeaturesStream>, Status> {
    println!("ListFeatures = {:?}", request);

    let (tx, rx) = mpsc::channel(4);
    let features = self.features.clone();

    tokio::spawn(async move {
        for feature in &features[..] {
            if in_range(&feature.location().to_owned(), request.get_ref()) {
                println!("  => send {feature:?}");
                tx.send(Ok(feature.clone())).await.unwrap();
            }
        }
        println!(" /// done sending");
    });

    let output_stream = ReceiverStream::new(rx);
    Ok(Response::new(Box::pin(output_stream)))
}

ดังที่คุณเห็น เราได้รับออบเจ็กต์คำขอ (Rectangle ที่ไคลเอ็นต์ต้องการค้นหา Features) คราวนี้เราต้องแสดงผลสตรีมของค่า เราสร้างแชแนลและสร้างงานแบบไม่พร้อมกันใหม่ซึ่งเราจะทำการค้นหา โดยส่งฟีเจอร์ที่ตรงตามข้อจำกัดของเราไปยังแชแนล ระบบจะแสดงผลครึ่งหนึ่งของสตรีมของแชแนลไปยังผู้เรียก โดยห่อไว้ใน tonic::Response

RPC แบบสตรีมมิงฝั่งไคลเอ็นต์: RecordRoute

ตอนนี้มาดูสิ่งที่ซับซ้อนขึ้นเล็กน้อย นั่นคือเมธอด RecordRoute แบบสตรีมมิงฝั่งไคลเอ็นต์ ซึ่งเราจะได้รับสตรีม Points จากไคลเอ็นต์และแสดงผล RouteSummary รายการเดียวพร้อมข้อมูลเกี่ยวกับการเดินทาง เมธอดนี้จะได้รับสตรีมเป็นอินพุต ซึ่งเซิร์ฟเวอร์สามารถใช้เพื่ออ่านและเขียนข้อความได้ เมธอดนี้สามารถวนซ้ำข้อความจากไคลเอ็นต์โดยใช้เมธอด next() และแสดงผลการตอบกลับรายการเดียว

async fn record_route(
        &self,
        request: Request<tonic::Streaming<Point>>,
    ) -> Result<Response<RouteSummary>, Status> {
    println!("RecordRoute");
    let mut stream = request.into_inner();
    let mut summary = RouteSummary::default();
    let mut last_point = None;
    let now = Instant::now();

    while let Some(point) = stream.next().await {
        let point = point?;
        println!("  ==> Point = {point:?}");

        // Increment the point count
        summary.set_point_count(summary.point_count() + 1);

        // Find features
        for feature in &self.features[..] {
            if feature.location().latitude() == point.latitude() {
                if feature.location().longitude() == point.longitude(){
                    summary.set_feature_count(summary.feature_count() + 1);
                }
            }
        }

        // Calculate the distance
        if let Some(ref last_point) = last_point {
            let new_dist = summary.distance() + calc_distance(last_point, &point);
            summary.set_distance(new_dist);
        }
        last_point = Some(point);
    }
    summary.set_elapsed_time(now.elapsed().as_secs() as i32);
    Ok(Response::new(summary))
}

ในเนื้อหาของเมธอด เราใช้เมธอด next() ของสตรีมเพื่ออ่านคำขอของไคลเอ็นต์ซ้ำๆ ไปยังออบเจ็กต์คำขอ (ในกรณีนี้คือ Point) จนกว่าจะไม่มีข้อความเหลือ หากเป็น None แสดงว่าสตรีมยังใช้งานได้และอ่านต่อได้

RPC แบบสตรีมมิงแบบสองทิศทาง: RouteChat

สุดท้าย มาดู RPC แบบสตรีมมิงแบบสองทิศทาง RouteChat() กัน

async fn route_chat(
        &self,
        request: Request<tonic::Streaming<RouteNote>>,
    ) -> Result<Response<RouteChatStream>, Status> {
    println!("RouteChat");

    let mut notes: HashMap<(i32, i32), Vec<RouteNote>> = HashMap::new();
    let mut stream = request.into_inner();

    let output = async_stream::try_stream! {
        while let Some(note) = stream.next().await {
            let note = note?;
            let location = note.location();
            let key = (location.latitude(), location.longitude());
            let location_notes = notes.entry(key).or_insert(vec![]);
            location_notes.push(note);
            for note in location_notes {
                yield note.clone();
            }
        }
    };
    Ok(Response::new(Box::pin(output)))
}

คราวนี้เราจะได้รับสตรีมที่ใช้เพื่ออ่านและเขียนข้อความได้เช่นเดียวกับตัวอย่างการสตรีมมิงฝั่งไคลเอ็นต์ อย่างไรก็ตาม คราวนี้เราจะแสดงผลค่าผ่านสตรีมของเมธอดขณะที่ไคลเอ็นต์ยังคงเขียนข้อความลงในสตรีมข้อความ ไวยากรณ์สำหรับการอ่านและการเขียนที่นี่คล้ายกับเมธอดสตรีมมิงฝั่งไคลเอ็นต์มาก ยกเว้นว่าเซิร์ฟเวอร์จะแสดงผล RouteChatStream แม้ว่าแต่ละฝ่ายจะได้รับข้อความของอีกฝ่ายตามลำดับที่เขียนเสมอ แต่ทั้งไคลเอ็นต์และเซิร์ฟเวอร์สามารถอ่านและเขียนได้ตามลำดับใดก็ได้ เนื่องจากสตรีมทำงานแยกกันอย่างสมบูรณ์

เราสร้างสตรีมเอาต์พุตโดยใช้ try_stream! ซึ่งบ่งบอกว่าสตรีมอาจแสดงผลข้อผิดพลาด

เริ่มเซิร์ฟเวอร์

เมื่อใช้เมธอดนี้แล้ว เราต้องเริ่มเซิร์ฟเวอร์ gRPC ด้วยเพื่อให้ไคลเอ็นต์ใช้บริการของเราได้จริง กรอกข้อมูลใน main()

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let addr = "[::1]:10000".parse().unwrap();
    println!("RouteGuideServer listening on: {addr}");
    let route_guide = RouteGuideService {
        features: load(),
    };
    let svc = RouteGuideServer::new(route_guide);
    Server::builder().add_service(svc).serve(addr).await?;
    Ok(())
}

ต่อไปนี้คือสิ่งที่เกิดขึ้นใน main() ทีละขั้นตอน:

  1. ระบุพอร์ตที่เราต้องการใช้เพื่อรอรับคำขอจากไคลเอ็นต์
  2. สร้าง RouteGuideService ที่โหลดฟีเจอร์ไว้
  3. สร้างอินสแตนซ์ของเซิร์ฟเวอร์ gRPC โดยใช้ RouteGuideServer::new() โดยใช้บริการที่เราสร้างขึ้น
  4. ลงทะเบียนการใช้งานบริการกับเซิร์ฟเวอร์ gRPC
  5. เรียก serve() ในเซิร์ฟเวอร์พร้อมรายละเอียดพอร์ตเพื่อรอแบบบล็อกจนกว่าจะมีการหยุดกระบวนการ

6. สร้างไคลเอ็นต์

ในส่วนนี้ เราจะดูการสร้างไคลเอ็นต์ Rust สำหรับบริการ RouteGuide ใน src/client/client.rs

ก่อนอื่นให้นำโค้ดที่สร้างขึ้นมาใช้

mod grpc_pb {
    grpc::include_generated_proto!("generated", "routeguide");
}

use grpc_pb::route_guide_client::RouteGuideClient;
use grpc_pb::{Point, Rectangle, RouteNote};

เรียกเมธอดของบริการ

ตอนนี้มาดูวิธีเรียกเมธอดของบริการกัน ใน gRPC-Rust, RPC แบบสตรีมมิงเป็นแบบไม่พร้อมกันและไม่บล็อก โดยใช้ไวยากรณ์ async/await ของ Rust และสตรีม Tokio

RPC แบบสตรีมมิงฝั่งเซิร์ฟเวอร์: PrintFeatures

ใน RPC แบบสตรีมมิงฝั่งเซิร์ฟเวอร์ ไคลเอ็นต์จะส่งข้อความคำขอเดียวไปยังเซิร์ฟเวอร์ และได้รับข้อความตอบกลับเป็นสตรีม ต่อไปนี้คือตำแหน่งที่เราเรียกเมธอด list_features() แบบสตรีมมิงฝั่งเซิร์ฟเวอร์ (ซึ่งสอดคล้องกับการประกาศ RPC ListFeatures ที่พบใน proto ของเรา) ใน client.rs จากนั้นเซิร์ฟเวอร์จะส่งข้อความ Feature เป็นสตรีมกลับมา

async fn print_features(client: &RouteGuideClient<Channel>) -> Result<(), Box<dyn Error>> {
    let rectangle = proto!(Rectangle {
        lo: proto!(Point {
            latitude: 400_000_000,
            longitude: -750_000_000,
        }),
        hi: proto!(Point {
            latitude: 420_000_000,
            longitude: -730_000_000,
        }),
    });

    let mut stream = client.list_features(rectangle).await;

    while let Some(feature) = stream.recv().await {
        println!(
            "FEATURE: Name = \"{}\", Lat = {}, Lon = {}",
            feature.name(),
            feature.location().latitude(),
            feature.location().longitude()
        );
    }
    let status = stream.status().await;
    assert!(status.is_ok(), "{:?}", status);
    Ok(())
}

RPC แบบสตรีมมิงฝั่งไคลเอ็นต์: RecordRoute

เมื่อใช้การสตรีมมิงฝั่งไคลเอ็นต์ ไคลเอ็นต์จะเปิดสตรีมไปยังเซิร์ฟเวอร์และส่งลำดับข้อความ ไคลเอ็นต์จะได้รับข้อความตอบกลับเดียวเมื่อสตรีมเสร็จสิ้น

ในที่นี้ เราจะเริ่มการเรียกด้วย client.record_route().await ส่งพิกัด Point ที่สร้างขึ้นหลายรายการทีละรายการผ่านสตรีมโดยใช้ stream.send(point).await แล้วปิดสตรีมด้วย stream.close_and_recv().await เพื่อรับข้อความ RouteSummary เดียวจากเซิร์ฟเวอร์

async fn run_record_route(client: &RouteGuideClient<Channel>) -> Result<(), Box<dyn Error>> {
    let mut rng = rand::rng();
    let point_count: i32 = rng.random_range(2..100);

    let mut points = vec![];
    for _ in 0..=point_count {
        points.push(random_point(&mut rng));
    }

    println!("Traversing {} points", points.len());
    let mut stream = client.record_route().await;

    for point in &points {
        if stream.send(point).await.is_err() {
            break;
        }
    }

    match stream.close_and_recv().await {
        Ok(response) => {
            println!(
                "SUMMARY: Feature Count = {}, Distance = {}",
                response.feature_count(),
                response.distance()
            );
        }
        Err(e) => println!("something went wrong: {e:?}"),
    }
    Ok(())
}

RPC แบบสตรีมมิงแบบสองทิศทาง: RouteChat

สุดท้าย มาดู RPC แบบสตรีมมิงแบบสองทิศทาง RouteChat() กัน ในที่นี้ ทั้งไคลเอ็นต์และเซิร์ฟเวอร์จะส่งลำดับข้อความไปมา เราสร้างงาน tokio เพื่อส่งข้อความไปยังเซิร์ฟเวอร์อย่างต่อเนื่องด้วย tx.send(note).await.is_err() ขณะเดียวกัน rx.recv().await จะรอรับข้อความตอบกลับจากเซิร์ฟเวอร์และพิมพ์ข้อความเมื่อได้รับ

async fn run_route_chat(client: &RouteGuideClient<Channel>) -> Result<(), Box<dyn Error>> {
    let (mut tx, mut rx) = client.route_chat().await;

    let start = time::Instant::now();
    tokio::spawn(async move {
        let mut interval = time::interval(Duration::from_millis(50));
        for _ in 0..10 {
            let time = interval.tick().await;
            let elapsed = time.duration_since(start);
            let note = proto!(RouteNote {
                location: proto!(Point {
                    latitude: 409146138 + elapsed.as_millis() as i32,
                    longitude: -746188906,
                }),
                message: format!("at {elapsed:?}"),
            });
            if tx.send(note).await.is_err() {
                return;
            }
        }
        tx.close();
    });

    while let Some(note) = rx.recv().await {
        println!(
            "Note: Latitude = {}, Longitude = {}, Message = \"{}\"",
            note.location().latitude(),
            note.location().longitude(),
            note.message()
        );
    }
    let status = rx.status().await;
    assert!(status.is_ok(), "{:?}", status);
    Ok(())
}

แม้ว่าแต่ละฝ่ายจะได้รับข้อความของอีกฝ่ายตามลำดับที่เขียนเสมอ แต่ทั้งไคลเอ็นต์และเซิร์ฟเวอร์สามารถอ่านและเขียนได้ตามลำดับใดก็ได้ เนื่องจากสตรีมทำงานแยกกันอย่างสมบูรณ์

สร้างและส่งไคลเอ็นต์

หากต้องการเรียกเมธอดของบริการ เราต้องสร้าง แชแนล เพื่อสื่อสารกับเซิร์ฟเวอร์ก่อน เราสร้างแชแนลนี้โดยสร้างปลายทางก่อน จากนั้นเชื่อมต่อกับปลายทางดังกล่าว และส่งแชแนลที่สร้างขึ้นเมื่อเชื่อมต่อกับ RouteGuideClient::new() ดังนี้

// Create channel to connect to server
let channel = Channel::builder(
    "dns:///[::1]:10000",
    Arc::new(LocalChannelCredentials::new()),
)
.build();

// Create a new client
let client = RouteGuideClient::new(channel);

เมื่อสร้างไคลเอ็นต์นี้แล้ว เราจะเรียกเมธอดที่เราเขียนไว้ข้างต้นโดยส่งไคลเอ็นต์ไปยังเมธอดเหล่านั้น เราเพิ่มโค้ดทั้งหมดนี้ลงใน main() ซึ่งใช้รันไทม์แบบไม่พร้อมกันของ Tokio โค้ดทั้งหมดมีดังนี้

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Create channel to connect to server
    let channel = Channel::builder(
        "dns:///[::1]:10000",
        Arc::new(LocalChannelCredentials::new()),
    )
    .build();

    // Create a new client
    let client = RouteGuideClient::new(channel);

    println!("\n*** SERVER STREAMING ***");
    print_features(&client).await?;

    println!("\n*** CLIENT STREAMING ***");
    run_record_route(&client).await?;

    println!("\n*** BIDIRECTIONAL STREAMING ***");
    run_route_chat(&client).await?;

    Ok(())
}

7. ลองเลย

หากต้องการเรียกใช้ไคลเอ็นต์และเซิร์ฟเวอร์ ให้ตรวจสอบก่อนว่าได้กำหนดเป้าหมายไบนารีทั้ง 2 รายการไว้ใน Cargo.toml แล้ว

[[bin]]
name = "routeguide-server"
path = "src/server/server.rs"

[[bin]]
name = "routeguide-client"
path = "src/client/client.rs"

จากนั้นเรียกใช้คำสั่งต่อไปนี้จากไดเรกทอรีการทำงาน

  1. เรียกใช้เซิร์ฟเวอร์ในเทอร์มินัลหนึ่ง
cargo run --bin routeguide-server
  1. เรียกใช้ไคลเอ็นต์จากเทอร์มินัลอื่น
cargo run --bin routeguide-client

คุณจะเห็นเอาต์พุตลักษณะนี้

*** SERVER STREAMING ***
FEATURE: Name = "Patriots Path, Mendham, NJ 07945, USA", Lat = 407838351, Lon = -746143763
FEATURE: Name = "101 New Jersey 10, Whippany, NJ 07981, USA", Lat = 408122808, Lon = -743999179
FEATURE: Name = "U.S. 6, Shohola, PA 18458, USA", Lat = 413628156, Lon = -749015468
...
*** CLIENT STREAMING ***
Traversing 86 points
SUMMARY: Feature Count = 0, Distance = 803709356

*** BIDIRECTIONAL STREAMING ***
Note: Latitude = 409146138, Longitude = -746188906, Message = "at 112.45µs"
Note: Latitude = 409146139, Longitude = -746188906, Message = "at 1.00011245s"
Note: Latitude = 409146140, Longitude = -746188906, Message = "at 2.00011245s"

8. ขั้นตอนถัดไป

9. ผู้ร่วมให้ข้อมูลใน Codelab นี้

  • Cathy Zhao
  • Arvind Bright
  • Nathaniel Ford