kaspa-grpc-server 0.15.0

Kaspa gRPC server
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
115
116
117
118
119
120
121
122
123
124
125
126
127
128

use crate::error::GrpcServerResult;
use async_trait::async_trait;
use futures::Future;
use std::{pin::Pin, sync::Arc};

pub enum RoutingPolicy<Request, Response> {
    Enqueue,
    DropIfFull(DropFn<Request, Response>),
}

impl<Request, Response> Clone for RoutingPolicy<Request, Response> {
    fn clone(&self) -> Self {
        match self {
            Self::Enqueue => Self::Enqueue,
            Self::DropIfFull(drop_fn) => Self::DropIfFull(drop_fn.clone()),
        }
    }
}

#[async_trait]
pub trait MethodTrait<ServerContext, ConnectionContext, Request, Response>: Send + Sync + 'static {
    async fn call(&self, server_ctx: ServerContext, connection_ctx: ConnectionContext, request: Request)
        -> GrpcServerResult<Response>;

    fn method_fn(&self) -> MethodFn<ServerContext, ConnectionContext, Request, Response>;
    fn tasks(&self) -> usize;
    fn queue_size(&self) -> usize;
    fn routing_policy(&self) -> RoutingPolicy<Request, Response>;
}

/// RPC method function type
pub type MethodFn<ServerContext, ConnectionContext, Request, Response> =
    Arc<Box<dyn Send + Sync + Fn(ServerContext, ConnectionContext, Request) -> MethodFnReturn<Response> + 'static>>;

/// RPC method function return type
pub type MethodFnReturn<T> = Pin<Box<(dyn Send + 'static + Future<Output = GrpcServerResult<T>>)>>;

/// RPC drop function type
pub type DropFn<Request, Response> = Arc<Box<dyn Send + Sync + Fn(&Request) -> GrpcServerResult<Response>>>;

/// RPC method wrapper. Contains the method closure function.
pub struct Method<ServerContext, ConnectionContext, Request, Response>
where
    ServerContext: Send + Sync + 'static,
    ConnectionContext: Send + Sync + 'static,
    Request: Send + Sync + 'static,
    Response: Send + Sync + 'static,
{
    /// Function called when executing the method
    method_fn: MethodFn<ServerContext, ConnectionContext, Request, Response>,

    /// Number of connection concurrent request handlers
    tasks: usize,

    /// Size of the request queue
    queue_size: usize,

    /// Policy applied when the routing channel is full
    routing_policy: RoutingPolicy<Request, Response>,
}

impl<ServerContext, ConnectionContext, Request, Response> Method<ServerContext, ConnectionContext, Request, Response>
where
    ServerContext: Send + Sync + 'static,
    ConnectionContext: Send + Sync + 'static,
    Request: Send + Sync + 'static,
    Response: Send + Sync + 'static,
{
    pub fn new<FN>(method_fn: FN) -> Method<ServerContext, ConnectionContext, Request, Response>
    where
        FN: Send + Sync + Fn(ServerContext, ConnectionContext, Request) -> MethodFnReturn<Response> + 'static,
    {
        Method {
            method_fn: Arc::new(Box::new(method_fn)),
            tasks: 1,
            queue_size: Self::default_queue_size(),
            routing_policy: RoutingPolicy::Enqueue,
        }
    }

    pub fn with_properties(
        method_fn: MethodFn<ServerContext, ConnectionContext, Request, Response>,
        tasks: usize,
        queue_size: usize,
        routing_policy: RoutingPolicy<Request, Response>,
    ) -> Method<ServerContext, ConnectionContext, Request, Response> {
        Method { method_fn, tasks, queue_size, routing_policy }
    }

    pub fn default_queue_size() -> usize {
        256
    }
}

#[async_trait]
impl<ServerContext, ConnectionContext, Request, Response> MethodTrait<ServerContext, ConnectionContext, Request, Response>
    for Method<ServerContext, ConnectionContext, Request, Response>
where
    ServerContext: Clone + Send + Sync + 'static,
    ConnectionContext: Send + Sync + 'static,
    Request: Send + Sync + 'static,
    Response: Send + Sync + 'static,
{
    async fn call(
        &self,
        server_ctx: ServerContext,
        connection_ctx: ConnectionContext,
        request: Request,
    ) -> GrpcServerResult<Response> {
        (self.method_fn)(server_ctx, connection_ctx, request).await
    }

    fn method_fn(&self) -> MethodFn<ServerContext, ConnectionContext, Request, Response> {
        self.method_fn.clone()
    }

    fn tasks(&self) -> usize {
        self.tasks
    }

    fn queue_size(&self) -> usize {
        self.queue_size
    }

    fn routing_policy(&self) -> RoutingPolicy<Request, Response> {
        self.routing_policy.clone()
    }
}