Struct SegmentProcessor 

pub struct SegmentProcessor { /* private fields */ }

Runs an OutcomeSegment as a Tower Service<Exchange>.

Bridges segment-mode pipelines (which produce PipelineOutcome) into the Tower-result world so they can be wrapped by ProcessorLayer implementations. Completed and Failed map straight through; Stopped(ex) maps to Ok(ex) after UNCONDITIONALLY setting the CamelStop property to true, so run_steps re-derives the stop at the next Process step. The unconditional overwrite is deliberate: an exchange carrying CamelStop = false from an earlier step must not defeat the re-derivation.

Implementations

impl SegmentProcessor

pub fn new(inner: OutcomeSegment) -> Self

Wrap inner as a Tower service.

Example

A Stopped outcome leaves the wrapper as Ok(ex) with CamelStop forced to true, even though the incoming exchange carried CamelStop = false:

use std::future::Future;
use std::pin::Pin;

use camel_api::{
    Exchange, Message, OutcomePipeline, OutcomeSegment, PipelineOutcome, SegmentProcessor,
    Value, is_camel_stop,
};
use tower::ServiceExt;

/// Always reports `Stopped`, leaving the exchange untouched.
#[derive(Clone)]
struct StoppedPipeline;

impl OutcomePipeline for StoppedPipeline {
    fn clone_box(&self) -> Box<dyn OutcomePipeline> {
        Box::new(self.clone())
    }

    fn run<'a>(
        &'a mut self,
        exchange: Exchange,
    ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
        Box::pin(async move { PipelineOutcome::Stopped(exchange) })
    }
}

let mut exchange = Exchange::new(Message::new("x"));
exchange.set_property(camel_api::CAMEL_STOP, Value::Bool(false));

let processor = SegmentProcessor::new(OutcomeSegment::new(Box::new(StoppedPipeline)));
let response = ServiceExt::oneshot(processor, exchange)
    .await
    .expect("a stopped outcome is not an error");

assert!(
    is_camel_stop(&response),
    "the wrapper must force CamelStop=true"
);

Trait Implementations

impl Clone for SegmentProcessor

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Service<Exchange> for SegmentProcessor

type Response = Exchange

Responses given by the service.

type Error = CamelError

Errors produced by the service.

type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>

The future response value.

fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>

Returns Poll::Ready(Ok(())) when the service is able to process requests.

fn call(&mut self, exchange: Exchange) -> Self::Future

Process the request and return the response asynchronously.