use std::sync::Arc;
use photon_telemetry::{install_ops_log, OpsLog};
use photon_backend::{
instrumentation, BackendContext, EmbeddedBackend, ExecutorServices, GenericPhotonBackend,
InProcStoragePort, PhotonBackend, PhotonError, Result, RetentionHook, RetentionPolicy,
StoragePort, TopicRegistry, TransportCrypto,
};
use crate::executor::ExecutorController;
use crate::{Photon, PhotonRuntimeState};
type BackendInstallFn =
Box<dyn FnOnce(BackendContext) -> Result<Arc<dyn PhotonBackend>> + Send>;
#[derive(Default)]
pub struct PhotonBuilder {
storage_port: Option<Arc<dyn StoragePort>>,
backend: Option<Arc<dyn PhotonBackend>>,
backend_install: Option<BackendInstallFn>,
use_auto_registry: bool,
ops_log: Option<Arc<dyn OpsLog>>,
retention_policy: Option<RetentionPolicy>,
retention_hook: Option<Arc<dyn RetentionHook>>,
}
impl PhotonBuilder {
#[must_use]
pub fn storage_port(mut self, port: Arc<dyn StoragePort>) -> Self {
self.storage_port = Some(port);
self
}
#[must_use]
pub fn backend(mut self, backend: Arc<dyn PhotonBackend>) -> Self {
self.backend = Some(backend);
self.backend_install = None;
self
}
#[must_use]
pub fn backend_with_context<F>(mut self, install: F) -> Self
where
F: FnOnce(BackendContext) -> Result<Arc<dyn PhotonBackend>> + Send + 'static,
{
self.backend = None;
self.backend_install = Some(Box::new(install));
self
}
#[must_use]
pub fn mem_backend(mut self) -> Self {
self.backend = None;
self.backend_install = Some(Box::new(EmbeddedBackend::install_mem));
self
}
#[must_use]
pub fn ops_log(mut self, log: impl OpsLog + 'static) -> Self {
self.ops_log = Some(Arc::new(log));
self
}
#[must_use]
pub fn ops_log_arc(mut self, log: Arc<dyn OpsLog>) -> Self {
self.ops_log = Some(log);
self
}
#[must_use]
pub const fn auto_registry(mut self) -> Self {
self.use_auto_registry = true;
self
}
#[must_use]
pub fn retention_policy(mut self, policy: RetentionPolicy) -> Self {
self.retention_policy = Some(policy);
self
}
#[must_use]
pub fn retention_hook(mut self, hook: Arc<dyn RetentionHook>) -> Self {
self.retention_hook = Some(hook);
self
}
pub fn build(self) -> Result<Photon> {
if let Some(log) = self.ops_log {
install_ops_log(log);
}
let registry = if self.use_auto_registry {
TopicRegistry::auto_discover()
} else {
TopicRegistry::new()
};
let port = match self.storage_port {
Some(port) => port,
None => Arc::new(InProcStoragePort::new(TransportCrypto::from_env()?)),
};
let ctx = BackendContext {
registry: registry.clone(),
};
let backend = match (self.backend, self.backend_install) {
(Some(b), None) => b,
(None, Some(install)) => install(ctx)?,
(None, None) => GenericPhotonBackend::install_with_port(
BackendContext { registry },
Arc::clone(&port),
)?,
(Some(_), Some(_)) => {
return Err(PhotonError::Internal(
"PhotonBuilder: set backend() or backend_with_context(), not both".into(),
));
}
};
let retention_policy = self.retention_policy.unwrap_or_default();
let runtime = PhotonRuntimeState {
storage_port: Arc::clone(&port),
executor_services: Arc::new(ExecutorServices::new(
port,
retention_policy,
self.retention_hook,
)),
executor: Arc::new(ExecutorController::default()),
};
let backend = instrumentation::wrap_backend(backend);
Ok(Photon::new(backend, runtime))
}
}