basilisk_rust_client/
basilisk_client.rs

1use crate::bus_client::{BusClient, ForwardRequest};
2use crate::error::ClientResult;
3use crate::gateway_api::{AuthInfo, GatewayApiClient, InstanceInfo, RegistrationRequest};
4use crate::protocol::{ServiceBusEventEnvelope, ServiceBusForwardResponse};
5use std::collections::HashMap;
6
7/// Configuration used by `BasiliskClient::connect`.
8#[derive(Debug, Clone)]
9pub struct BasiliskClientConfig {
10    /// Base URL for gateway registry APIs.
11    pub gateway_base_url: String,
12    /// Service bus host.
13    pub bus_host: String,
14    /// Service bus TCP port.
15    pub bus_port: u16,
16    /// Logical service identifier.
17    pub service_id: String,
18    /// Service fingerprint/version marker.
19    pub fingerprint: String,
20    /// Path prefixes advertised to the gateway.
21    pub path_prefixes: Vec<String>,
22    /// Upstream URL scheme.
23    pub scheme: String,
24    /// Upstream host.
25    pub host: String,
26    /// Upstream port.
27    pub port: u16,
28    /// Load-balancing weight for this instance.
29    pub weight: i32,
30    /// Registration auth type (for example `token`).
31    pub registration_auth_type: String,
32    /// Registration token/secret for gateway auth.
33    pub registration_token: String,
34}
35
36/// High-level client that combines gateway registration APIs and the service-bus client.
37#[derive(Clone)]
38pub struct BasiliskClient {
39    /// Gateway API client handle.
40    pub gateway: GatewayApiClient,
41    /// Service bus client handle.
42    pub bus: BusClient,
43    /// Current service id.
44    pub service_id: String,
45    /// Registered instance id.
46    pub instance_id: String,
47}
48
49impl BasiliskClient {
50    /// Registers the instance with the gateway and opens an authenticated bus connection.
51    pub async fn connect(config: BasiliskClientConfig) -> anyhow::Result<Self> {
52        let BasiliskClientConfig {
53            gateway_base_url,
54            bus_host,
55            bus_port,
56            service_id,
57            fingerprint,
58            path_prefixes,
59            scheme,
60            host,
61            port,
62            weight,
63            registration_auth_type,
64            registration_token,
65        } = config;
66
67        let gateway = GatewayApiClient::new(gateway_base_url);
68        let registration = RegistrationRequest {
69            service_id: service_id.clone(),
70            fingerprint,
71            path_prefixes,
72            instance: InstanceInfo {
73                instance_id: String::new(),
74                scheme,
75                host,
76                port,
77                weight,
78            },
79            auth: AuthInfo {
80                auth_type: registration_auth_type,
81                token: registration_token,
82            },
83        };
84
85        let registration_response = gateway.register_instance_auto(&registration).await?;
86        let instance_id = registration_response.instance_id;
87        let token = registration_response.token;
88        let bus = BusClient::connect(
89            &bus_host,
90            bus_port,
91            service_id.clone(),
92            instance_id.clone(),
93            token,
94        )
95        .await?;
96
97        Ok(Self {
98            gateway,
99            bus,
100            service_id,
101            instance_id,
102        })
103    }
104
105    /// Deregisters this instance from the gateway registry.
106    pub async fn deregister(&self) -> anyhow::Result<()> {
107        self.gateway
108            .deregister_instance(&self.service_id, &self.instance_id)
109            .await
110    }
111
112    /// Subscribes this client to the provided topics.
113    pub async fn subscribe(&self, topics: Vec<String>) -> ClientResult<()> {
114        self.bus.subscribe(topics).await
115    }
116
117    /// Unsubscribes this client from the provided topics.
118    pub async fn unsubscribe(&self, topics: Vec<String>) -> ClientResult<()> {
119        self.bus.unsubscribe(topics).await
120    }
121
122    /// Publishes an event and returns the number of subscribers that received it.
123    pub async fn publish(
124        &self,
125        topic: impl Into<String>,
126        message_type: impl Into<String>,
127        payload: HashMap<String, serde_json::Value>,
128    ) -> ClientResult<i32> {
129        self.bus.publish(topic, message_type, payload).await
130    }
131
132    /// Publishes a fully constructed event envelope.
133    pub async fn publish_event(&self, event: ServiceBusEventEnvelope) -> ClientResult<i32> {
134        self.bus.publish_event(event).await
135    }
136
137    /// Sends a forward request and waits for a forward response.
138    pub async fn forward(
139        &self,
140        request: ForwardRequest,
141    ) -> ClientResult<ServiceBusForwardResponse> {
142        self.bus.forward(request).await
143    }
144
145    /// Registers an async event handler for a topic.
146    pub async fn on_event<F, Fut>(&self, topic: impl Into<String>, handler: F) -> ClientResult<()>
147    where
148        F: Fn(ServiceBusEventEnvelope) -> Fut + Send + Sync + 'static,
149        Fut: Future<Output = ()> + Send + 'static,
150    {
151        self.bus.on_event(topic, handler).await
152    }
153
154    /// Registers an async request responder keyed by message type.
155    pub async fn on_request<F, Fut>(
156        &self,
157        topic: impl Into<String>,
158        responder: F,
159    ) -> ClientResult<()>
160    where
161        F: Fn(crate::bus_client::ServiceBusRequest, crate::bus_client::RequestResponder) -> Fut
162            + Send
163            + Sync
164            + 'static,
165        Fut: Future<Output = ClientResult<()>> + Send + 'static,
166    {
167        self.bus.on_request(topic, responder).await
168    }
169}
basilisk_rust_client/basilisk_client.rs

basilisk_rust_client/
basilisk_client.rs
