grpc_gateway 0.1.3

A GRPC Gateway which handle the http request and convert into GRPC
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114

use std::{
    sync::{Arc, RwLock},
    time::Duration,
};

use anyhow::Result;
use prost_reflect::{MethodDescriptor, ServiceDescriptor};
use tokio::time::Instant;
use tonic::transport::Channel;

use crate::discriptor;

use discriptor::discriptor::CachedDescriptors;

#[derive(Debug)]
pub struct ReflectionDiscriptorManager {
    pub cache: Arc<RwLock<CachedDescriptors>>,
    pub channel: Channel,
    pub refresh_interval: Duration,
    pub last_refresh: Arc<RwLock<Instant>>,
}

impl ReflectionDiscriptorManager {
    pub async fn new(endpoint: &str) -> Result<Self> {
        let channel = tonic::transport::Channel::from_shared(endpoint.to_string())?
            .connect()
            .await?;

        let manager = Self {
            cache: Arc::new(RwLock::new(CachedDescriptors::new())),
            channel: channel,
            refresh_interval: Duration::from_secs(300), // 5 -min
            last_refresh: Arc::new(RwLock::new(Instant::now())),
        };

        // Ensure descriptors are loaded; propagate errors so callers see real cause
        manager.refresh_discriptors().await?;

        Ok(manager)
    }

    pub async fn refresh_discriptors(&self) -> Result<()> {
        // Build a fresh descriptor cache without holding locks across await
        let mut new_cache = CachedDescriptors::new();
        new_cache.load_discriptor(self.channel.clone()).await?;

        {
            let mut cache_guard = self.cache.write().unwrap();
            *cache_guard = new_cache;
        }

        // Debug: print discovered services to help diagnose mismatches
        {
            let cache = self.cache.read().unwrap();
            let services = cache.get_all_service();
            println!("[descriptor] loaded services: {:?}", services);
        }

        let mut refresh = self.last_refresh.write().unwrap();
        *refresh = Instant::now();
        Ok(())
    }

    // Get method descriptor from refresh discriptor
    pub async fn get_method(
        &self,
        service: &str,
        method: &str,
    ) -> Result<Option<MethodDescriptor>> {
        //check cache needs refresh
        // {
        //     let last_refresh = self.last_refresh.read().unwrap();
        //     if last_refresh.elapsed() > self.refresh_interval {
        //         drop(last_refresh);

        //         // refresh auto, didn't block the request
        //         let manager = self.clone();
        //         tokio::spawn(async move {
        //             match manager.refresh_discriptors().await {
        //                 Err(e) => {}
        //                 _ => (),
        //             }
        //         });
        //     }
        // }

        let cache = self.cache.read().unwrap();
        Ok(cache.get_method(service, method).cloned())
    }

    pub async fn get_service(&self, service: &str) -> Option<ServiceDescriptor> {
        let cache = self.cache.read().unwrap();
        cache.get_service(service).cloned()
    }

    pub async fn list_services(&self) -> Vec<String> {
        let cache = self.cache.read().unwrap();
        cache.get_all_service()
    }

    pub async fn force_refreshd(&self) -> Result<()> {
        self.refresh_discriptors().await
    }
}

pub async fn get_discriptor_manager(endpoint: &str) -> Arc<ReflectionDiscriptorManager> {
    let manager = Arc::new(
        ReflectionDiscriptorManager::new(endpoint)
            .await
            .expect("msg"),
    );

    manager
}