[RHCLOUD-36586] Various improvements of email connector:

connector-common/src/test/java/com/redhat/cloud/notifications/connector/ConnectorRoutesTest.java

@@ -42,7 +42,7 @@
 
 public abstract class ConnectorRoutesTest extends CamelQuarkusTestSupport {
 
-    private static final String KAFKA_SOURCE_MOCK = "direct:kafka-source-mock";
+    public static final String KAFKA_SOURCE_MOCK = "direct:kafka-source-mock";
 
     @Inject
     protected ConnectorConfig connectorConfig;

connector-email/pom.xml

@@ -66,6 +66,12 @@
         <!-- Scope: test -->
         <!-- Some test dependencies are declared in the "profiles" section of this POM. -->
 
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-junit5-mockito</artifactId>
+            <scope>test</scope>
+        </dependency>
+
         <!-- Camel -->
         <dependency>
             <groupId>org.apache.camel.quarkus</groupId>

connector-email/src/main/java/com/redhat/cloud/notifications/connector/email/CloudEventHistoryBuilder.java

@@ -0,0 +1,31 @@
+package com.redhat.cloud.notifications.connector.email;
+
+import com.redhat.cloud.notifications.connector.http.HttpOutgoingCloudEventBuilder;
+import io.vertx.core.json.JsonObject;
+import jakarta.annotation.Priority;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Alternative;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+
+@ApplicationScoped
+@Alternative
+@Priority(0) // The value doesn't matter.
+public class CloudEventHistoryBuilder extends HttpOutgoingCloudEventBuilder {
+
+    public static final String TOTAL_RECIPIENTS_KEY = "total_recipients";
+
+    @Override
+    public void process(Exchange exchange) throws Exception {
+        super.process(exchange);
+        int totalRecipients = exchange.getProperty(TOTAL_RECIPIENTS_KEY, 0, Integer.class);
+
+        Message in = exchange.getIn();
+        JsonObject cloudEvent = new JsonObject(in.getBody(String.class));
+        JsonObject data = new JsonObject(cloudEvent.getString("data"));
+        data.getJsonObject("details").put(TOTAL_RECIPIENTS_KEY, totalRecipients);
+
+        cloudEvent.put("data", data.encode());
+        in.setBody(cloudEvent.encode());
+    }
+}

connector-email/src/main/java/com/redhat/cloud/notifications/connector/email/EmailCloudEventDataExtractor.java

@@ -3,14 +3,12 @@
 import com.redhat.cloud.notifications.connector.CloudEventDataExtractor;
 import com.redhat.cloud.notifications.connector.email.config.EmailConnectorConfig;
 import com.redhat.cloud.notifications.connector.email.constants.ExchangeProperty;
-import com.redhat.cloud.notifications.connector.email.model.settings.RecipientSettings;
-import io.vertx.core.json.JsonArray;
+import com.redhat.cloud.notifications.connector.email.model.EmailNotification;
 import io.vertx.core.json.JsonObject;
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
 import org.apache.camel.Exchange;
 
-import java.util.List;
 import java.util.Set;
 
 import static java.util.stream.Collectors.toSet;
@@ -30,38 +28,21 @@ public class EmailCloudEventDataExtractor extends CloudEventDataExtractor {
     @Override
     public void extract(final Exchange exchange, final JsonObject cloudEventData) {
 
-        final List<RecipientSettings> recipientSettings = cloudEventData.getJsonArray("recipient_settings")
-            .stream()
-            .map(JsonObject.class::cast)
-            .map(jsonSetting -> jsonSetting.mapTo(RecipientSettings.class))
-            .toList();
-
-        final Set<String> subscribers = cloudEventData.getJsonArray("subscribers", JsonArray.of())
-            .stream()
-            .map(String.class::cast)
-            .collect(toSet());
-
-        final Set<String> unsubscribers = cloudEventData.getJsonArray("unsubscribers", JsonArray.of())
-                .stream()
-                .map(String.class::cast)
-                .collect(toSet());
-
-        final JsonObject recipientsAuthorizationCriterion = cloudEventData.getJsonObject("recipients_authorization_criterion");
-
-        final Set<String> emails = recipientSettings.stream()
+        EmailNotification emailNotification =  cloudEventData.mapTo(EmailNotification.class);
+        final Set<String> emails = emailNotification.recipientSettings().stream()
                 .filter(settings -> settings.getEmails() != null)
                 .flatMap(settings -> settings.getEmails().stream())
                 .collect(toSet());
 
-        exchange.setProperty(ExchangeProperty.RENDERED_BODY, cloudEventData.getString("email_body"));
-        exchange.setProperty(ExchangeProperty.RENDERED_SUBJECT, cloudEventData.getString("email_subject"));
-        exchange.setProperty(ExchangeProperty.RECIPIENT_SETTINGS, recipientSettings);
-        exchange.setProperty(ExchangeProperty.SUBSCRIBED_BY_DEFAULT, cloudEventData.getBoolean("subscribed_by_default"));
-        exchange.setProperty(ExchangeProperty.SUBSCRIBERS, subscribers);
-        exchange.setProperty(ExchangeProperty.UNSUBSCRIBERS, unsubscribers);
-        exchange.setProperty(ExchangeProperty.RECIPIENTS_AUTHORIZATION_CRITERION, recipientsAuthorizationCriterion);
+        exchange.setProperty(ExchangeProperty.RENDERED_BODY, emailNotification.emailBody());
+        exchange.setProperty(ExchangeProperty.RENDERED_SUBJECT, emailNotification.emailSubject());
+        exchange.setProperty(ExchangeProperty.RECIPIENT_SETTINGS, emailNotification.recipientSettings());
+        exchange.setProperty(ExchangeProperty.SUBSCRIBED_BY_DEFAULT, emailNotification.subscribedByDefault());
+        exchange.setProperty(ExchangeProperty.SUBSCRIBERS, emailNotification.subscribers());
+        exchange.setProperty(ExchangeProperty.UNSUBSCRIBERS, emailNotification.unsubscribers());
+        exchange.setProperty(ExchangeProperty.RECIPIENTS_AUTHORIZATION_CRITERION, emailNotification.recipientsAuthorizationCriterion());
         exchange.setProperty(ExchangeProperty.EMAIL_RECIPIENTS, emails);
-        exchange.setProperty(ExchangeProperty.EMAIL_SENDER, cloudEventData.getString("email_sender"));
+        exchange.setProperty(ExchangeProperty.EMAIL_SENDER, emailNotification.emailSender());
 
         exchange.setProperty(ExchangeProperty.USE_EMAIL_BOP_V1_SSL, emailConnectorConfig.isEnableBopServiceWithSslChecks());
     }

connector-email/src/main/java/com/redhat/cloud/notifications/connector/email/model/EmailNotification.java

@@ -0,0 +1,39 @@
+package com.redhat.cloud.notifications.connector.email.model;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.redhat.cloud.notifications.connector.email.model.settings.RecipientSettings;
+import io.vertx.core.json.JsonObject;
+import java.util.Collection;
+
+/**
+ * Represents the data structure that the email connector is expecting.
+ * @param emailBody             the rendered body of the email to be sent.
+ * @param emailSubject          the rendered subject of the email to be sent.
+ * @param emailSender           the sender that will appear in the email when
+ *                              the user receives it.
+ * @param orgId                 the organization ID associated with the
+ *                              triggered event.
+ * @param recipientSettings     the collection of recipient settings extracted
+ *                              from both the event and the related endpoints
+ *                              to the event.
+ * @param subscribers           the list of usernames who subscribed to the
+ *                              event type.
+ * @param unsubscribers         the list of usernames who unsubscribed from the
+ *                              event type.
+ * @param subscribedByDefault   true if the event type is subscribed by
+ *                              default.
+ * @param recipientsAuthorizationCriterion   forward received authorization criterion.
+ *
+ */
+public record EmailNotification(
+    @JsonProperty("email_body")             String emailBody,
+    @JsonProperty("email_subject")          String emailSubject,
+    @JsonProperty("email_sender")           String emailSender,
+    @JsonProperty("orgId")                  String orgId,
+    @JsonProperty("recipient_settings")     Collection<RecipientSettings> recipientSettings,
+    @JsonProperty("subscribers")            Collection<String> subscribers,
+    @JsonProperty("unsubscribers")          Collection<String> unsubscribers,
+    @JsonProperty("subscribed_by_default")  boolean subscribedByDefault,
+    @JsonProperty("recipients_authorization_criterion") JsonObject recipientsAuthorizationCriterion
+)  { }
+

connector-email/src/main/java/com/redhat/cloud/notifications/connector/email/processors/recipients/RecipientsResolverResponseProcessor.java

@@ -16,6 +16,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 
+import static com.redhat.cloud.notifications.connector.email.CloudEventHistoryBuilder.TOTAL_RECIPIENTS_KEY;
 import static java.util.stream.Collectors.toSet;
 
 @ApplicationScoped
@@ -49,6 +50,7 @@ public void process(final Exchange exchange) throws JsonProcessingException {
             emails.removeAll(forbiddenEmail);
         }
         recipientsList.addAll(emails);
+        exchange.setProperty(TOTAL_RECIPIENTS_KEY, recipientsList.size());
 
         // We have to remove one from the limit, because a default recipient (like [email protected]) will be automatically added
         exchange.setProperty(ExchangeProperty.FILTERED_USERS, partition(recipientsList, emailConnectorConfig.getMaxRecipientsPerEmail() - 1));

connector-email/src/main/resources/application.properties

@@ -11,6 +11,8 @@ notifications.connector.seda.concurrent-consumers=20
 # The following value matches the default value of the `connectionsPerRoute` option from the Camel `http` component.
 notifications.connector.seda.queue-size=20
 notifications.connector.supported-connector-headers=${notifications.connector.name}
+# Re-injections should not be enabled for this connector
+notifications.connector.kafka.maximum-reinjections=0
 
 quarkus.http.port=9003
 

connector-email/src/test/java/com/redhat/cloud/notifications/connector/email/BuildBopEndpointTest.java

@@ -0,0 +1,59 @@
+package com.redhat.cloud.notifications.connector.email;
+
+import com.redhat.cloud.notifications.connector.email.config.EmailConnectorConfig;
+import com.redhat.cloud.notifications.connector.http.SslTrustAllManager;
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.mockito.InjectSpy;
+import jakarta.inject.Inject;
+import org.apache.camel.Endpoint;
+import org.apache.camel.quarkus.test.CamelQuarkusTestSupport;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import static org.mockito.Mockito.when;
+
+@QuarkusTest
+public class BuildBopEndpointTest extends CamelQuarkusTestSupport {
+
+    @InjectSpy
+    EmailConnectorConfig emailConnectorConfig;
+
+    @Inject
+    EmailRouteBuilder emailRouteBuilder;
+
+    @Override
+    public boolean isUseAdviceWith() {
+        return true;
+    }
+
+    /**
+     * Disables the route builder to ensure that the Camel Context does not get
+     * started before the routes have been advised. More information is
+     * available at the <a href="https://people.apache.org/~dkulp/camel/camel-test.html">dkulp's Apache Camel Test documentation page</a>.
+     * @return {@code false} in order to stop the Camel Context from booting
+     * before the routes have been advised.
+     */
+    @Override
+    public boolean isUseRouteBuilder() {
+        return false;
+    }
+
+    /**
+     * Tests that the function under test creates the BOP endpoint with the
+     * {@link SslTrustAllManager} class as the SSL context parameters, and that
+     * that class is essentially a NOOP class.
+     * @throws Exception if the endpoint could not be created.
+     */
+    @Test
+    void testBuildBOPEndpoint() throws Exception {
+        String initialBopUrl = emailConnectorConfig.getBopURL();
+        when(emailConnectorConfig.getBopURL()).thenReturn("https://test.com");
+
+        Endpoint bopEndpoint = this.emailRouteBuilder.setUpBOPEndpointV1().resolve(this.context);
+        Assertions.assertEquals(this.emailConnectorConfig.getBopURL(), bopEndpoint.getEndpointBaseUri(), "the base URI of the endpoint is not the same as the one set through the properties");
+
+        final String bopEndpointURI = bopEndpoint.getEndpointUri();
+        Assertions.assertTrue(bopEndpointURI.contains("trustManager%3Dcom.redhat.cloud.notifications.connector.http.SslTrustAllManager"), "the endpoint does not contain a reference to the SslTrustAllManager");
+        Assertions.assertTrue(bopEndpointURI.contains("x509HostnameVerifier=NO_OP"), "the base URI does not contain a mention to the NO_OP hostname verifier");
+    }
+}