gRPC-Rust का इस्तेमाल शुरू करना - स्ट्रीमिंग

1. परिचय

इस कोडलैब में, gRPC-Rust का इस्तेमाल करके एक क्लाइंट और सर्वर बनाया जाएगा. ये दोनों, Rust में लिखे गए रूट-मैपिंग ऐप्लिकेशन की बुनियादी ज़रूरतें पूरी करते हैं.

ट्यूटोरियल के आखिर तक, आपके पास एक ऐसा क्लाइंट होगा जो gRPC का इस्तेमाल करके रिमोट सर्वर से कनेक्ट होता है. इससे क्लाइंट के रूट पर मौजूद सुविधाओं के बारे में जानकारी मिलती है, क्लाइंट के रूट की खास जानकारी बनाई जाती है, और सर्वर और अन्य क्लाइंट के साथ रूट की जानकारी शेयर की जाती है. जैसे, ट्रैफ़िक अपडेट.

सेवा को प्रोटोकॉल बफ़र फ़ाइल में तय किया जाता है. इसका इस्तेमाल क्लाइंट और सर्वर के लिए बॉयलरप्लेट कोड जनरेट करने के लिए किया जाएगा, ताकि वे एक-दूसरे से कम्यूनिकेट कर सकें. इससे आपको इस सुविधा को लागू करने में समय और मेहनत नहीं करनी पड़ेगी.

जनरेट किया गया यह कोड, सर्वर और क्लाइंट के बीच कम्यूनिकेशन की जटिलताओं के साथ-साथ डेटा के क्रमबद्ध और क्रम से हटाने की प्रोसेस को भी मैनेज करता है.

आपको क्या सीखने को मिलेगा

• किसी सेवा के एपीआई को तय करने के लिए, प्रोटोकॉल बफ़र का इस्तेमाल कैसे करें.
• ऑटोमेटेड कोड जनरेशन का इस्तेमाल करके, Protocol Buffers की परिभाषा से gRPC पर आधारित क्लाइंट और सर्वर बनाने का तरीका.
• gRPC के साथ क्लाइंट-सर्वर स्ट्रीमिंग कम्यूनिकेशन के बारे में जानकारी.

यह कोडलैब, उन Rust डेवलपर के लिए है जो gRPC का इस्तेमाल पहली बार कर रहे हैं या gRPC के बारे में फिर से जानकारी पाना चाहते हैं. इसके अलावा, यह उन लोगों के लिए भी है जो डिस्ट्रिब्यूटेड सिस्टम बनाना चाहते हैं. इसके लिए, gRPC का अनुभव होना ज़रूरी नहीं है.

2. शुरू करने से पहले

ज़रूरी शर्तें

पक्का करें कि आपने ये इंस्टॉल किए हों:

• GCC. यहां दिए गए निर्देशों का पालन करें
• Rust का नया वर्शन. इंस्टॉल करने के निर्देशों के लिए, यहां जाएं.

कोड प्राप्त करें

इसलिए, आपको शुरू से काम न करना पड़े, इस कोडलैब में ऐप्लिकेशन के सोर्स कोड का एक स्केफ़ोल्ड दिया गया है. आपको इसे पूरा करना होगा. यहां दिए गए तरीके से, ऐप्लिकेशन को पूरा करने का तरीका जानें. इसमें, बॉयलरप्लेट gRPC कोड जनरेट करने के लिए, प्रोटोकॉल बफ़र कंपाइलर प्लगिन का इस्तेमाल करने का तरीका भी शामिल है.

सबसे पहले, कोडलैब की वर्किंग डायरेक्ट्री बनाएं और उसमें cd करें:

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

कोडलैब को डाउनलोड और एक्सट्रैक्ट करें:

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

इसके अलावा, सिर्फ़ कोडलैब डायरेक्ट्री वाली .zip फ़ाइल डाउनलोड करके, उसे मैन्युअल तरीके से अनज़िप किया जा सकता है.

ज़िप फ़ाइल डाउनलोड करें

अगर आपको लागू करने के लिए टाइप नहीं करना है, तो पूरा सोर्स कोड GitHub पर उपलब्ध है.

3. संदेशों और सेवाओं के बारे में जानकारी

सबसे पहले, प्रोटोकॉल बफ़र का इस्तेमाल करके, ऐप्लिकेशन की gRPC सेवा, उसके आरपीसी तरीके, और उसके अनुरोध और जवाब के मैसेज टाइप तय करें. आपकी सेवा से ये सुविधाएं मिलेंगी:

• सर्वर लागू करता है और क्लाइंट कॉल करता है, जिन्हें RPC तरीके ListFeatures, RecordRoute, और RouteChat कहा जाता है.
• मैसेज टाइप Point, Feature, Rectangle, RouteNote, और RouteSummary, जो ऊपर दिए गए तरीकों को कॉल करते समय क्लाइंट और सर्वर के बीच डेटा स्ट्रक्चर के तौर पर शेयर किए जाते हैं.

ये आरपीसी तरीके और इनके मैसेज टाइप, दिए गए सोर्स कोड की proto/routeguide.proto फ़ाइल में तय किए जाएंगे.

प्रोटोकॉल बफ़र को आम तौर पर, protobufs के नाम से जाना जाता है. gRPC की शब्दावली के बारे में ज़्यादा जानने के लिए, gRPC के मुख्य कॉन्सेप्ट, आर्किटेक्चर, और लाइफ़साइकल देखें.

मैसेज के टाइप तय करना

सबसे पहले, हम उन मैसेज को तय करते हैं जिनका इस्तेमाल हमारे आरपीसी करेंगे. सोर्स कोड की routeguide/route_guide.proto फ़ाइल में, सबसे पहले Point मैसेज टाइप तय करें. Point, मैप पर अक्षांश-देशांतर के निर्देशांकों के जोड़े को दिखाता है. इस कोडलैब के लिए, पूर्णांकों का इस्तेमाल करें:

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

1 और 2 नंबर, message स्ट्रक्चर में मौजूद हर फ़ील्ड के लिए यूनीक आईडी नंबर होते हैं.

इसके बाद, Feature मैसेज टाइप तय करें. Feature, Point में बताई गई जगह पर मौजूद किसी चीज़ के नाम या डाक पते के लिए string फ़ील्ड का इस्तेमाल करता है:

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

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

इसके बाद, Rectangle मैसेज है. यह अक्षांश-देशांतर आयत को दिखाता है. इसे दो विकर्ण विपरीत बिंदुओं "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 पर भेजा गया मैसेज भी चाहिए होगा. यह मैसेज, 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 होता है. यह ऐप्लिकेशन की सेवा के ज़रिए उपलब्ध कराए गए एक या उससे ज़्यादा तरीकों के बारे में बताता है.

अपनी सेवा की परिभाषा में आरपीसी के तरीके तय करें. साथ ही, उनके अनुरोध और जवाब के टाइप तय करें. इस कोडलैब के इस सेक्शन में, हम इन चीज़ों के बारे में जानेंगे:

ListFeatures

यह कुकी, दिए गए Rectangle में उपलब्ध Features को इकट्ठा करती है. नतीजे एक साथ नहीं दिखाए जाते, बल्कि स्ट्रीम किए जाते हैं.उदाहरण के लिए, बार-बार दिखने वाले फ़ील्ड के साथ जवाब के मैसेज में ऐसा होता है. ऐसा इसलिए, क्योंकि रेक्टैंगल में बहुत बड़ा एरिया शामिल हो सकता है और इसमें कई सुविधाएं हो सकती हैं.

इस RPC के लिए, सर्वर-साइड स्ट्रीमिंग RPC सबसे सही है: क्लाइंट, सर्वर को अनुरोध भेजता है और उसे वापस मैसेज की एक स्ट्रीम मिलती है. क्लाइंट, जवाब के तौर पर मिली स्ट्रीम से तब तक डेटा पढ़ता है, जब तक कोई और मैसेज नहीं मिलता. हमारे उदाहरण में दिखाया गया है कि रिस्पॉन्स टाइप से पहले stream कीवर्ड रखकर, सर्वर-साइड स्ट्रीमिंग के तरीके के बारे में बताया जाता है.

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

RecordRoute

यह फ़ंक्शन, तय किए गए रास्ते पर मौजूद Points की स्ट्रीम को स्वीकार करता है. साथ ही, रास्ता तय हो जाने पर RouteSummary दिखाता है.

इस मामले में, क्लाइंट-साइड स्ट्रीमिंग आरपीसी सही लगती है: क्लाइंट, मैसेज का क्रम लिखता है और उन्हें सर्वर को भेजता है. इसके लिए, वह उपलब्ध स्ट्रीम का फिर से इस्तेमाल करता है. क्लाइंट के मैसेज लिखने के बाद, वह सर्वर के उन सभी मैसेज को पढ़ने और जवाब देने का इंतज़ार करता है. अनुरोध के टाइप से पहले stream कीवर्ड रखकर, क्लाइंट-साइड स्ट्रीमिंग का तरीका तय किया जाता है.

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

RouteChat

यह कुकी, रास्ते पर चलते समय भेजे गए RouteNotes की स्ट्रीम को स्वीकार करती है. साथ ही, अन्य RouteNotes (जैसे, अन्य उपयोगकर्ताओं से) को स्वीकार करती है.

दोनों दिशाओं में स्ट्रीमिंग की सुविधा का इस्तेमाल, इस तरह के मामलों में किया जा सकता है. द्विदिश स्ट्रीमिंग आरपीसी में, दोनों पक्ष पढ़ने-लिखने वाली स्ट्रीम का इस्तेमाल करके मैसेज का क्रम भेजते हैं. ये दोनों स्ट्रीम अलग-अलग काम करती हैं. इसलिए, क्लाइंट और सर्वर अपनी पसंद के हिसाब से किसी भी क्रम में डेटा पढ़ और लिख सकते हैं.

उदाहरण के लिए, सर्वर अपने जवाब लिखने से पहले, सभी क्लाइंट मैसेज पाने का इंतज़ार कर सकता है. इसके अलावा, वह एक मैसेज पढ़ सकता है और फिर एक मैसेज लिख सकता है. इसके अलावा, वह पढ़ने और लिखने की किसी अन्य प्रोसेस का इस्तेमाल कर सकता है.

हर स्ट्रीम में मैसेज का क्रम बना रहता है. इस तरह के तरीके को तय करने के लिए, अनुरोध और जवाब, दोनों से पहले stream कीवर्ड का इस्तेमाल किया जाता है.

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

4. क्लाइंट और सर्वर कोड जनरेट करना

हमने जनरेट की गई डायरेक्ट्री में मौजूद .proto फ़ाइल से जनरेट किया गया कोड आपको पहले ही दे दिया है.

अगर आपको .proto फ़ाइल से कोड जनरेट करने का तरीका जानना है या .proto फ़ाइल में कोई बदलाव करके उसे आज़माना है, तो ये निर्देश पढ़ें.

जनरेट किए गए कोड में ये शामिल हैं:

• मैसेज टाइप Point, Feature, Rectangle, RouteNote, और RouteSummary के लिए स्ट्रक्चर की परिभाषाएं.
• हमें सेवा की इस विशेषता को लागू करना होगा: route_guide_server::RouteGuide.
• सर्वर को कॉल करने के लिए, हम इस क्लाइंट टाइप का इस्तेमाल करेंगे: route_guide_client::RouteGuideClient<T>.

इसके बाद, हम सर्वर-साइड पर तरीकों को लागू करेंगे, ताकि जब क्लाइंट कोई अनुरोध भेजे, तो सर्वर जवाब दे सके.

5. सेवा लागू करना

सबसे पहले, देखते हैं कि RouteGuide सर्वर कैसे बनाया जाता है. RouteGuide सेवा को काम करने के लिए, दो चीज़ें ज़रूरी होती हैं:

• हमारी सेवा की परिभाषा से जनरेट किए गए सेवा इंटरफ़ेस को लागू करना: हमारी सेवा का असली "काम" करना.
• क्लाइंट से मिले अनुरोधों को सुनने और उन्हें सही तरीके से लागू करने के लिए, gRPC सर्वर को चालू करना.

src/server/server.rs में, जनरेट किए गए कोड को gRPC के include_generated_proto! मैक्रो की मदद से स्कोप में लाया जा सकता है. साथ ही, 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
};

ध्यान दें: अगर आपने कोड खुद जनरेट किया है और out_dir को मैन्युअल तरीके से सेट करने के बजाय, Cargo के डिफ़ॉल्ट OUT_DIR एनवायरमेंट वैरिएबल का इस्तेमाल किया है, तो grpc::include_proto! मैक्रो का इस्तेमाल किया जा सकता है. कोड जनरेट करने से जुड़ा पूरा ट्यूटोरियल देखने के लिए, यहाँ क्लिक करें.

हम अपनी सेवा को दिखाने के लिए, एक स्ट्रक्चर तय करके शुरुआत कर सकते हैं. फ़िलहाल, हम 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> {
        ...
    }
}

आइए, हर आरपीसी को लागू करने के तरीके के बारे में विस्तार से जानते हैं.

सर्वर-साइड स्ट्रीमिंग आरपीसी

चलिए, ListFeatures से शुरू करते हैं. यह सर्वर-साइड स्ट्रीमिंग आरपीसी है. इसलिए, हमें अपने क्लाइंट को कई 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 में रैप करके वापस भेज दिया जाता है.

क्लाइंट-साइड स्ट्रीमिंग आरपीसी

अब हम एक ऐसे तरीके के बारे में बात करते हैं जो थोड़ा मुश्किल है: क्लाइंट-साइड स्ट्रीमिंग का तरीका 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 है, तो स्ट्रीम अब भी अच्छी है और इसे पढ़ना जारी रखा जा सकता है.

दोनों दिशाओं में डेटा ट्रांसफ़र करने वाली स्ट्रीमिंग आरपीसी

आखिर में, आइए हम दोनों दिशाओं में स्ट्रीम करने वाले आरपीसी 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! का इस्तेमाल करके output स्ट्रीम बनाते हैं. इससे पता चलता है कि स्ट्रीम में गड़बड़ियां हो सकती हैं.

सर्वर शुरू करना

इस तरीके को लागू करने के बाद, हमें एक 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. हमने जो सेवा बनाई है उसका इस्तेमाल करके, RouteGuideServer::new() की मदद से gRPC सर्वर का इंस्टेंस बनाएं.
4. gRPC सर्वर के साथ, हमारी सेवा को लागू करने की सुविधा रजिस्टर करें.
5. पोर्ट करने की जानकारी के साथ सर्वर पर serve() को कॉल करें, ताकि प्रोसेस बंद होने तक इंतज़ार किया जा सके.

6. क्लाइंट बनाना

इस सेक्शन में, हम src/client/client.rs में RouteGuide सेवा के लिए Rust क्लाइंट बनाने का तरीका जानेंगे.

सबसे पहले, जनरेट किए गए कोड को स्कोप में लाएं.

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

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

कॉल करने की सेवा के तरीके

अब देखते हैं कि हम अपनी सेवा के तरीकों को कैसे कॉल करते हैं. gRPC-Rust में, आरपीसी ब्लॉक करने/सिंक्रोनस मोड में काम करते हैं. इसका मतलब है कि आरपीसी कॉल, सर्वर के जवाब देने का इंतज़ार करता है. इसके बाद, यह या तो जवाब देगा या गड़बड़ी दिखाएगा.

सर्वर-साइड स्ट्रीमिंग आरपीसी

यहां हम सर्वर साइड स्ट्रीमिंग वाले ListFeatures तरीके को कॉल करते हैं. यह भौगोलिक Feature ऑब्जेक्ट की स्ट्रीम दिखाता है.

async fn print_features(client: &mut 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(Request::new(rectangle))
        .await?
        .into_inner();

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

हम इस तरीके को एक अनुरोध भेजते हैं और हमें ListFeaturesStream का एक इंस्टेंस वापस मिलता है. क्लाइंट, सर्वर के जवाबों को पढ़ने के लिए ListFeaturesStream स्ट्रीम का इस्तेमाल कर सकता है. हम ListFeaturesStream के message() तरीके का इस्तेमाल करते हैं, ताकि सर्वर के जवाबों को बार-बार पढ़ा जा सके. ऐसा तब तक किया जाता है, जब तक कोई और मैसेज नहीं मिलता. इस मामले में, जवाब प्रोटोकॉल बफ़र ऑब्जेक्ट (इस मामले में Feature) का इस्तेमाल किया जाता है.

क्लाइंट-साइड स्ट्रीमिंग आरपीसी

यहां record_route के लिए, हम पॉइंट के वेक्टर को स्ट्रीम में बदलते हैं. इसके बाद, हम इस स्ट्रीम को अनुरोध के तौर पर record_route() में पास करते हैं. सर्वर से स्ट्रीम पूरी तरह प्रोसेस होने के बाद, हमें एक RouteSummary जवाब मिलता है.

async fn run_record_route(client: &mut 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 request = Request::new(tokio_stream::iter(points));

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

    Ok(())
}

दोनों दिशाओं में डेटा ट्रांसफ़र करने वाली स्ट्रीमिंग आरपीसी

आखिर में, आइए हम दोनों दिशाओं में स्ट्रीम करने वाले आरपीसी RouteChat() पर नज़र डालें. हम इस तरीके को एक स्ट्रीम अनुरोध भेजते हैं. हम इस अनुरोध को लिखते हैं और हमें एक स्ट्रीम मिलती है, जिससे हम मैसेज पढ़ सकते हैं. इस बार, हम अपने तरीके की स्ट्रीम के ज़रिए वैल्यू वापस भेजते हैं. इस दौरान, सर्वर अपनी मैसेज स्ट्रीम में मैसेज लिखता रहता है.

async fn run_route_chat(client: &mut RouteGuideClient<Channel>) -> Result<(), Box<dyn Error>> {
    let start = time::Instant::now();
    let outbound = async_stream::stream! {
        let mut interval = time::interval(Duration::from_secs(1));
        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_secs() as i32,
                    longitude: -746188906,
                }),
                message: format!("at {elapsed:?}"),
            });
            yield note;
        }
    };
    let response = client.route_chat(Request::new(outbound)).await?;
    let mut inbound = response.into_inner();
    while let Some(note) = inbound.message().await? {
        println!("Note: Latitude = {}, Longitude = {}, Message = \"{}\"",
            note.location().latitude(),
            note.location().longitude(),
            note.message());
        }
    Ok(())
}

दोनों पक्षों को एक-दूसरे के मैसेज, उसी क्रम में मिलते हैं जिस क्रम में उन्हें लिखा गया था. हालांकि, क्लाइंट और सर्वर, मैसेज को किसी भी क्रम में पढ़ और लिख सकते हैं. स्ट्रीम पूरी तरह से स्वतंत्र रूप से काम करती हैं.

कॉल हेल्पर के तरीके

सेवा के तरीकों को कॉल करने के लिए, हमें सर्वर से कम्यूनिकेट करने के लिए सबसे पहले एक चैनल बनाना होगा. हम इसे बनाने के लिए, सबसे पहले एक एंडपॉइंट बनाते हैं. इसके बाद, उस एंडपॉइंट से कनेक्ट करते हैं. फिर, RouteGuideClient::new() से कनेक्ट होने पर बनाए गए चैनल को इस तरह पास करते हैं:

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Create endpoint to connect to
    let endpoint = Endpoint::new("http://[::1]:10000")?; 
    let channel = endpoint.connect().await?;             

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

main() में, अभी-अभी बनाए गए तरीकों को लागू करें.

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Create endpoint to connect to
    let endpoint = Endpoint::new("http://[::1]:10000")?; 
    let channel = endpoint.connect().await?;             

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

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

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

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

    Ok(())
}

7. इसे आज़माएं

सबसे पहले, क्लाइंट और सर्वर को चलाने के लिए, उन्हें अपने क्रेट में बाइनरी टारगेट के तौर पर जोड़ें. हमें Cargo.toml फ़ाइल में बदलाव करना होगा:

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

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

किसी भी प्रोजेक्ट की तरह, हमें उन डिपेंडेंसी के बारे में भी सोचना होगा जो हमारे कोड के काम करने के लिए ज़रूरी हैं. Rust प्रोजेक्ट के लिए, डिपेंडेंसी Cargo.toml में होंगी. हमने Cargo.toml फ़ाइल में ज़रूरी डिपेंडेंसी पहले ही लिस्ट कर दी हैं.

इसके बाद, हमारी वर्किंग डायरेक्ट्री से ये कमांड चलाएं:

1. सर्वर को एक टर्मिनल में चलाएं:

RUSTFLAGS="-Awarnings" cargo run --bin routeguide-server 

2. क्लाइंट को किसी दूसरे टर्मिनल से चलाएं:

RUSTFLAGS="-Awarnings" cargo run --bin routeguide-client

ध्यान दें: "RUSTFLAGS="-Awarnings" flag hides warnings from the the Rust 1.89.0 update.

आपको इस तरह का आउटपुट दिखेगा:

*** 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. आगे क्या करना है

• gRPC के बारे में जानकारी और मुख्य सिद्धांत में, gRPC के काम करने का तरीका जानें.
• बुनियादी बातों से जुड़ा ट्यूटोरियल देखें.