dynamo_runtime/
pipeline.rs

1// SPDX-FileCopyrightText: Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2// SPDX-License-Identifier: Apache-2.0
3//
4// Licensed under the Apache License, Version 2.0 (the "License");
5// you may not use this file except in compliance with the License.
6// You may obtain a copy of the License at
7//
8// http://www.apache.org/licenses/LICENSE-2.0
9//
10// Unless required by applicable law or agreed to in writing, software
11// distributed under the License is distributed on an "AS IS" BASIS,
12// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13// See the License for the specific language governing permissions and
14// limitations under the License.
15
16/// In a Pipeline, the [`AsyncEngine`] is constrained to take a [`Context`] as input and return
17/// a [`super::engine::ResponseStream`] as output.
18use serde::{Deserialize, Serialize};
19
20mod nodes;
21pub use nodes::{
22    Operator, PipelineNode, PipelineOperator, SegmentSink, SegmentSource, Service, ServiceBackend,
23    ServiceFrontend, Sink, Source,
24};
25
26pub mod context;
27pub mod error;
28pub mod network;
29pub use network::egress::addressed_router::{AddressedPushRouter, AddressedRequest};
30pub use network::egress::push_router::{PushRouter, RouterMode};
31pub mod registry;
32
33pub use crate::engine::{
34    self as engine, async_trait, AsyncEngine, AsyncEngineContext, AsyncEngineContextProvider, Data,
35    DataStream, Engine, EngineStream, EngineUnary, ResponseStream,
36};
37pub use anyhow::Error;
38pub use context::Context;
39pub use error::{PipelineError, PipelineErrorExt, TwoPartCodecError};
40
41/// Pipeline inputs carry a [`Context`] which can be used to carry metadata or additional information
42/// about the request. This information propagates through the stages, both local and distributed.
43pub type SingleIn<T> = Context<T>;
44
45/// Pipeline inputs carry a [`Context`] which can be used to carry metadata or additional information
46/// about the request. This information propagates through the stages, both local and distributed.
47pub type ManyIn<T> = Context<DataStream<T>>;
48
49/// Type alias for the output of pipeline that returns a single value
50pub type SingleOut<T> = EngineUnary<T>;
51
52/// Type alias for the output of pipeline that returns multiple values
53pub type ManyOut<T> = EngineStream<T>;
54
55pub type ServiceEngine<T, U> = Engine<T, U, Error>;
56
57/// Unary Engine is a pipeline that takes a single input and returns a single output
58pub type UnaryEngine<T, U> = ServiceEngine<SingleIn<T>, SingleOut<U>>;
59
60/// `ClientStreaming` Engine is a pipeline that takes multiple inputs and returns a single output
61/// Typically the engine will consume the entire input stream; however, it can also decided to exit
62/// early and emit a response without consuming the entire input stream.
63pub type ClientStreamingEngine<T, U> = ServiceEngine<ManyIn<T>, SingleOut<U>>;
64
65/// `ServerStreaming` takes a single input and returns multiple outputs.
66pub type ServerStreamingEngine<T, U> = ServiceEngine<SingleIn<T>, ManyOut<U>>;
67
68/// `BidirectionalStreaming` takes multiple inputs and returns multiple outputs. Input and output values
69/// are considered independent of each other; however, they could be constrained to be related.
70pub type BidirectionalStreamingEngine<T, U> = ServiceEngine<ManyIn<T>, ManyOut<U>>;
71
72pub trait AsyncTransportEngine<T: PipelineIO, U: PipelineIO>:
73    AsyncEngine<T, U, Error> + Send + Sync + 'static
74{
75}
76
77// pub type TransportEngine<T, U> = Arc<dyn AsyncTransportEngine<T, U>>;
78
79mod sealed {
80    use super::*;
81
82    #[allow(dead_code)]
83    pub struct Token;
84
85    pub trait Connectable {
86        type DataType: Data;
87    }
88
89    impl<T: Data> Connectable for Context<T> {
90        type DataType = T;
91    }
92    impl<T: Data> Connectable for EngineUnary<T> {
93        type DataType = T;
94    }
95    impl<T: Data> Connectable for EngineStream<T> {
96        type DataType = T;
97    }
98}
99
100pub trait PipelineIO: Data + sealed::Connectable + AsyncEngineContextProvider {
101    fn id(&self) -> String;
102}
103
104impl<T: Data> PipelineIO for Context<T> {
105    fn id(&self) -> String {
106        self.id().to_string()
107    }
108}
109impl<T: Data> PipelineIO for EngineUnary<T> {
110    fn id(&self) -> String {
111        self.context().id().to_string()
112    }
113}
114impl<T: Data> PipelineIO for EngineStream<T> {
115    fn id(&self) -> String {
116        self.context().id().to_string()
117    }
118}
119
120#[derive(Serialize, Deserialize, Debug, Clone)]
121pub struct Event {
122    pub id: String,
123}
dynamo_runtime/pipeline.rs

dynamo_runtime/
pipeline.rs
