massa_api/
api.rs

1//! Copyright (c) 2022 MASSA LABS <info@massa.net>
2//! Json RPC API for a massa-node
3use std::net::SocketAddr;
4
5use crate::api_trait::MassaApiServer;
6use crate::{ApiServer, ApiV2, StopHandle, API};
7use async_trait::async_trait;
8use futures::future::{self, Either};
9use futures::StreamExt;
10use jsonrpsee::core::{client::Error as JsonRpseeError, RpcResult, SubscriptionResult};
11use jsonrpsee::{PendingSubscriptionSink, SubscriptionMessage};
12use massa_api_exports::config::APIConfig;
13use massa_api_exports::error::ApiError;
14use massa_api_exports::page::{PageRequest, PagedVec, PagedVecV2};
15use massa_api_exports::ApiRequest;
16use massa_consensus_exports::{ConsensusBroadcasts, ConsensusController};
17use massa_execution_exports::ExecutionController;
18use massa_models::address::Address;
19use massa_models::block_id::BlockId;
20use massa_models::slot::Slot;
21use massa_models::timeslots::get_latest_block_slot_at_timestamp;
22use massa_models::version::Version;
23use massa_pool_exports::PoolBroadcasts;
24use massa_time::MassaTime;
25use serde::Serialize;
26use tokio_stream::wrappers::BroadcastStream;
27
28impl API<ApiV2> {
29    /// generate a new massa API
30    pub fn new(
31        consensus_controller: Box<dyn ConsensusController>,
32        consensus_broadcasts: ConsensusBroadcasts,
33        execution_controller: Box<dyn ExecutionController>,
34        pool_broadcasts: PoolBroadcasts,
35        api_settings: APIConfig,
36        version: Version,
37    ) -> Self {
38        API(ApiV2 {
39            consensus_controller,
40            consensus_broadcasts,
41            execution_controller,
42            pool_broadcasts,
43            api_settings,
44            version,
45        })
46    }
47}
48
49#[async_trait]
50impl ApiServer for API<ApiV2> {
51    async fn serve(
52        self,
53        url: &SocketAddr,
54        api_config: &APIConfig,
55    ) -> Result<StopHandle, JsonRpseeError> {
56        crate::serve(self.into_rpc(), url, api_config).await
57    }
58}
59
60#[doc(hidden)]
61#[async_trait]
62impl MassaApiServer for API<ApiV2> {
63    async fn get_largest_stakers(
64        &self,
65        api_request: Option<ApiRequest>,
66    ) -> RpcResult<PagedVecV2<(Address, u64)>> {
67        let cfg = self.0.api_settings.clone();
68
69        let now = MassaTime::now();
70
71        let latest_block_slot_at_timestamp_result = get_latest_block_slot_at_timestamp(
72            cfg.thread_count,
73            cfg.t0,
74            cfg.genesis_timestamp,
75            now,
76        );
77
78        let curr_cycle = match latest_block_slot_at_timestamp_result {
79            Ok(Some(cur_slot)) if cur_slot.period <= self.0.api_settings.last_start_period => {
80                Slot::new(self.0.api_settings.last_start_period, 0).get_cycle(cfg.periods_per_cycle)
81            }
82            Ok(Some(cur_slot)) => cur_slot.get_cycle(cfg.periods_per_cycle),
83            Ok(None) => 0,
84            Err(e) => return Err(ApiError::ModelsError(e).into()),
85        };
86
87        let mut staker_vec = self
88            .0
89            .execution_controller
90            .get_cycle_active_rolls(curr_cycle)
91            .into_iter()
92            .collect::<Vec<(Address, u64)>>();
93
94        staker_vec
95            .sort_by(|&(_, roll_counts_a), &(_, roll_counts_b)| roll_counts_b.cmp(&roll_counts_a));
96
97        // Omitted api_request and a present request with no page_request must
98        // behave the same: default to the first 50 largest stakers.
99        const DEFAULT_PAGE_LIMIT: usize = 50;
100        let page_request = api_request
101            .and_then(|request| request.page_request)
102            .or(Some(PageRequest {
103                offset: 0,
104                limit: DEFAULT_PAGE_LIMIT,
105            }));
106        let paged_vec = PagedVec::new(staker_vec, page_request);
107
108        Ok(paged_vec.into())
109    }
110
111    async fn get_next_block_best_parents(&self) -> RpcResult<Vec<(BlockId, u64)>> {
112        Ok(self.0.consensus_controller.get_best_parents())
113    }
114
115    async fn get_version(&self) -> RpcResult<Version> {
116        Ok(self.0.version)
117    }
118
119    async fn subscribe_new_blocks(&self, pending: PendingSubscriptionSink) -> SubscriptionResult {
120        broadcast_via_ws(self.0.consensus_broadcasts.block_sender.clone(), pending).await
121    }
122
123    async fn subscribe_new_blocks_headers(
124        &self,
125        pending: PendingSubscriptionSink,
126    ) -> SubscriptionResult {
127        broadcast_via_ws(
128            self.0.consensus_broadcasts.block_header_sender.clone(),
129            pending,
130        )
131        .await
132    }
133
134    async fn subscribe_new_filled_blocks(
135        &self,
136        pending: PendingSubscriptionSink,
137    ) -> SubscriptionResult {
138        broadcast_via_ws(
139            self.0.consensus_broadcasts.filled_block_sender.clone(),
140            pending,
141        )
142        .await
143    }
144
145    async fn subscribe_new_operations(
146        &self,
147        pending: PendingSubscriptionSink,
148    ) -> SubscriptionResult {
149        broadcast_via_ws(self.0.pool_broadcasts.operation_sender.clone(), pending).await
150    }
151}
152
153// Broadcast the stream(sender) content via a WebSocket
154async fn broadcast_via_ws<T: Serialize + Send + Clone + 'static>(
155    sender: tokio::sync::broadcast::Sender<T>,
156    pending: PendingSubscriptionSink,
157) -> SubscriptionResult {
158    let sink = pending.accept().await?;
159    let closed = sink.closed();
160    let stream = BroadcastStream::new(sender.subscribe());
161    futures::pin_mut!(closed, stream);
162
163    loop {
164        match future::select(closed, stream.next()).await {
165            // subscription closed.
166            Either::Left((_, _)) => break Ok(()),
167
168            // received new item from the stream.
169            Either::Right((Some(Ok(item)), c)) => {
170                let notif = SubscriptionMessage::from_json(&item)?;
171
172                if sink.send(notif).await.is_err() {
173                    break Ok(());
174                }
175
176                closed = c;
177            }
178
179            // Send back back the error.
180            Either::Right((Some(Err(e)), _)) => break Err(e.into()),
181
182            // Stream is closed.
183            Either::Right((None, _)) => break Ok(()),
184        }
185    }
186}
massa_api/api.rs

massa_api/
api.rs
