harmos_sdk/operations/
registration.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17use std::any::Any;
18use std::future::Future;
19use std::pin::Pin;
20
21// Third Party Imports
22
23// Local Imports
24use super::{
25    Carrier, Contract, Dependencies, Executable, ExecutionContext, OperationDeclaration, OperationDefinition, Resource, ServingError,
26    WireDeclaration, WireTable,
27};
28use crate::services::Service;
29use crate::{Message, ServicePolicy};
30
31/* === Constants === */
32
33/* === Macros === */
34
35/* === Type Aliases === */
36
37pub type ServiceFuture<'a, T> = Pin<Box<dyn Future<Output = Result<T, ServingError>> + Send + 'a>>;
38type Install = fn(&mut WireTable, Dependencies, Contract) -> Result<(), ServingError>;
39type InitializeService = fn(ExecutionContext) -> ServiceFuture<'static, Box<dyn ServiceInstance>>;
40type PrepareNative = fn(Dependencies, Box<dyn Any + Send>, Box<dyn Any + Send>) -> Result<Box<dyn NativeInstance>, ServingError>;
41type PrepareEncoded = fn(Dependencies, &[u8]) -> Result<PreparedNative, ServingError>;
42
43/* === Structs === */
44
45#[derive(Clone)]
46pub struct RegistrationBatch {
47    pub dependencies: Dependencies,
48    pub operations: Vec<OperationFactory>,
49    pub services: Vec<ServiceFactory>,
50}
51
52pub struct PreparedNative {
53    #[cfg(feature = "streaming")]
54    pub streams: Option<super::WireStreams>,
55    pub instance: Box<dyn NativeInstance>,
56    pub grant: Option<crate::plugin::Grant>,
57    pub key: Option<super::CacheKey>,
58}
59
60#[derive(Clone)]
61pub struct OperationFactory {
62    pub definition: OperationDefinition,
63    pub operations: &'static [WireDeclaration],
64    declared: fn() -> OperationDefinition,
65    builtin: bool,
66    resources: &'static [Resource],
67    pub dependencies: Dependencies,
68    contract: fn(&harmos_domain::config::SchemaSettings) -> Result<Contract, super::ContractError>,
69    install: Install,
70    native: PrepareNative,
71    endpoints_valid: fn() -> bool,
72    native_types: fn() -> (std::any::TypeId, std::any::TypeId, std::any::TypeId),
73    encoded: PrepareEncoded,
74    #[cfg(all(feature = "grpc", feature = "streaming"))]
75    grpc: fn(Dependencies, &[u8], super::WireStreamLimits) -> Result<PreparedNative, ServingError>,
76    #[cfg(all(feature = "grpc", feature = "streaming"))]
77    grpc_memory: fn(&[u8]) -> Result<usize, crate::MessageError>,
78    decoding_memory: fn(&[u8]) -> Result<usize, crate::MessageError>,
79}
80
81#[derive(Clone)]
82pub struct ServiceFactory {
83    pub local_id: String,
84    pub policy: ServicePolicy,
85    pub dependencies: Dependencies,
86    initialize: InitializeService,
87}
88
89struct InitializedService<S: Service> {
90    service: S,
91    dependencies: Dependencies,
92}
93
94struct NativeOperation<O: Carrier> {
95    dependencies: Dependencies,
96    operation: Option<O>,
97    request: <O::Declaration as OperationDeclaration>::Request,
98    endpoints: O::StreamEndpoints,
99    stopped: bool,
100    encoded: bool,
101    #[cfg(all(feature = "grpc", feature = "streaming"))]
102    protobuf: bool,
103    #[cfg(feature = "streaming")]
104    streams: Option<super::WireStreams>,
105    #[cfg(feature = "streaming")]
106    epoch: u64,
107}
108
109/* === Enums === */
110
111/* === Traits === */
112
113pub trait ServiceInstance: Send {
114    fn run(&mut self, context: ExecutionContext) -> ServiceFuture<'_, ()>;
115    fn stop(&mut self) -> ServiceFuture<'_, ()>;
116    fn cleanup(&mut self) -> ServiceFuture<'_, ()>;
117}
118
119pub trait NativeInstance: Send {
120    fn classify(&self, error: &(dyn std::error::Error + Send + Sync + 'static)) -> crate::operations::FailureKind;
121    fn initialize(&mut self, context: ExecutionContext) -> ServiceFuture<'_, ()>;
122    fn run(&mut self, context: ExecutionContext) -> ServiceFuture<'_, super::response::ErasedResponse>;
123    fn stop(&mut self) -> ServiceFuture<'_, ()>;
124    fn cleanup(&mut self) -> ServiceFuture<'_, ()>;
125    fn transferred(&self) -> bool;
126}
127
128/* === Implementations === */
129
130impl RegistrationBatch {
131    pub fn new(dependencies: Dependencies) -> Self {
132        Self {
133            dependencies,
134            operations: Vec::new(),
135            services: Vec::new(),
136        }
137    }
138
139    pub fn operation<O: Executable>(self) -> Self {
140        self.with::<O, O::Carrier>(O::RESOURCES)
141    }
142
143    #[doc(hidden)]
144    pub fn carrier<D, O>(self) -> Self
145    where
146        D: OperationDeclaration,
147        O: Carrier<Declaration = D>,
148    {
149        self.with::<D, O>(&[])
150    }
151
152    fn with<D, O>(mut self, resources: &'static [Resource]) -> Self
153    where
154        D: OperationDeclaration,
155        O: Carrier<Declaration = D>,
156    {
157        self.operations.push(OperationFactory {
158            resources,
159            operations: D::OPERATIONS,
160            definition: D::definition(),
161            declared: D::definition,
162            builtin: std::any::TypeId::of::<D>() == std::any::TypeId::of::<crate::observers::Control>(),
163            dependencies: self.dependencies.clone(),
164            contract: |settings| {
165                if std::any::TypeId::of::<D>() == std::any::TypeId::of::<crate::observers::Control>() {
166                    return crate::observers::contract();
167                }
168                WireDeclaration::of::<D>().contract_with_settings(settings)
169            },
170            install: WireTable::install::<D, O>,
171            native_types: || {
172                (
173                    std::any::TypeId::of::<<O::Declaration as OperationDeclaration>::Request>(),
174                    std::any::TypeId::of::<<O::Declaration as OperationDeclaration>::Response>(),
175                    std::any::TypeId::of::<O::StreamEndpoints>(),
176                )
177            },
178            endpoints_valid: || {
179                use super::EndpointSet;
180                O::StreamEndpoints::item_types() == super::item_types::<D>()
181            },
182            native: |dependencies, request, endpoints| {
183                use super::EndpointSet;
184                if O::StreamEndpoints::item_types() != super::item_types::<D>() {
185                    return Err(ServingError::Catalog);
186                }
187                let request = request.downcast::<<O::Declaration as OperationDeclaration>::Request>().map_err(|_| ServingError::Catalog)?;
188                let endpoints = endpoints.downcast::<O::StreamEndpoints>().map_err(|_| ServingError::Catalog)?;
189                Ok(Box::new(NativeOperation::<O> {
190                    dependencies,
191                    operation: None,
192                    request: *request,
193                    endpoints: *endpoints,
194                    stopped: false,
195                    encoded: false,
196                    #[cfg(all(feature = "grpc", feature = "streaming"))]
197                    protobuf: false,
198                    #[cfg(feature = "streaming")]
199                    streams: None,
200                    #[cfg(feature = "streaming")]
201                    epoch: 0,
202                }))
203            },
204            encoded: |dependencies, payload| {
205                use super::EndpointSet;
206                let endpoints = O::StreamEndpoints::empty().ok_or(ServingError::StreamUnsupported)?;
207                let request =
208                    <<O::Declaration as OperationDeclaration>::Request as crate::Message>::decode(payload).map_err(ServingError::Codec)?;
209                let grant = D::grant(&request);
210                let key = D::key(&request);
211                let instance = NativeOperation::<O> {
212                    dependencies,
213                    operation: None,
214                    request,
215                    endpoints,
216                    stopped: false,
217                    encoded: true,
218                    #[cfg(all(feature = "grpc", feature = "streaming"))]
219                    protobuf: false,
220                    #[cfg(feature = "streaming")]
221                    streams: None,
222                    #[cfg(feature = "streaming")]
223                    epoch: 0,
224                };
225                Ok(PreparedNative {
226                    instance: Box::new(instance),
227                    grant,
228                    key,
229                    #[cfg(feature = "streaming")]
230                    streams: None,
231                })
232            },
233            #[cfg(all(feature = "grpc", feature = "streaming"))]
234            grpc: |dependencies, payload, limits| {
235                use super::EndpointSet;
236                let request =
237                    crate::grpc::decode::<<O::Declaration as OperationDeclaration>::Request>(crate::grpc::DecodeInput::new(payload))
238                        .map_err(ServingError::Codec)?;
239                let (endpoints, streams) = match O::StreamEndpoints::empty() {
240                    Some(endpoints) => (endpoints, None),
241                    None => {
242                        let (endpoints, streams) = O::StreamEndpoints::bind_wire(limits)?;
243                        (endpoints, Some(streams.protobuf(super::serving::protobuf::Codec::of::<D>())?))
244                    }
245                };
246                let grant = D::grant(&request);
247                let key = D::key(&request);
248                let instance = NativeOperation::<O> {
249                    dependencies,
250                    operation: None,
251                    request,
252                    endpoints,
253                    stopped: false,
254                    encoded: true,
255                    protobuf: true,
256                    streams: streams.clone(),
257                    epoch: 0,
258                };
259                Ok(PreparedNative {
260                    instance: Box::new(instance),
261                    grant,
262                    key,
263                    streams,
264                })
265            },
266            #[cfg(all(feature = "grpc", feature = "streaming"))]
267            grpc_memory: |payload| {
268                crate::grpc::prepare::<<O::Declaration as OperationDeclaration>::Request>(crate::grpc::DecodeInput::new(payload))
269                    .map(|prepared| prepared.memory())
270            },
271            decoding_memory: <D::Request as crate::Message>::decoding_memory,
272        });
273        self
274    }
275
276    pub fn service<S: Service>(mut self, local_id: &str, policy: ServicePolicy) -> Self {
277        self.services.push(ServiceFactory {
278            local_id: local_id.to_owned(),
279            policy,
280            dependencies: self.dependencies.clone(),
281            initialize: |context| {
282                Box::pin(async move {
283                    let dependencies = context.dependencies().clone();
284                    let service = S::initialize(context).await.map_err(service_error)?;
285                    Ok(Box::new(InitializedService { service, dependencies }) as Box<dyn ServiceInstance>)
286                })
287            },
288        });
289        self
290    }
291}
292
293impl OperationFactory {
294    pub fn accepts_target(&self, target: &super::Target) -> bool {
295        (self.native_types)() == ((target.request)(), (target.response)(), std::any::TypeId::of::<()>())
296    }
297
298    pub fn accepts_native<D: OperationDeclaration, P: 'static>(&self) -> bool {
299        (self.native_types)() == (std::any::TypeId::of::<D::Request>(), std::any::TypeId::of::<D::Response>(), std::any::TypeId::of::<P>())
300    }
301
302    // Every Declared Resource Comes From The Snapshot Or The Carrier's Execution Context.
303    pub fn verify_resources(&self, provided: &[std::any::TypeId]) -> Result<(), ServingError> {
304        let missing =
305            self.resources.iter().find(|resource| !self.dependencies.contains(resource.id()) && !provided.contains(&resource.id()));
306        match missing {
307            Some(resource) => Err(ServingError::Resource {
308                operation: self.definition.id.to_string(),
309                resource: resource.name(),
310            }),
311            None => Ok(()),
312        }
313    }
314
315    pub fn validate(&self) -> Result<(), ServingError> {
316        if !self.builtin {
317            self.definition.validate().map_err(ServingError::from)?;
318        }
319        if self.definition != (self.declared)() {
320            return Err(ServingError::Catalog);
321        }
322        if !(self.endpoints_valid)() {
323            return Err(ServingError::EndpointTypes);
324        }
325        Ok(())
326    }
327
328    pub fn decoding_memory(&self, payload: &[u8]) -> Result<usize, crate::MessageError> {
329        (self.decoding_memory)(payload)
330    }
331
332    pub fn prepare_encoded(&self, payload: &[u8]) -> Result<PreparedNative, ServingError> {
333        (self.encoded)(self.dependencies.clone(), payload)
334    }
335
336    #[cfg(all(feature = "grpc", feature = "streaming"))]
337    pub fn prepare_grpc(&self, payload: &[u8], limits: super::WireStreamLimits) -> Result<PreparedNative, ServingError> {
338        (self.grpc)(self.dependencies.clone(), payload, limits)
339    }
340    #[cfg(all(feature = "grpc", feature = "streaming"))]
341    pub fn grpc_decoding_memory(&self, payload: &[u8]) -> Result<usize, crate::MessageError> {
342        (self.grpc_memory)(payload)
343    }
344
345    pub fn prepare_native(
346        &self, request: Box<dyn Any + Send>, endpoints: Box<dyn Any + Send>,
347    ) -> Result<Box<dyn NativeInstance>, ServingError> {
348        (self.native)(self.dependencies.clone(), request, endpoints)
349    }
350
351    pub fn contract(&self) -> Result<Contract, super::ContractError> {
352        self.contract_with_settings(&harmos_domain::config::SchemaSettings::default())
353    }
354
355    pub fn contract_with_settings(&self, settings: &harmos_domain::config::SchemaSettings) -> Result<Contract, super::ContractError> {
356        (self.contract)(settings)
357    }
358
359    pub fn install(self, table: &mut WireTable) -> Result<(), ServingError> {
360        self.validate()?;
361        self.verify_resources(table.provided())?;
362        let contract = self.contract()?;
363        (self.install)(table, self.dependencies, contract)
364    }
365}
366
367impl<O: Carrier> NativeInstance for NativeOperation<O> {
368    fn classify(&self, error: &(dyn std::error::Error + Send + Sync + 'static)) -> crate::operations::FailureKind {
369        error.downcast_ref::<O::Error>().map_or(crate::operations::FailureKind::Permanent, O::classify)
370    }
371
372    fn initialize(&mut self, context: ExecutionContext) -> ServiceFuture<'_, ()> {
373        Box::pin(async move {
374            if self.operation.is_none() {
375                let context = context.for_registration(self.dependencies.clone());
376                self.operation = Some(O::bind(context).await.map_err(operation_error::<O>)?);
377            }
378            Ok(())
379        })
380    }
381
382    fn run(&mut self, context: ExecutionContext) -> ServiceFuture<'_, super::response::ErasedResponse> {
383        Box::pin(async move {
384            let operation = self.operation.as_mut().ok_or(ServingError::Uninitialized)?;
385            let context = context.for_registration(self.dependencies.clone());
386            #[cfg(feature = "streaming")]
387            let attempt = if let Some(streams) = &self.streams {
388                self.epoch = self.epoch.checked_add(1).ok_or(ServingError::StreamPhase)?;
389                Some(streams.start(self.epoch)?)
390            } else {
391                None
392            };
393            let response = operation.attempt(context, &mut self.request, &mut self.endpoints).await.map_err(operation_error::<O>)?;
394            #[cfg(feature = "streaming")]
395            if let Some(attempt) = attempt {
396                attempt.end();
397            }
398            #[cfg(all(feature = "grpc", feature = "streaming"))]
399            if self.protobuf {
400                let mut bytes = Vec::new();
401                crate::grpc::encode(&*response, &mut bytes).map_err(ServingError::Codec)?;
402                return Ok(super::Response::new(bytes).into_erased());
403            }
404            if self.encoded {
405                return Ok(super::Response::new(response.encode()).into_erased());
406            }
407            Ok(response.into_erased())
408        })
409    }
410
411    fn stop(&mut self) -> ServiceFuture<'_, ()> {
412        Box::pin(async move {
413            if self.stopped {
414                return Ok(());
415            }
416            if let Some(operation) = self.operation.as_mut() {
417                operation.stop().await.map_err(operation_error::<O>)?;
418            }
419            self.stopped = true;
420            Ok(())
421        })
422    }
423
424    fn cleanup(&mut self) -> ServiceFuture<'_, ()> {
425        Box::pin(async move {
426            if let Some(operation) = self.operation.as_mut() {
427                operation.cleanup().await.map_err(operation_error::<O>)?;
428            }
429            self.operation = None;
430            Ok(())
431        })
432    }
433
434    fn transferred(&self) -> bool {
435        use super::EndpointSet;
436        self.endpoints.transferred() || self.operation.as_ref().is_some_and(Carrier::transferred)
437    }
438}
439
440impl ServiceFactory {
441    pub fn initialize(&self, context: ExecutionContext) -> ServiceFuture<'static, Box<dyn ServiceInstance>> {
442        (self.initialize)(context.for_registration(self.dependencies.clone()))
443    }
444}
445
446impl<S: Service> ServiceInstance for InitializedService<S> {
447    fn run(&mut self, context: ExecutionContext) -> ServiceFuture<'_, ()> {
448        Box::pin(async move { self.service.run(context.for_registration(self.dependencies.clone())).await.map_err(service_error) })
449    }
450    fn stop(&mut self) -> ServiceFuture<'_, ()> {
451        Box::pin(async move { self.service.stop().await.map_err(service_error) })
452    }
453    fn cleanup(&mut self) -> ServiceFuture<'_, ()> {
454        Box::pin(async move { self.service.cleanup().await.map_err(service_error) })
455    }
456}
457
458/* === Public Functions === */
459
460/* === Private Functions === */
461
462fn service_error(error: impl std::error::Error + 'static) -> ServingError {
463    ServingError::Handler(crate::Failure::originate(crate::operations::FailureKind::Permanent, &error, crate::failure::Origin::Plugin))
464}
465
466fn operation_error<O: Carrier>(error: O::Error) -> ServingError {
467    ServingError::Native(std::sync::Arc::new(error))
468}
469
470/* === Tests === */
471
472#[cfg(test)]
473mod tests {
474    use super::*;
475    use std::sync::{
476        Arc,
477        atomic::{AtomicUsize, Ordering},
478    };
479
480    struct Worker(Arc<AtomicUsize>);
481
482    #[crate::operation(harmos_path = crate, resources = [u64], id = "fixture.read", response = u64,
483        policy = crate::OperationPolicy {
484            timeout: std::time::Duration::from_secs(1),
485            attempt_timeout: std::time::Duration::from_secs(1),
486            retries: crate::RetryPolicy::Never,
487        },
488        key = |_: &Read| Some(super::super::CacheKey::new(vec![42]).unwrap()),
489    )]
490    struct Read;
491
492    impl crate::Operation for Read {
493        type Error = std::convert::Infallible;
494
495        async fn run(self, ctx: crate::Context<'_, Self>) -> Result<crate::Response<u64>, Self::Error> {
496            Ok(crate::Response::from_owner(ctx.resource::<u64>()))
497        }
498    }
499
500    fn ready<T>(mut future: ServiceFuture<'_, T>) -> T {
501        let mut task = std::task::Context::from_waker(std::task::Waker::noop());
502        match future.as_mut().poll(&mut task) {
503            std::task::Poll::Ready(Ok(value)) => value,
504            _ => panic!("synchronous fixture must complete successfully"),
505        }
506    }
507
508    #[test]
509    fn carrier_assignments_are_immutable() {
510        use crate::operations::{AssignedOperation, OperationAssignments};
511        let first = AssignedOperation {
512            operation: format!("fixture.{}", Read::ID),
513            fingerprint: Read::definition().fingerprint.0,
514            generation: 1,
515        };
516        let assignments = OperationAssignments::default();
517        let bytes = AssignedOperation::encode(std::slice::from_ref(&first)).unwrap();
518        assignments.install(&bytes).unwrap();
519        assignments.clone().install(&bytes).unwrap();
520        assert_eq!(assignments.resolve::<Read>().unwrap(), &first);
521        let mut replacement = first.clone();
522        replacement.generation = 2;
523        assert!(assignments.install(&AssignedOperation::encode(&[replacement]).unwrap()).is_err());
524        assert_eq!(assignments.resolve::<Read>().unwrap().generation, 1);
525        let mut wrong = first.clone();
526        wrong.fingerprint[0] ^= 1;
527        let incompatible = OperationAssignments::default();
528        incompatible.install(&AssignedOperation::encode(&[wrong]).unwrap()).unwrap();
529        assert!(incompatible.resolve::<Read>().is_err());
530        let duplicate = AssignedOperation {
531            operation: format!("other.{}", Read::ID),
532            ..first.clone()
533        };
534        assert!(OperationAssignments::default().install(&AssignedOperation::encode(&[first, duplicate]).unwrap()).is_err());
535    }
536
537    #[test]
538    fn native_response_retains_owner_across_retries_and_cleanup() {
539        let value = Arc::new(42u64);
540        let retained = Arc::downgrade(&value);
541        let mut dependencies = Dependencies::new();
542        dependencies.insert(value).unwrap();
543        let batch = RegistrationBatch::new(dependencies).operation::<Read>();
544        let factory = &batch.operations[0];
545        assert!(factory.prepare_native(Box::new(7u64), Box::new(())).is_err());
546        let mut invocation = factory.prepare_native(Box::new(Read), Box::new(())).unwrap();
547        let context = ExecutionContext::new(Dependencies::new(), super::super::CallContext::default());
548
549        ready(invocation.initialize(context.clone()));
550        let first = ready(invocation.run(context.clone())).restore::<u64>().unwrap();
551        let second = ready(invocation.run(context)).restore::<u64>().unwrap();
552        assert!(first.ptr_eq(&second));
553        assert_eq!(*first, 42);
554        ready(invocation.stop());
555        ready(invocation.cleanup());
556
557        drop(invocation);
558        drop(batch);
559        assert!(retained.upgrade().is_some());
560        drop(first);
561        drop(second);
562        assert!(retained.upgrade().is_none());
563    }
564
565    #[test]
566    fn encoded_factory_preserves_request_cache_key_and_response() {
567        use crate::Message;
568
569        let mut dependencies = Dependencies::new();
570        dependencies.insert(Arc::new(42u64)).unwrap();
571        let batch = RegistrationBatch::new(dependencies).operation::<Read>();
572        let mut prepared = batch.operations[0].prepare_encoded(&Read.encode()).unwrap();
573        assert_eq!(prepared.key, Read::key(&Read));
574        let context = ExecutionContext::new(Dependencies::new(), super::super::CallContext::default());
575        ready(prepared.instance.initialize(context.clone()));
576        let response = ready(prepared.instance.run(context)).restore::<Vec<u8>>().unwrap();
577        assert_eq!(u64::decode(&response).unwrap(), 42);
578        ready(prepared.instance.stop());
579        ready(prepared.instance.cleanup());
580    }
581
582    impl Service for Worker {
583        type Error = crate::ResourceError;
584
585        async fn initialize(context: ExecutionContext) -> Result<Self, Self::Error> {
586            let counter = context.resource::<AtomicUsize>()?;
587            counter.fetch_add(1, Ordering::SeqCst);
588            Ok(Self(counter))
589        }
590
591        async fn run(&mut self, _: ExecutionContext) -> Result<(), Self::Error> {
592            self.0.fetch_add(1, Ordering::SeqCst);
593            Ok(())
594        }
595    }
596
597    #[test]
598    fn factory_rejects_a_substituted_declaration() {
599        let mut batch = RegistrationBatch::new(Dependencies::new()).operation::<Read>();
600        batch.operations[0].definition.id = crate::Identifier::new("different").unwrap();
601        assert!(matches!(batch.operations[0].validate(), Err(ServingError::Catalog)));
602    }
603
604    #[test]
605    fn wire_identity_fences_replacement_before_decoding() {
606        use crate::operations::{ServingError, WireTable};
607        use crate::{Message, QualifiedId};
608        let mut dependencies = Dependencies::new();
609        dependencies.insert(Arc::new(42u64)).unwrap();
610        let batch = RegistrationBatch::new(dependencies).operation::<Read>();
611        let fingerprint = crate::OperationDefinition::of::<Read>().unwrap().fingerprint.0;
612        let mut table = WireTable::new("fixture").unwrap();
613        table.capacity(std::num::NonZeroUsize::MIN);
614        table.register(batch.clone()).unwrap();
615        assert!(matches!(table.prepare(1, "spoof.fixture.read", fingerprint, 1, Read.encode()), Err(ServingError::Unknown { .. })));
616        assert!(matches!(table.prepare(1, "fixture.fixture.read", [0; 32], 1, vec![255]), Err(ServingError::Identity { .. })));
617        assert!(matches!(table.prepare(1, "fixture.fixture.read", fingerprint, 0, vec![255]), Err(ServingError::Identity { .. })));
618        table.prepare(1, "fixture.fixture.read", fingerprint, 7, Read.encode()).unwrap();
619        ready(Box::pin(table.unregister(&QualifiedId::qualify("fixture", "fixture.read").unwrap())));
620        table.register(batch).unwrap();
621        assert!(matches!(table.prepare(2, "fixture.fixture.read", fingerprint, 7, vec![255]), Err(ServingError::Identity { .. })));
622        table.prepare(2, "fixture.fixture.read", fingerprint, 8, Read.encode()).unwrap();
623        assert!(matches!(table.prepare(3, "fixture.fixture.read", fingerprint, 9, vec![255]), Err(ServingError::Identity { .. })));
624        ready(Box::pin(table.settle()));
625    }
626
627    #[test]
628    fn service_factory_reconstructs_from_its_snapshot() {
629        let counter = Arc::new(AtomicUsize::new(0));
630        let mut dependencies = Dependencies::new();
631        dependencies.insert(counter.clone()).unwrap();
632        let policy = ServicePolicy {
633            stop_timeout: std::time::Duration::from_secs(1),
634            restart: crate::RestartPolicy::Never,
635        };
636        let batch = RegistrationBatch::new(dependencies).service::<Worker>("worker", policy);
637        let factory = &batch.services[0];
638        let context = ExecutionContext::new(Dependencies::new(), super::super::CallContext::default());
639        let mut first = factory.initialize(context.clone());
640        let mut second = factory.initialize(context);
641        let mut task = std::task::Context::from_waker(std::task::Waker::noop());
642        assert!(matches!(first.as_mut().poll(&mut task), std::task::Poll::Ready(Ok(_))));
643        assert!(matches!(second.as_mut().poll(&mut task), std::task::Poll::Ready(Ok(_))));
644        assert_eq!(counter.load(Ordering::SeqCst), 2);
645    }
646}
harmos_sdk/operations/registration.rs

harmos_sdk/operations/
registration.rs
