nisshi_broker/
lib.rs

1// Copyright ⓒ 2024-2025 Peter Morgan <peter.james.morgan@gmail.com>
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7// http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::{
16    collections::HashMap,
17    env::vars,
18    fmt, io,
19    net::AddrParseError,
20    num::TryFromIntError,
21    result,
22    str::{FromStr, Utf8Error},
23    string::FromUtf8Error,
24    sync::{Arc, LazyLock, PoisonError},
25    time::{Duration, SystemTimeError},
26};
27
28use glob::PatternError;
29use jsonschema::ValidationError;
30use nisshi_sans_io::ErrorCode;
31use opentelemetry::{InstrumentationScope, global, metrics::Meter};
32use opentelemetry_otlp::ExporterBuildError;
33use opentelemetry_semantic_conventions::SCHEMA_URL;
34use regex::{Regex, Replacer};
35use thiserror::Error;
36use tokio::{sync::broadcast::error::SendError, task::JoinError};
37use tracing_subscriber::filter::ParseError;
38use url::Url;
39
40pub mod broker;
41pub mod coordinator;
42pub mod otel;
43pub mod service;
44
45#[derive(Copy, Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
46pub enum CancelKind {
47    Interrupt,
48    Terminate,
49}
50
51impl From<CancelKind> for Duration {
52    fn from(cancellation: CancelKind) -> Self {
53        Duration::from_millis(match cancellation {
54            CancelKind::Interrupt => 0,
55            CancelKind::Terminate => 5_000,
56        })
57    }
58}
59
60pub const NODE_ID: i32 = 111;
61
62pub(crate) static METER: LazyLock<Meter> = LazyLock::new(|| {
63    global::meter_with_scope(
64        InstrumentationScope::builder(env!("CARGO_PKG_NAME"))
65            .with_version(env!("CARGO_PKG_VERSION"))
66            .with_schema_url(SCHEMA_URL)
67            .build(),
68    )
69});
70
71#[derive(Clone, Debug, Error)]
72pub enum Error {
73    AddrParse(#[from] AddrParseError),
74    Api(ErrorCode),
75    Auth(#[from] nisshi_auth::Error),
76    Custom(String),
77    DuplicateApiService(i16),
78    EmptyCoordinatorWrapper,
79    EmptyJoinGroupRequestProtocol,
80    ExpectedJoinGroupRequestProtocol(&'static str),
81    ExporterBuild(Arc<ExporterBuildError>),
82
83    Hyper(Arc<hyper::http::Error>),
84    Io(Arc<io::Error>),
85    Join(Arc<JoinError>),
86    Json(Arc<serde_json::Error>),
87    KafkaProtocol(#[from] nisshi_sans_io::Error),
88
89    #[cfg(feature = "libsql")]
90    LibSql(Arc<libsql::Error>),
91
92    Message(String),
93    Model(#[from] nisshi_model::Error),
94
95    #[cfg(feature = "dynostore")]
96    ObjectStore(Arc<object_store::Error>),
97
98    ParseFilter(Arc<ParseError>),
99    ParseInt(#[from] std::num::ParseIntError),
100    Pattern(Arc<PatternError>),
101    Poison,
102
103    #[cfg(feature = "postgres")]
104    Pool(Arc<deadpool_postgres::PoolError>),
105
106    Regex(#[from] regex::Error),
107
108    SchemaRegistry(Arc<nisshi_schema::Error>),
109    Service(#[from] nisshi_service::Error),
110    Storage(#[from] nisshi_storage::Error),
111    StringUtf8(#[from] FromUtf8Error),
112    SystemTime(#[from] SystemTimeError),
113
114    #[cfg(feature = "postgres")]
115    TokioPostgres(Arc<tokio_postgres::error::Error>),
116    TryFromInt(#[from] TryFromIntError),
117
118    #[cfg(feature = "turso")]
119    Turso(Arc<turso::Error>),
120
121    UnsupportedApiService(i16),
122    UnsupportedStorageUrl(Url),
123    UnsupportedTracingFormat(String),
124    Url(#[from] url::ParseError),
125    Utf8(#[from] Utf8Error),
126    Uuid(#[from] uuid::Error),
127    SchemaValidation,
128    Send(Arc<SendError<CancelKind>>),
129}
130
131#[cfg(feature = "libsql")]
132impl From<libsql::Error> for Error {
133    fn from(value: libsql::Error) -> Self {
134        Self::from(Arc::new(value))
135    }
136}
137
138#[cfg(feature = "libsql")]
139impl From<Arc<libsql::Error>> for Error {
140    fn from(value: Arc<libsql::Error>) -> Self {
141        Self::LibSql(value)
142    }
143}
144
145#[cfg(feature = "turso")]
146impl From<turso::Error> for Error {
147    fn from(value: turso::Error) -> Self {
148        Self::from(Arc::new(value))
149    }
150}
151
152#[cfg(feature = "turso")]
153impl From<Arc<turso::Error>> for Error {
154    fn from(value: Arc<turso::Error>) -> Self {
155        Self::Turso(value)
156    }
157}
158
159impl From<PatternError> for Error {
160    fn from(value: PatternError) -> Self {
161        Self::Pattern(Arc::new(value))
162    }
163}
164
165impl From<ExporterBuildError> for Error {
166    fn from(value: ExporterBuildError) -> Self {
167        Self::ExporterBuild(Arc::new(value))
168    }
169}
170
171impl From<SendError<CancelKind>> for Error {
172    fn from(value: SendError<CancelKind>) -> Self {
173        Self::Send(Arc::new(value))
174    }
175}
176
177#[cfg(feature = "postgres")]
178impl From<tokio_postgres::error::Error> for Error {
179    fn from(value: tokio_postgres::error::Error) -> Self {
180        Self::from(Arc::new(value))
181    }
182}
183
184#[cfg(feature = "postgres")]
185impl From<Arc<tokio_postgres::error::Error>> for Error {
186    fn from(value: Arc<tokio_postgres::error::Error>) -> Self {
187        Self::TokioPostgres(value)
188    }
189}
190
191impl From<hyper::http::Error> for Error {
192    fn from(value: hyper::http::Error) -> Self {
193        Self::Hyper(Arc::new(value))
194    }
195}
196
197impl From<JoinError> for Error {
198    fn from(value: JoinError) -> Self {
199        Self::Join(Arc::new(value))
200    }
201}
202
203impl From<serde_json::Error> for Error {
204    fn from(value: serde_json::Error) -> Self {
205        Self::from(Arc::new(value))
206    }
207}
208
209impl From<Arc<serde_json::Error>> for Error {
210    fn from(value: Arc<serde_json::Error>) -> Self {
211        Self::Json(value)
212    }
213}
214
215#[cfg(feature = "dynostore")]
216impl From<object_store::Error> for Error {
217    fn from(value: object_store::Error) -> Self {
218        Self::from(Arc::new(value))
219    }
220}
221
222#[cfg(feature = "dynostore")]
223impl From<Arc<object_store::Error>> for Error {
224    fn from(value: Arc<object_store::Error>) -> Self {
225        Self::ObjectStore(value)
226    }
227}
228
229impl From<ParseError> for Error {
230    fn from(value: ParseError) -> Self {
231        Self::ParseFilter(Arc::new(value))
232    }
233}
234
235#[cfg(feature = "postgres")]
236impl From<deadpool_postgres::PoolError> for Error {
237    fn from(value: deadpool_postgres::PoolError) -> Self {
238        Self::from(Arc::new(value))
239    }
240}
241
242#[cfg(feature = "postgres")]
243impl From<Arc<deadpool_postgres::PoolError>> for Error {
244    fn from(value: Arc<deadpool_postgres::PoolError>) -> Self {
245        Self::Pool(value)
246    }
247}
248
249impl From<nisshi_schema::Error> for Error {
250    fn from(value: nisshi_schema::Error) -> Self {
251        Self::SchemaRegistry(Arc::new(value))
252    }
253}
254
255impl From<io::Error> for Error {
256    fn from(value: io::Error) -> Self {
257        Self::Io(Arc::new(value))
258    }
259}
260
261impl<T> From<PoisonError<T>> for Error {
262    fn from(_value: PoisonError<T>) -> Self {
263        Self::Poison
264    }
265}
266
267impl From<ValidationError<'_>> for Error {
268    fn from(_value: ValidationError<'_>) -> Self {
269        Self::SchemaValidation
270    }
271}
272
273impl fmt::Display for Error {
274    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
275        write!(f, "{self:?}")
276    }
277}
278
279pub type Result<T, E = Error> = result::Result<T, E>;
280
281#[derive(Copy, Clone, Debug)]
282pub enum TracingFormat {
283    Text,
284    Json,
285}
286
287impl FromStr for TracingFormat {
288    type Err = Error;
289
290    fn from_str(s: &str) -> Result<Self, Self::Err> {
291        match s {
292            "text" => Ok(Self::Text),
293            "json" => Ok(Self::Json),
294            otherwise => Err(Error::UnsupportedTracingFormat(otherwise.to_owned())),
295        }
296    }
297}
298
299#[derive(Clone, Debug)]
300pub struct VarRep(HashMap<String, String>);
301
302impl From<HashMap<String, String>> for VarRep {
303    fn from(value: HashMap<String, String>) -> Self {
304        Self(value)
305    }
306}
307
308impl VarRep {
309    fn replace(&self, haystack: &str) -> Result<String> {
310        Regex::new(r"\$\{(?<var>[^\}]+)\}")
311            .map(|re| re.replace(haystack, self).into_owned())
312            .map_err(Into::into)
313    }
314}
315
316impl Replacer for &VarRep {
317    fn replace_append(&mut self, caps: &regex::Captures<'_>, dst: &mut String) {
318        if let Some(variable) = caps.name("var")
319            && let Some(value) = self.0.get(variable.as_str())
320        {
321            dst.push_str(value);
322        }
323    }
324}
325
326#[derive(Clone, Debug)]
327pub struct EnvVarExp<T>(T);
328
329impl<T> EnvVarExp<T> {
330    pub fn into_inner(self) -> T {
331        self.0
332    }
333}
334
335impl<T> FromStr for EnvVarExp<T>
336where
337    T: FromStr,
338    Error: From<<T as FromStr>::Err>,
339{
340    type Err = Error;
341
342    fn from_str(s: &str) -> Result<Self, Self::Err> {
343        VarRep::from(vars().collect::<HashMap<_, _>>())
344            .replace(s)
345            .and_then(|s| T::from_str(&s).map_err(Into::into))
346            .map(|t| Self(t))
347    }
348}
nisshi_broker/lib.rs

nisshi_broker/
lib.rs
