Test double response

.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
+          if (( $SCORE < $TOTAL - 30 )); then echo $PANA; echo "minimum score not met!"; exit 1; fi

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")

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.

clients/client-dart/example/lib/infrastructure/driven_adapter/api_service.dart

@@ -33,6 +33,6 @@ class ApiService implements AsyncClientGateway {
   Future<http.Response> 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"));
   }
 }

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),

clients/client-dart/example/lib/ui/pages/request_page.dart

@@ -136,20 +136,22 @@ class _RequestPageState extends State<RequestPage> {
                         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(),
                       ),
                     ],
                   ),

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<Credentials> generateCredentials(String userIdentifier) {
         return asyncDataFlowGateway.generateCredentials(userIdentifier);
@@ -47,4 +49,65 @@ public Mono<Object> 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<Object> 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();
+    }
 }

examples/back-async-java/infrastructure/entry-points/reactive-web/src/main/java/co/com/bancolombia/api/Handler.java

@@ -35,6 +35,17 @@ public Mono<ServerResponse> listenGenerateCredentials(ServerRequest serverReques
 
     }
 
+    public Mono<ServerResponse> 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 <T> Mono<ServerResponse> responseHandler(T response, HttpStatus status) {
         return ServerResponse.status(status)
                 .contentType(MediaType.APPLICATION_JSON)

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<ServerResponse> routerFunction(Handler handler) {
         return route(GET("/api/business"), handler::listenBusiness)
+                .andRoute(GET("/api/business/two-events"), handler::listenBusinessTwoEvents)
                 .andRoute(GET("/api/credentials"), handler::listenGenerateCredentials);
 
     }