mssf_core/client/
query_client.rs

1// ------------------------------------------------------------
2// Copyright (c) Microsoft Corporation.  All rights reserved.
3// Licensed under the MIT License (MIT). See License.txt in the repo root for license information.
4// ------------------------------------------------------------
5use std::{ffi::c_void, time::Duration};
6
7use mssf_com::{
8    FabricClient::{
9        IFabricGetApplicationListResult2, IFabricGetDeployedServiceReplicaDetailResult,
10        IFabricGetNodeListResult2, IFabricGetPartitionListResult2,
11        IFabricGetPartitionLoadInformationResult, IFabricGetReplicaListResult2,
12        IFabricQueryClient10,
13    },
14    FabricTypes::{
15        FABRIC_APPLICATION_QUERY_DESCRIPTION,
16        FABRIC_DEPLOYED_SERVICE_REPLICA_DETAIL_QUERY_DESCRIPTION, FABRIC_NODE_QUERY_DESCRIPTION,
17        FABRIC_NODE_QUERY_DESCRIPTION_EX1, FABRIC_NODE_QUERY_DESCRIPTION_EX2,
18        FABRIC_NODE_QUERY_DESCRIPTION_EX3, FABRIC_PARTITION_LOAD_INFORMATION_QUERY_DESCRIPTION,
19        FABRIC_SERVICE_PARTITION_QUERY_DESCRIPTION, FABRIC_SERVICE_QUERY_DESCRIPTION,
20        FABRIC_SERVICE_REPLICA_QUERY_DESCRIPTION,
21    },
22};
23
24use crate::types::{
25    DeployedServiceReplicaDetailQueryDescription, DeployedServiceReplicaDetailQueryResult,
26    GetPartitionLoadInformationResult, NodeListResult, NodeQueryDescription,
27    PartitionLoadInformationQueryDescription, ServicePartitionList,
28    ServicePartitionQueryDescription, ServiceReplicaList, ServiceReplicaQueryDescription,
29};
30use crate::{
31    runtime::executor::BoxedCancelToken,
32    sync::{FabricReceiver, fabric_begin_end_proxy},
33    types::ServiceQueryDescription,
34};
35
36#[derive(Debug, Clone)]
37pub struct QueryClient {
38    com: IFabricQueryClient10,
39}
40
41// Internal implementation block
42// Internal functions focuses on changing SF callback to async future,
43// while the public apis impl focuses on type conversion.
44
45impl QueryClient {
46    pub fn get_node_list_internal(
47        &self,
48        query_description: &FABRIC_NODE_QUERY_DESCRIPTION,
49        timeout_milliseconds: u32,
50        cancellation_token: Option<BoxedCancelToken>,
51    ) -> FabricReceiver<crate::WinResult<IFabricGetNodeListResult2>> {
52        let com1 = &self.com;
53        let com2 = self.com.clone();
54
55        fabric_begin_end_proxy(
56            move |callback| unsafe {
57                com1.BeginGetNodeList(query_description, timeout_milliseconds, callback)
58            },
59            move |ctx| unsafe { com2.EndGetNodeList2(ctx) },
60            cancellation_token,
61        )
62    }
63
64    pub fn get_application_list_internal(
65        &self,
66        query_description: &FABRIC_APPLICATION_QUERY_DESCRIPTION,
67        timeout_milliseconds: u32,
68        cancellation_token: Option<BoxedCancelToken>,
69    ) -> FabricReceiver<crate::WinResult<IFabricGetApplicationListResult2>> {
70        let com1 = &self.com;
71        let com2 = self.com.clone();
72        fabric_begin_end_proxy(
73            move |callback| unsafe {
74                com1.BeginGetApplicationList(query_description, timeout_milliseconds, callback)
75            },
76            move |ctx| unsafe { com2.EndGetApplicationList2(ctx) },
77            cancellation_token,
78        )
79    }
80
81    fn get_service_list_internal(
82        &self,
83        desc: &FABRIC_SERVICE_QUERY_DESCRIPTION,
84        timeout_milliseconds: u32,
85        cancellation_token: Option<BoxedCancelToken>,
86    ) -> FabricReceiver<crate::WinResult<mssf_com::FabricClient::IFabricGetServiceListResult2>>
87    {
88        let com1 = &self.com;
89        let com2 = self.com.clone();
90        fabric_begin_end_proxy(
91            move |callback| unsafe {
92                com1.BeginGetServiceList(desc, timeout_milliseconds, callback)
93            },
94            move |ctx| unsafe { com2.EndGetServiceList2(ctx) },
95            cancellation_token,
96        )
97    }
98
99    fn get_partition_list_internal(
100        &self,
101        desc: &FABRIC_SERVICE_PARTITION_QUERY_DESCRIPTION,
102        timeout_milliseconds: u32,
103        cancellation_token: Option<BoxedCancelToken>,
104    ) -> FabricReceiver<crate::WinResult<IFabricGetPartitionListResult2>> {
105        let com1 = &self.com;
106        let com2 = self.com.clone();
107        fabric_begin_end_proxy(
108            move |callback| unsafe {
109                com1.BeginGetPartitionList(desc, timeout_milliseconds, callback)
110            },
111            move |ctx| unsafe { com2.EndGetPartitionList2(ctx) },
112            cancellation_token,
113        )
114    }
115
116    fn get_replica_list_internal(
117        &self,
118        desc: &FABRIC_SERVICE_REPLICA_QUERY_DESCRIPTION,
119        timeout_milliseconds: u32,
120        cancellation_token: Option<BoxedCancelToken>,
121    ) -> FabricReceiver<crate::WinResult<IFabricGetReplicaListResult2>> {
122        let com1 = &self.com;
123        let com2 = self.com.clone();
124        fabric_begin_end_proxy(
125            move |callback| unsafe {
126                com1.BeginGetReplicaList(desc, timeout_milliseconds, callback)
127            },
128            move |ctx| unsafe { com2.EndGetReplicaList2(ctx) },
129            cancellation_token,
130        )
131    }
132
133    fn get_partition_load_information_internal(
134        &self,
135        desc: &FABRIC_PARTITION_LOAD_INFORMATION_QUERY_DESCRIPTION,
136        timeout_milliseconds: u32,
137        cancellation_token: Option<BoxedCancelToken>,
138    ) -> FabricReceiver<crate::WinResult<IFabricGetPartitionLoadInformationResult>> {
139        let com1 = &self.com;
140        let com2 = self.com.clone();
141        fabric_begin_end_proxy(
142            move |callback| unsafe {
143                com1.BeginGetPartitionLoadInformation(desc, timeout_milliseconds, callback)
144            },
145            move |ctx| unsafe { com2.EndGetPartitionLoadInformation(ctx) },
146            cancellation_token,
147        )
148    }
149
150    fn get_deployed_replica_detail_internal(
151        &self,
152        desc: &FABRIC_DEPLOYED_SERVICE_REPLICA_DETAIL_QUERY_DESCRIPTION,
153        timeout_milliseconds: u32,
154        cancellation_token: Option<BoxedCancelToken>,
155    ) -> FabricReceiver<crate::WinResult<IFabricGetDeployedServiceReplicaDetailResult>> {
156        let com1 = &self.com;
157        let com2 = self.com.clone();
158        fabric_begin_end_proxy(
159            move |callback| unsafe {
160                com1.BeginGetDeployedReplicaDetail(desc, timeout_milliseconds, callback)
161            },
162            move |ctx| unsafe { com2.EndGetDeployedReplicaDetail(ctx) },
163            cancellation_token,
164        )
165    }
166}
167
168impl From<IFabricQueryClient10> for QueryClient {
169    fn from(com: IFabricQueryClient10) -> Self {
170        Self { com }
171    }
172}
173
174impl From<QueryClient> for IFabricQueryClient10 {
175    fn from(value: QueryClient) -> Self {
176        value.com
177    }
178}
179
180impl QueryClient {
181    // List nodes in the cluster
182    pub async fn get_node_list(
183        &self,
184        desc: &NodeQueryDescription,
185        timeout: Duration,
186        cancellation_token: Option<BoxedCancelToken>,
187    ) -> crate::Result<NodeListResult> {
188        // Note that the SF raw structs are scoped to avoid having them across await points.
189        // This makes api Send. All FabricClient api should follow this pattern.
190        let com = {
191            let ex3 = FABRIC_NODE_QUERY_DESCRIPTION_EX3 {
192                MaxResults: desc.paged_query.max_results.unwrap_or(0),
193                Reserved: std::ptr::null_mut(),
194            };
195
196            let ex2 = FABRIC_NODE_QUERY_DESCRIPTION_EX2 {
197                NodeStatusFilter: desc.node_status_filter.bits(),
198                Reserved: std::ptr::addr_of!(ex3) as *mut c_void,
199            };
200
201            let ex1 = FABRIC_NODE_QUERY_DESCRIPTION_EX1 {
202                ContinuationToken: desc.paged_query.continuation_token.as_ref().into(),
203                Reserved: std::ptr::addr_of!(ex2) as *mut c_void,
204            };
205
206            let arg = FABRIC_NODE_QUERY_DESCRIPTION {
207                NodeNameFilter: desc.node_name_filter.as_ref().into(),
208                Reserved: std::ptr::addr_of!(ex1) as *mut c_void,
209            };
210            self.get_node_list_internal(
211                &arg,
212                timeout.as_millis().try_into().unwrap(),
213                cancellation_token,
214            )
215        }
216        .await??;
217        Ok(NodeListResult::from(&com))
218    }
219
220    pub async fn get_application_list(
221        &self,
222        desc: &crate::types::ApplicationQueryDescription,
223        timeout: Duration,
224        cancellation_token: Option<BoxedCancelToken>,
225    ) -> crate::Result<crate::types::ApplicationListResult> {
226        let com = {
227            let (mut base, mut ex1, mut ex2, mut ex3, ex4) = desc.get_raw_parts();
228            base.Reserved = std::ptr::addr_of!(ex1) as *mut c_void;
229            ex1.Reserved = std::ptr::addr_of!(ex2) as *mut c_void;
230            ex2.Reserved = std::ptr::addr_of!(ex3) as *mut c_void;
231            ex3.Reserved = std::ptr::addr_of!(ex4) as *mut c_void;
232            self.get_application_list_internal(
233                &base,
234                timeout.as_millis().try_into().unwrap(),
235                cancellation_token,
236            )
237        }
238        .await??;
239        Ok(crate::types::ApplicationListResult::from(&com))
240    }
241    pub async fn get_service_list(
242        &self,
243        desc: &ServiceQueryDescription,
244        timeout: Duration,
245        cancellation_token: Option<BoxedCancelToken>,
246    ) -> crate::Result<crate::types::ServiceListResult> {
247        let com = {
248            let (mut base, mut ex1, mut ex2, ex3) = desc.get_raw_parts();
249            base.Reserved = &ex1 as *const _ as *mut c_void;
250            ex1.Reserved = &ex2 as *const _ as *mut c_void;
251            ex2.Reserved = &ex3 as *const _ as *mut c_void;
252
253            self.get_service_list_internal(&base, timeout.as_millis() as u32, cancellation_token)
254        }
255        .await??;
256        Ok(crate::types::ServiceListResult::from(&com))
257    }
258
259    pub async fn get_partition_list(
260        &self,
261        desc: &ServicePartitionQueryDescription,
262        timeout: Duration,
263        cancellation_token: Option<BoxedCancelToken>,
264    ) -> crate::Result<ServicePartitionList> {
265        let com = {
266            let raw: FABRIC_SERVICE_PARTITION_QUERY_DESCRIPTION = desc.into();
267            let mili = timeout.as_millis() as u32;
268            self.get_partition_list_internal(&raw, mili, cancellation_token)
269        }
270        .await??;
271        Ok(ServicePartitionList::from(&com))
272    }
273
274    pub async fn get_replica_list(
275        &self,
276        desc: &ServiceReplicaQueryDescription,
277        timeout: Duration,
278        cancellation_token: Option<BoxedCancelToken>,
279    ) -> crate::Result<ServiceReplicaList> {
280        let com = {
281            let raw: FABRIC_SERVICE_REPLICA_QUERY_DESCRIPTION = desc.into();
282            let mili = timeout.as_millis() as u32;
283            self.get_replica_list_internal(&raw, mili, cancellation_token)
284        }
285        .await??;
286        Ok(ServiceReplicaList::from(&com))
287    }
288
289    pub async fn get_partition_load_information(
290        &self,
291        desc: &PartitionLoadInformationQueryDescription,
292        timeout: Duration,
293        cancellation_token: Option<BoxedCancelToken>,
294    ) -> crate::Result<GetPartitionLoadInformationResult> {
295        let com = {
296            let raw: FABRIC_PARTITION_LOAD_INFORMATION_QUERY_DESCRIPTION = desc.into();
297            let timeout_ms = timeout.as_micros() as u32;
298            self.get_partition_load_information_internal(&raw, timeout_ms, cancellation_token)
299        }
300        .await??;
301        Ok(GetPartitionLoadInformationResult::from(&com))
302    }
303
304    pub async fn get_deployed_replica_detail(
305        &self,
306        desc: &DeployedServiceReplicaDetailQueryDescription,
307        timeout: Duration,
308        cancellation_token: Option<BoxedCancelToken>,
309    ) -> crate::Result<DeployedServiceReplicaDetailQueryResult> {
310        let com = {
311            let raw: FABRIC_DEPLOYED_SERVICE_REPLICA_DETAIL_QUERY_DESCRIPTION = desc.into();
312            let timeout_ms = timeout.as_micros() as u32;
313            self.get_deployed_replica_detail_internal(&raw, timeout_ms, cancellation_token)
314        }
315        .await??;
316        Ok(DeployedServiceReplicaDetailQueryResult::new(com))
317    }
318}
mssf_core/client/query_client.rs

mssf_core/client/
query_client.rs
