@@ -19,7 +19,12 @@ pub async fn flow_execution_to_http_response(
1919 input : Value ,
2020 store : Arc < base:: store:: AdapterStore > ,
2121) -> Response < Full < Bytes > > {
22- match store. execute_flow_with_emitter ( flow, Some ( input) ) . await {
22+ let result = store. execute_flow_with_emitter ( flow, Some ( input) ) . await ;
23+ flow_execution_result_to_http_response ( result)
24+ }
25+
26+ fn flow_execution_result_to_http_response ( result : FlowExecutionResult ) -> Response < Full < Bytes > > {
27+ match result {
2328 FlowExecutionResult :: Ongoing ( result) => {
2429 log:: debug!( "Received first ongoing response from emitter" ) ;
2530 value_to_http_response ( result)
@@ -186,3 +191,87 @@ fn create_http_response(
186191
187192 builder. body ( Full :: new ( Bytes :: from ( body) ) ) . unwrap ( )
188193}
194+
195+ #[ cfg( test) ]
196+ mod tests {
197+ use super :: * ;
198+ use http_body_util:: BodyExt ;
199+ use tucana:: shared:: { NumberValue , number_value} ;
200+
201+ #[ tokio:: test]
202+ async fn response_signal_content_is_used ( ) {
203+ let response_value = Value {
204+ kind : Some ( Kind :: StructValue ( Struct {
205+ fields : [
206+ (
207+ "headers" . to_string ( ) ,
208+ Value {
209+ kind : Some ( Kind :: StructValue ( Struct {
210+ fields : [
211+ (
212+ "content-type" . to_string ( ) ,
213+ Value {
214+ kind : Some ( Kind :: StringValue ( "text/plain" . to_string ( ) ) ) ,
215+ } ,
216+ ) ,
217+ (
218+ "x-flow-response" . to_string ( ) ,
219+ Value {
220+ kind : Some ( Kind :: StringValue ( "matched" . to_string ( ) ) ) ,
221+ } ,
222+ ) ,
223+ ]
224+ . into_iter ( )
225+ . collect ( ) ,
226+ } ) ) ,
227+ } ,
228+ ) ,
229+ (
230+ "http_status_code" . to_string ( ) ,
231+ Value {
232+ kind : Some ( Kind :: NumberValue ( NumberValue {
233+ number : Some ( number_value:: Number :: Integer ( 201 ) ) ,
234+ } ) ) ,
235+ } ,
236+ ) ,
237+ (
238+ "payload" . to_string ( ) ,
239+ Value {
240+ kind : Some ( Kind :: StringValue ( "response body" . to_string ( ) ) ) ,
241+ } ,
242+ ) ,
243+ ]
244+ . into_iter ( )
245+ . collect ( ) ,
246+ } ) ) ,
247+ } ;
248+
249+ let response =
250+ flow_execution_result_to_http_response ( FlowExecutionResult :: Ongoing ( response_value) ) ;
251+
252+ assert_eq ! ( response. status( ) , StatusCode :: CREATED ) ;
253+ assert_eq ! ( response. headers( ) [ "content-type" ] , "text/plain" ) ;
254+ assert_eq ! ( response. headers( ) [ "x-flow-response" ] , "matched" ) ;
255+ assert_eq ! (
256+ response. into_body( ) . collect( ) . await . unwrap( ) . to_bytes( ) ,
257+ "response body"
258+ ) ;
259+ }
260+
261+ #[ tokio:: test]
262+ async fn finished_flow_without_response_returns_no_content ( ) {
263+ let response =
264+ flow_execution_result_to_http_response ( FlowExecutionResult :: FinishedWithoutOngoing ) ;
265+
266+ assert_eq ! ( response. status( ) , StatusCode :: NO_CONTENT ) ;
267+ assert ! (
268+ response
269+ . into_body( )
270+ . collect( )
271+ . await
272+ . unwrap( )
273+ . to_bytes( )
274+ . is_empty( )
275+ ) ;
276+ }
277+ }
0 commit comments