Mock Parallel Third Party Server Communication

[coosmic]

Citrus Version
4.5.2 ( can be reproduced with 4.6 as well)

Question
How can I mock parallel server communication with third party system?

Use Case:
Citrus client calls SUT which calls third party system. Now I want to do the same but in parallel. Two parallel client requests are sufficient. For client requests I know I need to use a selector in the response (example from docs works great without mocking the server communication using citrus_message_id), but how can I do the same for server request and response mocking?

What I've tried so far

1. Use parallel container with sequential block (closest to what I could find in the docs and what I actually like to do)

runner.run(parallel()
                       .actions(
                           sequential().actions(
                               createForkedRequest(httpClient, "ClientA"),
                               mockThirdPartyServerRequest(),mockThirdPartyServerResponse(),
                               createResponse(httpClient, "ClientA")),
                           sequential().actions(
                               createForkedRequest(httpClient, "ClientB"),
                               mockThirdPartyServerRequest(),mockThirdPartyServerResponse(),
                               createResponse(httpClient, "ClientB"))
                       ));

This will run into "Failed to find reply channel for message correlation key: citrus_message_id = 'ffce4ac3-4f4e-47a4-a755-541deea1cff0'" error.
2. Use parallel container only for client request and response mocking (separate parallel containers), while server communication is mocked in a sequential container in between (this works, but doesn't cover my goal)

runner.run(parallel()
                       .actions(
                           createForkedRequest(httpClient, "ClientA"),
                           createForkedRequest(httpClient, "ClientB")
                       ));

        runner.run(sequential().actions(mockThirdPartyServerRequest(),mockThirdPartyServerResponse(),
                                        mockThirdPartyServerRequest(),mockThirdPartyServerResponse()));

        runner.run(parallel()
                       .actions(
                           createResponse(httpClient, "ClientA"),
                           createResponse(httpClient, "ClientB")
                       ));

3. Use 3 parallel container one for client request one for server communication one for client response (just to be on the save side, that client and s)

runner.run(parallel()
                       .actions(
                           createForkedRequest(httpClient, "ClientA"),
                           createForkedRequest(httpClient, "ClientB")
                       ));

        runner.run(parallel().actions(mockThirdPartyServerRequest(),mockThirdPartyServerResponse(),
                                        mockThirdPartyServerRequest(),mockThirdPartyServerResponse()));

        runner.run(parallel()
                       .actions(
                           createResponse(httpClient, "ClientA"),
                           createResponse(httpClient, "ClientB")
                       ));

Additional information
My overarching goal is to manipulate the third party requests and test how my SUT behaves under different scenarios, so running the 3rd party requests in sequential order is not an option.
e.g: SUT creates an entity in DB and locks it when a specific endpoint is called. While SUT holds the lock it will do some 3rd party communication which I need to delay so that the second request can run into the lock error reliable.

I added an Example Project containing a server which does a simple request on the only endpoint and junit test suite to demonstrate what I tried.

[coosmic]

Short update: Adding some sleeps in front of the server communication mocking solved the issue, now I think this is a bug.

runner.run(parallel()
                       .actions(
                           sequential().actions(
                               createForkedRequest(httpClient, "ClientA"),
                               sleep().milliseconds(16L),
                               mockThirdPartyServerRequest(),mockThirdPartyServerResponse(),
                               createResponse(httpClient, "ClientA")),
                           sequential().actions(
                               sleep().milliseconds(2L),
                               createForkedRequest(httpClient, "ClientB"),
                               mockThirdPartyServerRequest(),mockThirdPartyServerResponse(),
                               createResponse(httpClient, "ClientB"))
                       ));

[christophd]

You may have a look into selective consuming on the server in order to selectively pick one of the available requests. Please see https://citrusframework.org/citrus/reference/html/index.html#message-queue-selector

As the server uses message channels under the hood you can use the selective consumer pattern directly with the server endpoint.

IMO your solution can look like this:

// non-blocking client request #1 and #2
runner.run(createForkedRequest(httpClient, "ClientA"));
runner.run(createForkedRequest(httpClient, "ClientB"));

// selective consuming of #1 request
runner.run(mockThirdPartyServerRequest("key1", "value1"))
// selective consuming of #2 request
runner.run(mockThirdPartyServerRequest("key2", "value2"))

...

So you can basically avoid the parallel and sequential blocks. The key1 and value1 select the request message based on headers or message body expressions (jsonPath, xpath).

This way you should be able to deterministically receive the requests that land in parallel on the server.

Please let me know if this helps

[coosmic]

Thank you for the fast reply!
I tried to apply a selector for my requests, but still I run into Failed to find reply channel for message correlation key: citrus_message_id ... error. I assume it is because the parallel requests are identical and can not be distinguished therefore.

Here is my changed reproducing setup:

SUT moved to POST and sending a body so that we can apply a body selector on it:

@RestController
public class Controller {

    @Autowired
    private RestTemplate restTemplate;

    @GetMapping("/hello")
    public String hello() {
        String response = restTemplate.postForObject(
            "http://localhost:9997/third/party/endpoint",
            "NoneDistinguishablePayload", String.class);
        return "Hello World";
    }
}

Test Setup remains as is, as "key1", "value1" etc. is not applicable for the use case:

void fullParallelShouldWork(@CitrusResource GherkinTestActionRunner runner) {
    runner.run(parallel()
                   .actions(
                       sequential().actions(
                           createForkedRequest(httpClient, "ClientA"),
                           mockThirdPartyServerRequest(),mockThirdPartyServerResponse(),
                           createResponse(httpClient, "ClientA")),
                       sequential().actions(
                           createForkedRequest(httpClient, "ClientB"),
                           mockThirdPartyServerRequest(),mockThirdPartyServerResponse(),
                           createResponse(httpClient, "ClientB"))
                   ));
}

mockThirdPartyServerRequest() implementation:

private HttpServerRequestActionBuilder mockThirdPartyServerRequest()
{
    return http().server(httpServer)
        .receive()
        .post("/third/party/endpoint")
        .selector(Collections.singletonMap("payload", "@startsWith(NoneDistinguishable)@")); 
}

Updated Gradle Project

Is there maybe some sort of selector to distinguish the requests made from SUT to the mocked 3rd party server, like it is done with the citrus_message_id in the forked requests from client to the SUT?
Maybe I also got you wrong with using a body selector and I can achieve parallel response serving by customizing MessageSelectingQueueChannel and apply it to the server somehow? In case how would I do this? There is no queue(...) or channel(...) method on HttpServer.
Assuming I convinced the server to use my custom MessageSelectingQueueChannel, what would I customize there? Would I override public Message<?> receive(MessageSelector selector) and always return a message if given?

Best Regards and thanks a lot for your time, not only for my question but on the framework in general. It is great!