fix(#1470): Add message store endpoint

connectors/citrus-kubernetes/src/test/java/org/citrusframework/kubernetes/endpoint/KubernetesEndpointComponentTest.java

@@ -22,6 +22,7 @@
 import org.citrusframework.endpoint.Endpoint;
 import org.citrusframework.endpoint.EndpointComponent;
 import org.citrusframework.endpoint.direct.DirectEndpointComponent;
+import org.citrusframework.endpoint.context.MessageStoreEndpointComponent;
 import org.citrusframework.http.client.HttpEndpointComponent;
 import org.citrusframework.http.client.HttpsEndpointComponent;
 import org.citrusframework.kubernetes.client.KubernetesClient;
@@ -69,9 +70,11 @@ public void testCreateClientEndpointWithParameters() throws Exception {
     @Test
     public void testLookupAll() {
         Map<String, EndpointComponent> validators = EndpointComponent.lookup();
-        Assert.assertEquals(validators.size(), 4L);
+        Assert.assertEquals(validators.size(), 5L);
         Assert.assertNotNull(validators.get("direct"));
         Assert.assertEquals(validators.get("direct").getClass(), DirectEndpointComponent.class);
+        Assert.assertNotNull(validators.get("message-store"));
+        Assert.assertEquals(validators.get("message-store").getClass(), MessageStoreEndpointComponent.class);
         Assert.assertNotNull(validators.get("http"));
         Assert.assertEquals(validators.get("http").getClass(), HttpEndpointComponent.class);
         Assert.assertNotNull(validators.get("https"));

connectors/citrus-selenium/src/test/java/org/citrusframework/selenium/endpoint/SeleniumEndpointComponentTest.java

@@ -22,6 +22,7 @@
 import org.citrusframework.endpoint.Endpoint;
 import org.citrusframework.endpoint.EndpointComponent;
 import org.citrusframework.endpoint.direct.DirectEndpointComponent;
+import org.citrusframework.endpoint.context.MessageStoreEndpointComponent;
 import org.citrusframework.http.client.HttpEndpointComponent;
 import org.citrusframework.http.client.HttpsEndpointComponent;
 import org.citrusframework.spi.ReferenceResolver;
@@ -87,9 +88,11 @@ public void testCreateBrowserEndpointWithParameters() throws Exception {
     @Test
     public void testLookupAll() {
         Map<String, EndpointComponent> validators = EndpointComponent.lookup();
-        Assert.assertEquals(validators.size(), 4L);
+        Assert.assertEquals(validators.size(), 5L);
         Assert.assertNotNull(validators.get("direct"));
         Assert.assertEquals(validators.get("direct").getClass(), DirectEndpointComponent.class);
+        Assert.assertNotNull(validators.get("message-store"));
+        Assert.assertEquals(validators.get("message-store").getClass(), MessageStoreEndpointComponent.class);
         Assert.assertNotNull(validators.get("http"));
         Assert.assertEquals(validators.get("http").getClass(), HttpEndpointComponent.class);
         Assert.assertNotNull(validators.get("https"));

core/citrus-api/src/main/java/org/citrusframework/endpoint/EndpointComponent.java

@@ -38,6 +38,7 @@ public interface EndpointComponent {
     Logger logger = LoggerFactory.getLogger(EndpointComponent.class);
 
     String ENDPOINT_NAME = "endpointName";
+    String ENDPOINT_CACHE = "endpointCache";
     String AUTO_CLOSE = "autoClose";
     String AUTO_REMOVE = "autoRemove";
 
@@ -49,29 +50,29 @@ public interface EndpointComponent {
 
     /**
      * Creates proper endpoint instance from endpoint uri.
-     * @param endpointUri
-     * @param context
-     * @return
      */
     Endpoint createEndpoint(String endpointUri, TestContext context);
 
     /**
      * Gets the name of this endpoint component.
-     * @return
      */
     String getName();
 
     /**
      * Construct endpoint name from endpoint uri.
-     * @param endpointUri
-     * @return
      */
     Map<String, String> getParameters(String endpointUri);
 
+    /**
+     * Returns setting defining whether this endpoint component allows endpoint caching.
+     */
+    default boolean supportsEndpointCaching() {
+        return true;
+    }
+
     /**
      * Resolves all available endpoint components from resource path lookup. Scans classpath for endpoint component meta information
      * and instantiates those components.
-     * @return
      */
     static Map<String, EndpointComponent> lookup() {
         Map<String, EndpointComponent> components = TYPE_RESOLVER.resolveAll();
@@ -86,8 +87,6 @@ static Map<String, EndpointComponent> lookup() {
      * Resolves endpoint component from resource path lookup with given resource name. Scans classpath for endpoint component meta information
      * with given name and returns instance of the component. Returns optional instead of throwing exception when no endpoint component
      * could be found.
-     * @param component
-     * @return
      */
     static Optional<EndpointComponent> lookup(String component) {
         try {

core/citrus-base/src/main/java/org/citrusframework/endpoint/AbstractEndpoint.java

@@ -35,7 +35,6 @@ public abstract class AbstractEndpoint implements Endpoint {
 
     /**
      * Default constructor using endpoint configuration.
-     * @param endpointConfiguration
      */
     public AbstractEndpoint(EndpointConfiguration endpointConfiguration) {
         this.endpointConfiguration = endpointConfiguration;
@@ -48,15 +47,13 @@ public EndpointConfiguration getEndpointConfiguration() {
 
     /**
      * Gets the endpoints consumer name.
-     * @return
      */
     public String getConsumerName() {
         return name + ":consumer";
     }
 
     /**
      * Gets the endpoints producer name.
-     * @return
      */
     public String getProducerName() {
         return name + ":producer";

core/citrus-base/src/main/java/org/citrusframework/endpoint/AbstractEndpointComponent.java

@@ -108,7 +108,7 @@ public Endpoint createEndpoint(String endpointUri, TestContext context) {
      * Removes other settings from given parameter map and returns just the endpoint parameters.
      */
     private Map<String, String> getEndpointParameters(Map<String, String> parameters) {
-        Set<String> internalSettings = Set.of(ENDPOINT_NAME, AUTO_CLOSE, AUTO_REMOVE);
+        Set<String> internalSettings = Set.of(ENDPOINT_NAME, ENDPOINT_CACHE, AUTO_CLOSE, AUTO_REMOVE);
         HashMap<String, String> result = new HashMap<>();
 
         for (String key : parameters.keySet()) {

core/citrus-base/src/main/java/org/citrusframework/endpoint/DefaultEndpointFactory.java

@@ -56,60 +56,42 @@ public class DefaultEndpointFactory implements EndpointFactory {
     @Override
     public Endpoint create(String endpointName, Annotation endpointConfig, TestContext context) {
         String qualifier = endpointConfig.annotationType().getAnnotation(CitrusEndpointConfig.class).qualifier();
-        Optional<AnnotationConfigParser> parser = Optional.ofNullable(context.getReferenceResolver().resolveAll(AnnotationConfigParser.class).get(qualifier));
+        AnnotationConfigParser parser = Optional.ofNullable(context.getReferenceResolver().resolveAll(AnnotationConfigParser.class).get(qualifier))
+                .or(() -> AnnotationConfigParser.lookup(qualifier))
+                .orElseThrow(() -> new CitrusRuntimeException(String.format("Unable to create endpoint annotation parser with name '%s'", qualifier)));
 
-        if (!parser.isPresent()) {
-            // try to get parser from default Citrus modules
-            parser = AnnotationConfigParser.lookup(qualifier);
-        }
-
-        if (parser.isPresent()) {
-            Endpoint endpoint = parser.get().parse(endpointConfig, context.getReferenceResolver());
-            endpoint.setName(endpointName);
-
-            if (endpoint instanceof ReferenceResolverAware referenceResolverAware) {
-                referenceResolverAware.setReferenceResolver(context.getReferenceResolver());
-            }
+        Endpoint endpoint = parser.parse(endpointConfig, context.getReferenceResolver());
+        endpoint.setName(endpointName);
 
-            if (endpoint instanceof InitializingPhase initializingBean) {
-                initializingBean.initialize();
-            }
+        if (endpoint instanceof ReferenceResolverAware referenceResolverAware) {
+            referenceResolverAware.setReferenceResolver(context.getReferenceResolver());
+        }
 
-            PropertyUtils.configure(endpointName, endpoint, context.getReferenceResolver());
-            return endpoint;
+        if (endpoint instanceof InitializingPhase initializingBean) {
+            initializingBean.initialize();
         }
 
-        throw new CitrusRuntimeException(String.format("Unable to create endpoint annotation parser with name '%s'", qualifier));
+        PropertyUtils.configure(endpointName, endpoint, context.getReferenceResolver());
+        return endpoint;
     }
 
     @Override
     public Endpoint create(String endpointName, CitrusEndpoint endpointConfig, Class<?> endpointType, TestContext context) {
-        Optional<EndpointBuilder> builder = context.getReferenceResolver().resolveAll(EndpointBuilder.class)
-                .values()
-                .stream()
-                .filter(endpointBuilder -> endpointBuilder.supports(endpointType))
-                .findFirst();
-
-        if (builder.isPresent()) {
-            Endpoint endpoint = builder.get().build(endpointConfig, context.getReferenceResolver());
-            endpoint.setName(endpointName);
-            return endpoint;
-        }
-
-        // try to get builder from default Citrus modules
-        Optional<EndpointBuilder<?>> lookup = EndpointBuilder.lookup()
+        EndpointBuilder<?> builder = context.getReferenceResolver().resolveAll(EndpointBuilder.class)
                 .values()
                 .stream()
                 .filter(endpointBuilder -> endpointBuilder.supports(endpointType))
-                .findFirst();
-
-        if (lookup.isPresent()) {
-            Endpoint endpoint = lookup.get().build(endpointConfig, context.getReferenceResolver());
-            endpoint.setName(endpointName);
-            return endpoint;
-        }
-
-        throw new CitrusRuntimeException(String.format("Unable to create endpoint builder for type '%s'", endpointType.getName()));
+                .findFirst()
+                .or(() -> EndpointBuilder.lookup()
+                        .values()
+                        .stream()
+                        .filter(endpointBuilder -> endpointBuilder.supports(endpointType))
+                        .findFirst())
+                .orElseThrow(() -> new CitrusRuntimeException(String.format("Unable to create endpoint builder for type '%s'", endpointType.getName())));
+
+        Endpoint endpoint = builder.build(endpointConfig, context.getReferenceResolver());
+        endpoint.setName(endpointName);
+        return endpoint;
     }
 
     @Override
@@ -131,18 +113,11 @@ public Endpoint create(String uri, TestContext context) {
         }
 
         String componentName = getComponentName(endpointUri);
-        Optional<EndpointComponent> component = Optional.ofNullable(getEndpointComponents(context.getReferenceResolver()).get(componentName));
+        EndpointComponent component = Optional.ofNullable(getEndpointComponents(context.getReferenceResolver()).get(componentName))
+                .or(() -> EndpointComponent.lookup(componentName))
+                .orElseThrow(() -> new CitrusRuntimeException(String.format("Unable to create endpoint component with name '%s'", componentName)));
 
-        if (component.isEmpty()) {
-            // try to get component from default Citrus modules
-            component = EndpointComponent.lookup(componentName);
-        }
-
-        if (component.isEmpty()) {
-            throw new CitrusRuntimeException(String.format("Unable to create endpoint component with name '%s'", componentName));
-        }
-
-        Map<String, String> parameters = component.get().getParameters(endpointUri);
+        Map<String, String> parameters = component.getParameters(endpointUri);
         String cachedEndpointName = parameters.getOrDefault(EndpointComponent.ENDPOINT_NAME, endpointUri);
 
         synchronized (endpointCache) {
@@ -152,23 +127,30 @@ public Endpoint create(String uri, TestContext context) {
                 }
                 return endpointCache.get(cachedEndpointName);
             } else {
-                Endpoint endpoint = component.get().createEndpoint(endpointUri, context);
-                endpointCache.put(cachedEndpointName, endpoint);
-
-                boolean autoRemove = Optional.ofNullable(parameters.get(EndpointComponent.AUTO_REMOVE))
-                                                    .map(Boolean::parseBoolean)
-                                                    .orElseGet(CitrusSettings::isAutoRemoveDynamicEndpoints);
-                if (autoRemove) {
-                    context.doFinally(() -> ctx -> {
-                        logger.info("Stopping and removing endpoint '{}' due to auto remove setting", endpoint.getName());
-                        if (endpoint instanceof ShutdownPhase destroyable) {
-                            destroyable.destroy();
-                        }
-
-                        synchronized (endpointCache) {
-                            endpointCache.remove(cachedEndpointName);
-                        }
-                    });
+                Endpoint endpoint = component.createEndpoint(endpointUri, context);
+
+                boolean endpointCacheEnabled = Optional.ofNullable(parameters.get(EndpointComponent.ENDPOINT_CACHE))
+                        .map(Boolean::parseBoolean)
+                        .orElse(component.supportsEndpointCaching());
+
+                if (endpointCacheEnabled) {
+                    endpointCache.put(cachedEndpointName, endpoint);
+
+                    boolean autoRemove = Optional.ofNullable(parameters.get(EndpointComponent.AUTO_REMOVE))
+                            .map(Boolean::parseBoolean)
+                            .orElseGet(CitrusSettings::isAutoRemoveDynamicEndpoints);
+                    if (autoRemove) {
+                        context.doFinally(() -> ctx -> {
+                            logger.info("Stopping and removing endpoint '{}' due to auto remove setting", endpoint.getName());
+                            if (endpoint instanceof ShutdownPhase destroyable) {
+                                destroyable.destroy();
+                            }
+
+                            synchronized (endpointCache) {
+                                endpointCache.remove(cachedEndpointName);
+                            }
+                        });
+                    }
                 }
                 return endpoint;
             }

core/citrus-base/src/main/java/org/citrusframework/endpoint/StaticEndpoint.java

@@ -17,16 +17,22 @@
 package org.citrusframework.endpoint;
 
 import org.citrusframework.context.TestContext;
+import org.citrusframework.exceptions.MessageTimeoutException;
 import org.citrusframework.message.DefaultMessage;
 import org.citrusframework.message.Message;
 import org.citrusframework.messaging.Consumer;
 import org.citrusframework.messaging.Producer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Special endpoint implementation that produces/consumes static messages.
  */
 public class StaticEndpoint extends AbstractEndpoint {
 
+    /** Logger */
+    private static final Logger logger = LoggerFactory.getLogger(StaticEndpoint.class);
+
     private Message message;
 
     public StaticEndpoint() {
@@ -65,7 +71,7 @@ public void send(Message message, TestContext context) {
 
             @Override
             public String getName() {
-                return StaticEndpoint.this.getName() + "-producer";
+                return StaticEndpoint.this.getProducerName();
             }
         };
     }
@@ -75,17 +81,24 @@ public Consumer createConsumer() {
         return new Consumer() {
             @Override
             public Message receive(TestContext context) {
-                return getMessage();
+                return receive(context, getEndpointConfiguration().getTimeout());
             }
 
             @Override
             public Message receive(TestContext context, long timeout) {
-                return getMessage();
+                Message received = getMessage();
+                if (received == null) {
+                    throw new MessageTimeoutException(timeout, StaticEndpoint.this.getName());
+                }
+
+                logger.info("Received message from static endpoint: '{}'", StaticEndpoint.this.getName());
+
+                return received;
             }
 
             @Override
             public String getName() {
-                return StaticEndpoint.this.getName() + "-consumer";
+                return StaticEndpoint.this.getConsumerName();
             }
         };
     }
@@ -100,6 +113,10 @@ public void setMessage(Message message) {
     }
 
     public Message getMessage() {
+        if (message == null) {
+            return null;
+        }
+
         if (getEndpointConfiguration().isReuseMessage()) {
             return message;
         } else {

core/citrus-base/src/main/java/org/citrusframework/endpoint/context/ContextEndpoints.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.citrusframework.endpoint.context;
+
+/**
+ * Combines all test context related endpoint builders.
+ */
+public class ContextEndpoints {
+
+    /**
+     * Private constructor because static instantiation method should be used.
+     */
+    private ContextEndpoints() {
+    }
+
+    /**
+     * Static entry method for in memory endpoint builders.
+     */
+    public static ContextEndpoints context() {
+        return new ContextEndpoints();
+    }
+
+    public MessageStoreEndpointBuilder messageStore() {
+        return new MessageStoreEndpointBuilder();
+    }
+}