AsyncAPI 3.x: the driver contract for publishing a message

client-java/controller-api/src/main/java/org/evomaster/client/java/controller/api/dto/ActionDto.java

@@ -1,6 +1,7 @@
 package org.evomaster.client.java.controller.api.dto;
 
 import org.evomaster.client.java.controller.api.dto.problem.ExternalServiceDto;
+import org.evomaster.client.java.controller.api.dto.problem.asyncapi.AsyncApiActionDto;
 import org.evomaster.client.java.controller.api.dto.problem.rpc.RPCActionDto;
 import org.evomaster.client.java.controller.api.dto.problem.rpc.ScheduleTaskInvocationDto;
 
@@ -39,6 +40,12 @@ public class ActionDto {
      * note that this is only used when handling RPC problem
      */
     public RPCActionDto rpcCall;
+
+    /**
+     * info to publish a message.
+     * note that this is only used when handling an AsyncAPI problem
+     */
+    public AsyncApiActionDto asyncApiCall;
 
 
     /**

client-java/controller-api/src/main/java/org/evomaster/client/java/controller/api/dto/SutInfoDto.java

@@ -6,6 +6,7 @@
 import org.evomaster.client.java.controller.api.dto.problem.RestProblemDto;
 import org.evomaster.client.java.controller.api.dto.problem.GraphQLProblemDto;
 import org.evomaster.client.java.controller.api.dto.problem.WebProblemDto;
+import org.evomaster.client.java.controller.api.dto.problem.AsyncApiProblemDto;
 
 import java.util.List;
 
@@ -35,6 +36,12 @@ public class SutInfoDto {
      */
     public WebProblemDto webProblem;
 
+    /**
+     * If the SUT is a service driven by messages and described by an AsyncAPI document,
+     * here there will be the info on how to interact with it
+     */
+    public AsyncApiProblemDto asyncApiProblem;
+
     /**
      * Whether the SUT is running or not
      */

client-java/controller-api/src/main/java/org/evomaster/client/java/controller/api/dto/problem/AsyncApiProblemDto.java

@@ -0,0 +1,26 @@
+package org.evomaster.client.java.controller.api.dto.problem;
+
+/**
+ * Info the driver gives about an AsyncAPI service, so that the core can learn what the service
+ * consumes and what shape those messages have.
+ *
+ * Note the transport client is not here, and never crosses: it is an open connection to a
+ * broker, held by the driver. Only the document travels.
+ */
+public class AsyncApiProblemDto extends ProblemInfoDto {
+
+    /**
+     * Where the AsyncAPI document can be fetched from: a URL, or a path on the machine running
+     * the driver. Null when the document is given inline instead.
+     */
+    public String schemaLocation;
+
+    /**
+     * The AsyncAPI document itself. Null when a location is given instead.
+     *
+     * Useful when the document is packaged with the service rather than served by it, which is
+     * the common case: unlike OpenAPI, an AsyncAPI service rarely exposes its own contract over
+     * HTTP, since it may not speak HTTP at all.
+     */
+    public String schemaText;
+}

client-java/controller-api/src/main/java/org/evomaster/client/java/controller/api/dto/problem/asyncapi/AsyncApiActionDto.java

@@ -0,0 +1,99 @@
+package org.evomaster.client.java.controller.api.dto.problem.asyncapi;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/**
+ * One message for the driver to publish, and what to do about a reply.
+ *
+ * Everything here is decided by the core: which operation, where it goes, what it says. The
+ * driver's job is to put it on the wire and, when a reply is expected, wait for the one that
+ * answers it.
+ */
+public class AsyncApiActionDto {
+
+    /**
+     * The two places a correlation id can travel, as {@link #correlationLocation} names them.
+     */
+    public static final String CORRELATION_IN_HEADER = "HEADER";
+
+    public static final String CORRELATION_IN_PAYLOAD = "PAYLOAD";
+
+    /**
+     * Key of the operation in the AsyncAPI document. Sent along so the driver can report and
+     * log in terms the user will recognise from their own contract.
+     */
+    public String operationId;
+
+    /**
+     * Key of the channel the message is published on.
+     */
+    public String channelName;
+
+    /**
+     * Where the message actually goes on the wire: a topic, a queue, a routing key. Already
+     * resolved by the core, including any binding that overrides the channel's address.
+     */
+    public String address;
+
+    /**
+     * Id of the message being published, as the document names it.
+     */
+    public String messageId;
+
+    /**
+     * The message body, serialised. Its content type is in {@link #contentType}.
+     */
+    public String payload;
+
+    /**
+     * What the document declares the payload is encoded as, eg "application/json".
+     */
+    public String contentType;
+
+    /**
+     * Headers to publish alongside the body, for a transport that has them.
+     * Key is the header name as the document declares it, value is what to send under it, as text.
+     */
+    public Map<String, String> headers = new LinkedHashMap<>();
+
+    /**
+     * The value stamped into this message so that a reply can be recognised as answering it.
+     *
+     * It is minted fresh by the core for every execution rather than being part of the message
+     * the search varies: pairing needs a value unique to the execution, and the service only
+     * echoes it back.
+     */
+    public String correlationId;
+
+    /**
+     * Where the correlation id has to be written, as the document declares it. One of
+     * {@link #CORRELATION_IN_HEADER} or {@link #CORRELATION_IN_PAYLOAD}, or null when the
+     * document says nothing, in which case it is up to the driver to decide -- a transport with
+     * native correlation should use it.
+     */
+    public String correlationLocation;
+
+    /**
+     * JSON Pointer to the field the correlation id goes in, within whatever
+     * {@link #correlationLocation} names. Null when there is no declared location.
+     */
+    public String correlationPointer;
+
+    /**
+     * Where a reply is expected to arrive, when the operation declares one. Null for a
+     * fire-and-forget operation, in which case the driver publishes and returns.
+     */
+    public String replyAddress;
+
+    /**
+     * How long to wait for a reply before giving up, in milliseconds.
+     *
+     * There is no right answer here: a slow service and a stuck one look the same from outside,
+     * so this is a tuning parameter with no equivalent in a synchronous protocol. It is set
+     * generously and reported with the result.
+     *
+     * Null when no reply is expected, as there is then nothing to wait for.
+     */
+    public Long replyTimeoutMs;
+}

client-java/controller-api/src/main/java/org/evomaster/client/java/controller/api/dto/problem/asyncapi/AsyncApiReplyDto.java

@@ -0,0 +1,89 @@
+package org.evomaster.client.java.controller.api.dto.problem.asyncapi;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/**
+ * What came of publishing one message.
+ *
+ * The four outcomes are deliberately distinguished, because they mean different things and only
+ * one of them is a fault:
+ *
+ * <ul>
+ *   <li>published, with no reply expected -- a fire-and-forget operation did what it could;</li>
+ *   <li>published, and a reply arrived -- the only case with something to classify;</li>
+ *   <li>published, and nothing arrived within the window -- the contract promised a reply and
+ *       did not deliver one, though a slow service and a stuck one look alike from outside;</li>
+ *   <li>could not be published at all -- a broken setup rather than a finding about the
+ *       service, which is why it is reported separately from silence.</li>
+ * </ul>
+ */
+public class AsyncApiReplyDto {
+
+    /**
+     * The index of the action this answers, echoing what was asked.
+     */
+    public Integer index;
+
+    /**
+     * Whether the message reached the broker. False means the driver could not publish, and
+     * {@link #errorMessage} says why.
+     *
+     * Null means the driver did not say, which is read the same way as false: without an answer
+     * here there is no knowing whether the message went out, and the rest of the test would
+     * mean nothing.
+     */
+    public Boolean published;
+
+    /**
+     * Whether a reply arrived and was recognised as answering this message.
+     *
+     * Null means the driver did not say, and is read as no reply having arrived.
+     */
+    public Boolean replyReceived;
+
+    /**
+     * Whether the driver waited for a reply at all. False for a fire-and-forget operation, so
+     * that the absence of a reply is not mistaken for silence in answer to a promise.
+     *
+     * Null means the driver did not say, and is read as not having waited.
+     */
+    public Boolean replyExpected;
+
+    /**
+     * The reply body, as it arrived.
+     */
+    public String replyPayload;
+
+    /**
+     * The reply's headers, for a transport that has them.
+     * Key is the header name, value is what arrived under it, as text.
+     */
+    public Map<String, String> replyHeaders = new LinkedHashMap<>();
+
+    /**
+     * Whether the reply carried back the correlation id that was stamped on the request.
+     *
+     * This is the honest answer to "did correlation work", which cannot be read off a contract:
+     * echoing the id is the service's own behaviour. A reply that arrives without it is
+     * recorded rather than treated as a fault, since from outside there is no telling a defect
+     * from a service that correlates by some business key instead.
+     *
+     * Null means the driver does not track correlation at all, which is not the same as having
+     * checked and found the id missing, and is recorded as neither.
+     */
+    public Boolean correlationMatched;
+
+    /**
+     * How long the driver waited, in milliseconds, whether or not anything arrived. Reported
+     * because the verdict on silence is only meaningful alongside how long it was waited for.
+     *
+     * Null when the driver did not wait at all: no reply was expected, or nothing was published.
+     */
+    public Long waitedMs;
+
+    /**
+     * Why publishing failed, when it did.
+     */
+    public String errorMessage;
+}

client-java/controller/src/main/java/org/evomaster/client/java/controller/internal/EMController.java

@@ -5,6 +5,7 @@
 import org.evomaster.client.java.controller.api.dto.*;
 import org.evomaster.client.java.controller.api.dto.database.operations.*;
 import org.evomaster.client.java.controller.api.dto.problem.*;
+import org.evomaster.client.java.controller.api.dto.problem.asyncapi.AsyncApiReplyDto;
 import org.evomaster.client.java.controller.api.dto.problem.param.DeriveParamResponseDto;
 import org.evomaster.client.java.controller.api.dto.problem.param.DerivedParamChangeReqDto;
 import org.evomaster.client.java.controller.api.dto.problem.param.RestDerivedParamDto;
@@ -235,6 +236,13 @@ public Response getSutInfo(@Context HttpServletRequest httpServletRequest) {
                 SimpleLogger.error(msg, e);
                 return Response.status(500).entity(WrappedResponseDto.withError(msg)).build();
             }
+        } else if (info instanceof AsyncApiProblem) {
+            AsyncApiProblem p = (AsyncApiProblem) info;
+            dto.asyncApiProblem = new AsyncApiProblemDto();
+            dto.asyncApiProblem.schemaLocation = p.getSchemaLocation();
+            dto.asyncApiProblem.schemaText = p.getSchemaText();
+            dto.asyncApiProblem.servicesToNotMock = servicesToNotMock;
+
         } else if(info instanceof WebProblem){
             WebProblem p = (WebProblem) info;
             dto.webProblem = new WebProblemDto();
@@ -830,6 +838,28 @@ public Response newAction(
                 }
 
             }
+
+            if (dto.asyncApiCall != null) {
+
+                AsyncApiReplyDto replyDto = new AsyncApiReplyDto();
+                replyDto.index = index;
+
+                try {
+                    sutController.executeAsyncApiAction(dto.asyncApiCall, replyDto);
+                    return Response.status(200).entity(WrappedResponseDto.withData(replyDto)).build();
+                } catch (Exception e) {
+                    /*
+                        Failing to publish is not a finding about the service, it is a broken
+                        setup, so it is reported as such rather than as silence in answer to a
+                        promised reply.
+                     */
+                    String msg = "Thrown exception when publishing a message: " + e.getMessage();
+                    SimpleLogger.error(msg, e);
+                    replyDto.published = false;
+                    replyDto.errorMessage = msg;
+                    return Response.status(500).entity(WrappedResponseDto.withData(replyDto)).build();
+                }
+            }
         }
 
         return Response.status(204).entity(WrappedResponseDto.withNoData()).build();

client-java/controller/src/main/java/org/evomaster/client/java/controller/internal/SutController.java

@@ -23,6 +23,8 @@
 import org.evomaster.client.java.controller.api.dto.MockDatabaseDto;
 import org.evomaster.client.java.controller.api.dto.database.schema.TableIdDto;
 import org.evomaster.client.java.controller.api.dto.problem.RPCProblemDto;
+import org.evomaster.client.java.controller.api.dto.problem.asyncapi.AsyncApiActionDto;
+import org.evomaster.client.java.controller.api.dto.problem.asyncapi.AsyncApiReplyDto;
 import org.evomaster.client.java.controller.api.dto.problem.rpc.*;
 import org.evomaster.client.java.controller.api.dto.problem.rpc.RPCTestDto;
 import org.evomaster.client.java.controller.internal.db.OpenSearchHandler;
@@ -1929,6 +1931,41 @@ public boolean isScheduleTaskCompleted(ScheduleTaskInvocationResultDto invocatio
         return false;
     }
 
+    /**
+     * Publish one message, and wait for the reply that answers it when one is expected.
+     *
+     * This is where a driver for an AsyncAPI service does its work, and it is the counterpart
+     * of {@link #executeAction(RPCActionDto, ActionResponseDto)} for RPC: the core decides
+     * what to send and reads what comes back, while everything that knows about a broker lives
+     * on this side. Only publish and await are protocol-specific, and they never leave here.
+     *
+     * A driver that does not test an AsyncAPI service has no reason to override this.
+     *
+     * Sketch of what an implementation does, for a transport whose correlation rides in
+     * metadata:
+     *
+     * <pre>
+     * publish(dto.address, dto.payload, dto.headers + {correlationId: dto.correlationId});
+     * reply.published = true;
+     * if (dto.replyAddress != null) {
+     *     reply.replyExpected = true;
+     *     awaitOn(dto.replyAddress, matching dto.correlationId, within dto.replyTimeoutMs);
+     * }
+     * </pre>
+     *
+     * Note what is not asked of the driver: it does not judge the reply, only reports it.
+     * Deciding what an outcome means is the core's job, so that it means the same thing
+     * whatever the transport.
+     *
+     * @param dto   what to publish, and where a reply is expected
+     * @param reply to be filled in with what happened
+     */
+    public void executeAsyncApiAction(AsyncApiActionDto dto, AsyncApiReplyDto reply) {
+        throw new IllegalStateException(
+                "Trying to publish a message, but this driver does not implement" +
+                        " executeAsyncApiAction. It must be overridden to test an AsyncAPI service.");
+    }
+
     @Override
     public void resetDatabase(List<String> tablesToClean) {