temporalio_client/
plugins.rs

1//! Experimental client plugin APIs.
2
3use crate::{ClientOptions, ConnectionOptions};
4use std::{any::Any, error::Error, sync::Arc};
5
6/// An error returned by a plugin configuration hook.
7#[derive(Debug, thiserror::Error)]
8#[error(transparent)]
9pub struct PluginError(Box<dyn Error + Send + Sync>);
10
11impl PluginError {
12    /// Wrap an error returned by a plugin.
13    pub fn new(error: impl Into<Box<dyn Error + Send + Sync>>) -> Self {
14        Self(error.into())
15    }
16}
17
18/// The configuration target being modified when a plugin failed.
19#[derive(Clone, Copy, Debug, Eq, PartialEq, derive_more::Display)]
20#[non_exhaustive]
21pub enum PluginTarget {
22    /// Connection options.
23    #[display("connection options")]
24    Connection,
25    /// Namespace-bound client options.
26    #[display("client options")]
27    Client,
28    /// Worker options.
29    #[display("worker options")]
30    Worker,
31    /// Workflow replayer options.
32    #[display("workflow replayer options")]
33    WorkflowReplayer,
34}
35
36/// An error applying a named plugin to a configuration target.
37#[derive(Debug, thiserror::Error)]
38#[error("plugin '{plugin_name}' failed to configure {target}: {source}")]
39#[non_exhaustive]
40pub struct PluginApplyError {
41    /// The plugin name reported by [`ClientPlugin::name`] or its worker equivalent.
42    pub plugin_name: String,
43    /// The configuration target being modified.
44    pub target: PluginTarget,
45    /// The error returned by the plugin.
46    #[source]
47    pub source: PluginError,
48}
49
50impl PluginApplyError {
51    /// Create an error for a plugin that failed to configure a target.
52    ///
53    /// **Internal:** This method is intended to be used during worker construction. Arguments can
54    /// change or be removed in breaking manners.
55    pub fn new(plugin_name: impl Into<String>, target: PluginTarget, source: PluginError) -> Self {
56        Self {
57            plugin_name: plugin_name.into(),
58            target,
59            source,
60        }
61    }
62}
63
64/// Configures connection and namespace-bound client options.
65///
66/// **Experimental:** This API may change or be removed.
67pub trait ClientPlugin: Send + Sync + 'static {
68    /// Return the stable, unique name used to identify this plugin in diagnostics and worker
69    /// heartbeats.
70    fn name(&self) -> &str;
71
72    /// Configure options before the connection is established.
73    fn configure_connection_options(
74        &self,
75        _options: &mut ConnectionOptions,
76    ) -> Result<(), PluginError> {
77        Ok(())
78    }
79
80    /// Configure options before the namespace-bound client is created.
81    fn configure_client_options(&self, _options: &mut ClientOptions) -> Result<(), PluginError> {
82        Ok(())
83    }
84}
85
86/// Marks opaque worker-plugin propagation data supplied by an SDK integration.
87///
88/// Implementing this trait will not make a type recognized as plugin. Only SDK known
89/// `WorkerPluginExtension` implementers will be used as plugins.
90pub trait WorkerPluginData: Any + Send + Sync + 'static {}
91
92/// A type-erased client plugin and worker-plugin propagation data.
93///
94/// The worker data is intentionally opaque to avoid taking a dependency on `temporalio-sdk`.
95///
96/// **Experimental:** This API may change or be removed.
97#[derive(Clone)]
98pub struct ErasedClientPlugin {
99    client: Arc<dyn ClientPlugin>,
100    worker_plugins: Vec<Arc<dyn WorkerPluginData>>,
101}
102
103impl ErasedClientPlugin {
104    /// Type-erase a client plugin for registration on [`ClientOptions`].
105    pub fn new<P: ClientPlugin>(plugin: P) -> Self {
106        Self {
107            client: Arc::new(plugin),
108            worker_plugins: Vec::new(),
109        }
110    }
111
112    /// Attach opaque worker-plugin propagation data.
113    ///
114    /// This is intended for SDK integrations that define their own worker plugin trait. Values
115    /// with types unknown to an SDK are ignored.
116    pub fn with_worker_plugin<T: WorkerPluginData>(mut self, plugin: T) -> Self {
117        self.worker_plugins.push(Arc::new(plugin));
118        self
119    }
120
121    /// Iterate over opaque worker-plugin propagation data.
122    ///
123    /// This is intended for SDK integrations that recognize their own private registration type.
124    pub fn worker_plugins(&self) -> impl Iterator<Item = &dyn WorkerPluginData> {
125        self.worker_plugins.iter().map(AsRef::as_ref)
126    }
127
128    /// Return the stable name so SDK integrations can report client-only plugins in worker
129    /// metadata.
130    ///
131    /// **Experimental:** This API may change or be removed.
132    pub fn name(&self) -> &str {
133        self.client.name()
134    }
135
136    pub(crate) fn plugin(&self) -> &dyn ClientPlugin {
137        self.client.as_ref()
138    }
139}
140
141pub(crate) fn apply_connection_plugins(
142    client_options: &ClientOptions,
143    connection_options: &mut ConnectionOptions,
144) -> Result<(), PluginApplyError> {
145    for registration in client_options.plugins() {
146        registration
147            .plugin()
148            .configure_connection_options(connection_options)
149            .map_err(|source| {
150                PluginApplyError::new(
151                    registration.plugin().name(),
152                    PluginTarget::Connection,
153                    source,
154                )
155            })?;
156    }
157    Ok(())
158}
159
160pub(crate) fn apply_client_plugins(options: &mut ClientOptions) -> Result<(), PluginApplyError> {
161    if options.client_plugins_applied() {
162        return Ok(());
163    }
164    let plugins = options.plugins().to_vec();
165    for registration in plugins {
166        registration
167            .plugin()
168            .configure_client_options(options)
169            .map_err(|source| {
170                PluginApplyError::new(registration.plugin().name(), PluginTarget::Client, source)
171            })?;
172    }
173    options.mark_client_plugins_applied();
174    Ok(())
175}
176
177#[cfg(test)]
178mod tests {
179    use super::*;
180    use std::sync::atomic::{AtomicUsize, Ordering};
181    use url::Url;
182
183    struct CountingPlugin {
184        connection_calls: Arc<AtomicUsize>,
185        client_calls: Arc<AtomicUsize>,
186    }
187
188    impl ClientPlugin for CountingPlugin {
189        fn name(&self) -> &str {
190            "counting"
191        }
192
193        fn configure_connection_options(
194            &self,
195            options: &mut ConnectionOptions,
196        ) -> Result<(), PluginError> {
197            self.connection_calls.fetch_add(1, Ordering::Relaxed);
198            options.identity.push_str("-configured");
199            Ok(())
200        }
201
202        fn configure_client_options(&self, options: &mut ClientOptions) -> Result<(), PluginError> {
203            self.client_calls.fetch_add(1, Ordering::Relaxed);
204            options.namespace.push_str("-configured");
205            Ok(())
206        }
207    }
208
209    #[test]
210    fn plugins_follow_target_lifecycles() {
211        let connection_calls = Arc::new(AtomicUsize::new(0));
212        let client_calls = Arc::new(AtomicUsize::new(0));
213        let mut client_options = ClientOptions::new("namespace")
214            .client_plugin(CountingPlugin {
215                connection_calls: connection_calls.clone(),
216                client_calls: client_calls.clone(),
217            })
218            .build();
219        let mut first_connection_options =
220            ConnectionOptions::new(Url::parse("http://localhost:7233").unwrap())
221                .identity("first")
222                .build();
223        let mut second_connection_options =
224            ConnectionOptions::new(Url::parse("http://localhost:7233").unwrap())
225                .identity("second")
226                .build();
227
228        apply_connection_plugins(&client_options, &mut first_connection_options).unwrap();
229        apply_client_plugins(&mut client_options).unwrap();
230        apply_connection_plugins(&client_options, &mut second_connection_options).unwrap();
231        apply_client_plugins(&mut client_options).unwrap();
232
233        assert_eq!(connection_calls.load(Ordering::Relaxed), 2);
234        assert_eq!(client_calls.load(Ordering::Relaxed), 1);
235        assert_eq!(first_connection_options.identity, "first-configured");
236        assert_eq!(second_connection_options.identity, "second-configured");
237        assert_eq!(client_options.namespace, "namespace-configured");
238    }
239}
temporalio_client/plugins.rs

temporalio_client/
plugins.rs
