diff --git a/.github/workflows/build-client-dart.yaml b/.github/workflows/build-client-dart.yaml index 03d0de9b..760c83fe 100644 --- a/.github/workflows/build-client-dart.yaml +++ b/.github/workflows/build-client-dart.yaml @@ -17,6 +17,8 @@ jobs: runs-on: ubuntu-latest steps: - uses: actions/checkout@v3 + - name: 'Dependency Review' + uses: actions/dependency-review-action@v4 - uses: dart-lang/setup-dart@d6a63dab3335f427404425de0fbfed4686d93c4f - name: Setup Flutter uses: bancolombia/flutter-setup-action@v1.1 @@ -58,4 +60,4 @@ jobs: PANA=$(pana . --no-warning); PANA_SCORE=$(echo $PANA | sed -n "s/.*Points: \([0-9]*\)\/\([0-9]*\)./\1\/\2/p") echo "score: $PANA_SCORE" IFS='/'; read -a SCORE_ARR <<< "$PANA_SCORE"; SCORE=${SCORE_ARR[0]}; TOTAL=${SCORE_ARR[1]} - if (( $SCORE < $TOTAL - 30 )); then echo $PANA; echo "minimum score not met!"; exit 1; fi \ No newline at end of file + if (( $SCORE < $TOTAL - 30 )); then echo $PANA; echo "minimum score not met!"; exit 1; fi diff --git a/channel-sender/lib/channel_sender_ex/transport/rest/rest_controller.ex b/channel-sender/lib/channel_sender_ex/transport/rest/rest_controller.ex index f7d1d7ed..f1bb1766 100644 --- a/channel-sender/lib/channel_sender_ex/transport/rest/rest_controller.ex +++ b/channel-sender/lib/channel_sender_ex/transport/rest/rest_controller.ex @@ -35,6 +35,7 @@ defmodule ChannelSenderEx.Transport.Rest.RestController do post("/ext/channel/create", do: create_channel(conn)) post("/ext/channel/deliver_message", do: deliver_message(conn)) post("/ext/channel/deliver_batch", do: deliver_message(conn)) + post("/ext/channel/deliver_two_events", do: deliver_two_events(conn)) delete("/ext/channel", do: close_channel(conn)) match(_, do: send_resp(conn, 404, "Route not found.")) @@ -408,6 +409,96 @@ defmodule ChannelSenderEx.Transport.Rest.RestController do end end + defp deliver_two_events(conn) do + route_deliver_two_events(conn.body_params, conn) + end + + + defp route_deliver_two_events( + %{ + channel_ref: channel_ref, + event_one: event_one, + event_two: event_two + }, + conn + ) do + add_trace_metadata(%{channel_ref: channel_ref}) + + with {:ok, message_one} <- assert_deliver_request(event_one), + {:ok, message_two} <- assert_deliver_request(event_two) do + # Deliver first event + perform_delivery_two_events(channel_ref, message_one, message_two) + + conn + |> put_resp_header("content-type", "application/json") + |> send_resp(202, Jason.encode!(%{result: "Ok", events_sent: 2})) + else + {:error, :invalid_message} -> + invalid_body(conn) + end + end + + defp route_deliver_two_events(_, conn), do: invalid_body(conn) + + @spec perform_delivery_two_events(String.t(), map(), map()) :: :ok + defp perform_delivery_two_events(channel_ref, message_one, message_two) do + parent_ctx = Ctx.get_current() + + Task.start(fn -> + Ctx.attach(parent_ctx) + span_ctx = Tracer.start_span("deliver_two_events_to_channel", %{parent: parent_ctx}) + Tracer.set_current_span(span_ctx) + + # Convert to protocol messages + protocol_msg_one = ProtocolMessage.to_protocol_message(message_one) + protocol_msg_two = ProtocolMessage.to_protocol_message(message_two) + + # Deliver first event + res_one = PubSubCore.deliver_to_channel(channel_ref, protocol_msg_one) + + Logger.debug( + "Delivering first event to channel #{channel_ref}, result: #{inspect(res_one)}" + ) + + # Small delay to ensure order (optional, can be removed if not needed) + Process.sleep(10) + + # Deliver second event + res_two = PubSubCore.deliver_to_channel(channel_ref, protocol_msg_two) + + Logger.debug( + "Delivering second event to channel #{channel_ref}, result: #{inspect(res_two)}" + ) + + case {res_one, res_two} do + {:error, _} -> + Logger.warning( + "Channel #{inspect(channel_ref)} not found, first message delivery failed" + ) + + Tracer.set_status(OpenTelemetry.status(:error, "Channel not found for first event")) + + {_, :error} -> + Logger.warning( + "Channel #{inspect(channel_ref)} not found, second message delivery failed" + ) + + Tracer.set_status(OpenTelemetry.status(:error, "Channel not found for second event")) + + _ -> + Tracer.add_event("deliver_two_events", %{ + detail: "Both events delivered successfully" + }) + + :ok + end + + Tracer.end_span() + end) + + :ok + end + defp build_and_send_response({202, body}, conn) do conn |> put_resp_header("content-type", "application/json") diff --git a/clients/client-dart/example/lib/async_client_service.dart b/clients/client-dart/example/lib/async_client_service.dart index 9c6ed090..fd97d8b6 100644 --- a/clients/client-dart/example/lib/async_client_service.dart +++ b/clients/client-dart/example/lib/async_client_service.dart @@ -66,6 +66,16 @@ class AsyncClientService extends InheritedWidget { 'Message from async dataflow, title: ${msg.payload['data']['reply']['messageData']['title']} detail: ${msg.payload['data']['reply']['messageData']['detail']}'); } + if (msg.event == 'ch-ms-async-callback.svp.p2p') { + responsesNotifier.addResponse( + 'Message from async dataflow, title: ${msg.payload['title']}'); + } + + if (msg.event == 'ch-ms-async-callback.svp.p2m') { + responsesNotifier.addResponse( + 'Message from async dataflow, title: ${msg.payload['title']} '); + } + if (msg.event == ':n_token') { // The client app can also subscrtibe to this ADF internal event to listen // when connector receives a new token from the backend. diff --git a/clients/client-dart/example/lib/infrastructure/driven_adapter/api_service.dart b/clients/client-dart/example/lib/infrastructure/driven_adapter/api_service.dart index c8581671..837a2e6c 100644 --- a/clients/client-dart/example/lib/infrastructure/driven_adapter/api_service.dart +++ b/clients/client-dart/example/lib/infrastructure/driven_adapter/api_service.dart @@ -33,6 +33,6 @@ class ApiService implements AsyncClientGateway { Future callBusinessUseCase( String channelRef, String userRef, int delay) async { return http.get(Uri.parse( - "$urlBusinessService/business?channel_ref=$channelRef&user_ref=$userRef&delay=$delay")); + "$urlBusinessService/business/two-events?channel_ref=$channelRef&user_ref=$userRef&delay=$delay")); } } diff --git a/clients/client-dart/example/lib/my_app.dart b/clients/client-dart/example/lib/my_app.dart index 5389c27f..051f3c18 100644 --- a/clients/client-dart/example/lib/my_app.dart +++ b/clients/client-dart/example/lib/my_app.dart @@ -14,7 +14,9 @@ class MyApp extends StatelessWidget { eventListen: const [ 'ch-ms-async-callback.svp.reply', 'businessEvent', - ':n_token' + ':n_token', + 'ch-ms-async-callback.svp.p2p', + 'ch-ms-async-callback.svp.p2m', ], asyncClientGateway: ApiService(context), appConfig: AppConfig.of(context), diff --git a/clients/client-dart/example/lib/ui/pages/request_page.dart b/clients/client-dart/example/lib/ui/pages/request_page.dart index 099cdcdc..5061528d 100644 --- a/clients/client-dart/example/lib/ui/pages/request_page.dart +++ b/clients/client-dart/example/lib/ui/pages/request_page.dart @@ -136,20 +136,22 @@ class _RequestPageState extends State { onPressed: () => homeHelper.switchProtocols(), ), const SizedBox(width: 10), - OutlinedButton( - child: const Row( - children: [ - Icon( - Icons.stop, - ), - SizedBox(width: 5), - Text( - "Disconnect", - style: TextStyle(fontSize: 16), - ), - ], + Expanded( + child: OutlinedButton( + child: const Row( + children: [ + Icon( + Icons.stop, + ), + SizedBox(width: 5), + Text( + "Disconn", + style: TextStyle(fontSize: 16), + ), + ], + ), + onPressed: () => homeHelper.disconnect(), ), - onPressed: () => homeHelper.disconnect(), ), ], ), diff --git a/examples/back-async-java/domain/usecase/src/main/java/co/com/bancolombia/usecase/business/BusinessUseCase.java b/examples/back-async-java/domain/usecase/src/main/java/co/com/bancolombia/usecase/business/BusinessUseCase.java index ac04dfc3..e1c7e84f 100644 --- a/examples/back-async-java/domain/usecase/src/main/java/co/com/bancolombia/usecase/business/BusinessUseCase.java +++ b/examples/back-async-java/domain/usecase/src/main/java/co/com/bancolombia/usecase/business/BusinessUseCase.java @@ -10,12 +10,14 @@ import reactor.core.scheduler.Schedulers; import java.time.Duration; +import java.util.Random; import java.util.UUID; @Log @RequiredArgsConstructor public class BusinessUseCase { private final AsyncDataFlowGateway asyncDataFlowGateway; + private static final Random random = new Random(); public Mono generateCredentials(String userIdentifier) { return asyncDataFlowGateway.generateCredentials(userIdentifier); @@ -47,4 +49,65 @@ public Mono asyncBusinessFlow(String delay, String channelRef, String us .subscribe(); return Mono.empty(); } + + /** + * Delivers two different events to the same channel sequentially. + * This simulates a scenario where a user makes a request and receives two different async responses. + * + * @param delay Delay in milliseconds before sending the events + * @param channelRef Channel reference to deliver messages to + * @param userRef User reference + * @param correlationId Correlation ID for both events + * @return Empty Mono + */ + public Mono asyncBusinessFlowTwoEvents(String delay, String channelRef, String userRef, String correlationId) { + log.info("Delaying async flow with two events for channel: " + channelRef); + + Mono.delay(Duration.ofMillis(Integer.parseInt(delay))) + .then(Mono.defer(() -> { + log.info("Delivering first event to channel: " + channelRef); + + // First event - Process Started + DeliverMessage firstEvent = DeliverMessage.builder() + .messageId(UUID.randomUUID().toString()) + .CorrelationId(correlationId) + .messageData(Message.builder() + .code("100") + .title("ch-ms-async-callback.svp.p2p") + .detail("Your request is being processed - correlation: " + correlationId) + .severity("INFO") + .build()) + .channelRef(channelRef) + .eventName("ch-ms-async-callback.svp.p2p") + .build(); + + + + // Second event - Process Completed + DeliverMessage secondEvent = DeliverMessage.builder() + .messageId(UUID.randomUUID().toString()) + .CorrelationId(correlationId) + .messageData(Message.builder() + .code("200") + .title("ch-ms-async-callback.svp.p2m") + .detail("Your request has been successfully processed - correlation: " + correlationId) + .severity("SUCCESS") + .build()) + .channelRef(channelRef) + .eventName("ch-ms-async-callback.svp.p2m") + .build(); + + // Deliver first event, then second event after a small delay + return asyncDataFlowGateway.deliverMessage(channelRef, userRef, firstEvent) + .doOnSuccess(ignored -> log.info("First event delivered to channel: " + channelRef)) + .then(Mono.delay(Duration.ofMillis(random.nextInt(10001)))) // Random delay between 0-10 seconds + .then(asyncDataFlowGateway.deliverMessage(channelRef, userRef, secondEvent)) + .doOnSuccess(ignored -> log.info("Second event delivered to channel: " + channelRef)) + .doOnSuccess(ignored -> log.info("Both events successfully delivered to channel: " + channelRef)); + })) + .subscribeOn(Schedulers.boundedElastic()) + .subscribe(); + + return Mono.empty(); + } } diff --git a/examples/back-async-java/infrastructure/entry-points/reactive-web/src/main/java/co/com/bancolombia/api/Handler.java b/examples/back-async-java/infrastructure/entry-points/reactive-web/src/main/java/co/com/bancolombia/api/Handler.java index 42e6922e..b22a9d9e 100644 --- a/examples/back-async-java/infrastructure/entry-points/reactive-web/src/main/java/co/com/bancolombia/api/Handler.java +++ b/examples/back-async-java/infrastructure/entry-points/reactive-web/src/main/java/co/com/bancolombia/api/Handler.java @@ -35,6 +35,17 @@ public Mono listenGenerateCredentials(ServerRequest serverReques } + public Mono listenBusinessTwoEvents(ServerRequest serverRequest) { + return ServerResponse.accepted() + .body( + useCase.asyncBusinessFlowTwoEvents( + serverRequest.queryParam("delay").orElse("5000"), + serverRequest.queryParam("channel_ref").orElse(""), + serverRequest.queryParam("user_ref").toString(), + serverRequest.queryParam("correlationId").orElse(UUID.randomUUID().toString())), + String.class); + } + public static Mono responseHandler(T response, HttpStatus status) { return ServerResponse.status(status) .contentType(MediaType.APPLICATION_JSON) diff --git a/examples/back-async-java/infrastructure/entry-points/reactive-web/src/main/java/co/com/bancolombia/api/RouterRest.java b/examples/back-async-java/infrastructure/entry-points/reactive-web/src/main/java/co/com/bancolombia/api/RouterRest.java index 9c22292d..4fdbf836 100644 --- a/examples/back-async-java/infrastructure/entry-points/reactive-web/src/main/java/co/com/bancolombia/api/RouterRest.java +++ b/examples/back-async-java/infrastructure/entry-points/reactive-web/src/main/java/co/com/bancolombia/api/RouterRest.java @@ -13,6 +13,7 @@ public class RouterRest { @Bean public RouterFunction routerFunction(Handler handler) { return route(GET("/api/business"), handler::listenBusiness) + .andRoute(GET("/api/business/two-events"), handler::listenBusinessTwoEvents) .andRoute(GET("/api/credentials"), handler::listenGenerateCredentials); }