gdc GET_STUDENT_COURSE messaging

.github/workflows/on.pr.yml

@@ -32,6 +32,8 @@ jobs:
           restore-keys: |
             ${{ runner.os }}-maven-5Jun-
       - name: Run unit tests
+        env:
+          SPRING_PROFILES_ACTIVE: test
         run: mvn -f pom.xml clean package
       - name: Run Trivy vulnerability scanner in repo mode
         uses: aquasecurity/[email protected]

api/pom.xml

@@ -28,6 +28,8 @@
         <log4j.version>2.18.0</log4j.version>
         <org.mapstruct.version>1.5.2.Final</org.mapstruct.version>
         <springdoc.version>1.6.9</springdoc.version>
+        <nats.version>2.11.0</nats.version>
+        <guava.version>33.2.1-jre</guava.version>
     </properties>
 
     <parent>
@@ -176,6 +178,16 @@
             <artifactId>spring-web</artifactId>
             <version>6.0.4</version>
         </dependency>
+        <dependency>
+            <groupId>io.nats</groupId>
+            <artifactId>jnats</artifactId>
+            <version>${nats.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <version>${guava.version}</version>
+        </dependency>
     </dependencies>
 
   <build>

api/src/main/java/ca/bc/gov/educ/api/course/constants/EventOutcome.java

@@ -0,0 +1,15 @@
+package ca.bc.gov.educ.api.course.constants;
+
+/**
+ * The enum Event outcome.
+ */
+public enum EventOutcome {
+    /**
+     * Course found event outcome.
+     */
+    COURSE_FOUND,
+    /**
+     * Course not found event outcome.
+     */
+    COURSE_NOT_FOUND,
+}

api/src/main/java/ca/bc/gov/educ/api/course/constants/EventStatus.java

@@ -0,0 +1,11 @@
+package ca.bc.gov.educ.api.course.constants;
+
+/**
+ * The enum Event status.
+ */
+public enum EventStatus {
+
+    DB_COMMITTED,
+
+    MESSAGE_PUBLISHED
+}

api/src/main/java/ca/bc/gov/educ/api/course/constants/EventType.java

@@ -0,0 +1,11 @@
+package ca.bc.gov.educ.api.course.constants;
+
+/**
+ * The enum Event type.
+ */
+public enum EventType {
+    /*
+        Get course event type
+     */
+    GET_STUDENT_COURSE
+}

api/src/main/java/ca/bc/gov/educ/api/course/constants/Topics.java

@@ -0,0 +1,16 @@
+package ca.bc.gov.educ.api.course.constants;
+
+/**
+ * The enum Topics.
+ */
+public enum Topics {
+    /**
+     * Grad course api topic topics.
+     */
+    GRAD_COURSE_API_TOPIC,
+
+    /**
+     * Grad course events topic.
+     */
+    GRAD_COURSE_EVENTS_TOPIC,
+}

api/src/main/java/ca/bc/gov/educ/api/course/helpers/LogHelper.java

@@ -0,0 +1,65 @@
+package ca.bc.gov.educ.api.course.helpers;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
+import lombok.extern.slf4j.Slf4j;
+import lombok.val;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.MDC;
+import org.springframework.lang.NonNull;
+
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
+
+@Slf4j
+public final class LogHelper {
+    private static final ObjectMapper mapper = new ObjectMapper();
+    private static final String EXCEPTION = "Exception ";
+    public static final String CORRELATION_ID = "correlationID";
+    private LogHelper() {
+
+    }
+
+    public static void logServerHttpReqResponseDetails(@NonNull final HttpServletRequest request, final HttpServletResponse response) {
+        try {
+            final int status = response.getStatus();
+            val totalTime = Instant.now().toEpochMilli() - (Long) request.getAttribute("startTime");
+            final Map<String, Object> httpMap = new HashMap<>();
+            httpMap.put("server_http_response_code", status);
+            httpMap.put("server_http_request_method", request.getMethod());
+            httpMap.put("server_http_query_params", request.getQueryString());
+            val correlationID = request.getHeader(CORRELATION_ID);
+            if (correlationID != null) {
+                httpMap.put("correlation_id", correlationID);
+            }
+            httpMap.put("server_http_request_url", String.valueOf(request.getRequestURL()));
+            httpMap.put("server_http_request_processing_time_ms", totalTime);
+            httpMap.put("server_http_request_payload", String.valueOf(request.getAttribute("payload")));
+            httpMap.put("server_http_request_remote_address", request.getRemoteAddr());
+            httpMap.put("server_http_request_client_name", StringUtils.trimToEmpty(request.getHeader("X-Client-Name")));
+            MDC.putCloseable("httpEvent", mapper.writeValueAsString(httpMap));
+            log.info("");
+            MDC.clear();
+        } catch (final Exception exception) {
+            log.error(EXCEPTION, exception);
+        }
+    }
+
+
+    /**
+     * the event is a json string.
+     *
+     * @param event the json string
+     */
+    public static void logMessagingEventDetails(final String event) {
+        try {
+            MDC.putCloseable("messageEvent", event);
+            log.info("");
+            MDC.clear();
+        } catch (final Exception exception) {
+            log.error(EXCEPTION, exception);
+        }
+    }
+}

api/src/main/java/ca/bc/gov/educ/api/course/messaging/MessagePublisher.java

@@ -0,0 +1,40 @@
+package ca.bc.gov.educ.api.course.messaging;
+
+import io.nats.client.Connection;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Profile;
+import org.springframework.stereotype.Component;
+
+/**
+ * This publisher will publish messages to NATS, for publishing directly to Jet Stream,
+ * please follow {@link ca.bc.gov.educ.api.course.messaging.jetstream.Publisher}.
+ */
+@Component
+@Slf4j
+@Profile("!test")
+public class MessagePublisher {
+
+
+    private final Connection connection;
+
+    /**
+     * Instantiates a new Message publisher.
+     *
+     * @param connection the nats connection
+     */
+    @Autowired
+    public MessagePublisher(final Connection connection) {
+        this.connection = connection;
+    }
+
+    /**
+     * Dispatch message.
+     *
+     * @param subject the subject
+     * @param message the message
+     */
+    public void dispatchMessage(String subject, byte[] message) {
+        connection.publish(subject, message);
+    }
+}

api/src/main/java/ca/bc/gov/educ/api/course/messaging/MessageSubscriber.java

@@ -0,0 +1,86 @@
+package ca.bc.gov.educ.api.course.messaging;
+
+import ca.bc.gov.educ.api.course.messaging.jetstream.Subscriber;
+import ca.bc.gov.educ.api.course.service.EventHandlerDelegatorService;
+import ca.bc.gov.educ.api.course.struct.Event;
+import ca.bc.gov.educ.api.course.util.JsonUtil;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import io.nats.client.Connection;
+import io.nats.client.Message;
+import io.nats.client.MessageHandler;
+import lombok.extern.slf4j.Slf4j;
+import org.jboss.threads.EnhancedQueueExecutor;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Profile;
+import org.springframework.stereotype.Component;
+
+import jakarta.annotation.PostConstruct;
+import java.time.Duration;
+import java.util.concurrent.Executor;
+
+import static ca.bc.gov.educ.api.course.constants.Topics.GRAD_COURSE_API_TOPIC;
+
+/**
+ * This is for subscribing directly to NATS. for subscribing to Jet Stream please
+ * follow {@link Subscriber}
+ */
+@Component
+@Slf4j
+@Profile("!test")
+public class MessageSubscriber {
+    private final Executor messageProcessingThreads;
+    private final EventHandlerDelegatorService eventHandlerDelegatorServiceV1;
+    private final Connection connection;
+
+    /**
+     * Instantiates a new Message subscriber.
+     *
+     * @param connection                     the nats connection
+     * @param eventHandlerDelegatorServiceV1 the event handler delegator service v 1
+     */
+    @Autowired
+    public MessageSubscriber(final Connection connection, EventHandlerDelegatorService eventHandlerDelegatorServiceV1) {
+        this.eventHandlerDelegatorServiceV1 = eventHandlerDelegatorServiceV1;
+        this.connection = connection;
+        messageProcessingThreads = new EnhancedQueueExecutor.Builder().setThreadFactory(new ThreadFactoryBuilder().setNameFormat("nats-message-subscriber-%d").build()).setCorePoolSize(10).setMaximumPoolSize(10).setKeepAliveTime(Duration.ofSeconds(60)).build();
+    }
+
+    /**
+     * This subscription will makes sure the messages are required to acknowledge manually to STAN.
+     * Subscribe.
+     */
+    @PostConstruct
+    public void subscribe() {
+        String queue = GRAD_COURSE_API_TOPIC.toString().replace("_", "-");
+        var dispatcher = connection.createDispatcher(onMessage());
+        dispatcher.subscribe(GRAD_COURSE_API_TOPIC.toString(), queue);
+    }
+
+    /**
+     * On message message handler.
+     *
+     * @return the message handler
+     */
+    private MessageHandler onMessage() {
+        return (Message message) -> {
+            if (message != null) {
+                try {
+                    var eventString = new String(message.getData());
+                    ca.bc.gov.educ.api.course.helpers.LogHelper.logMessagingEventDetails(eventString);
+                    var event = JsonUtil.getJsonObjectFromString(Event.class, eventString);
+                    if (event.getPayloadVersion() == null) {
+                        event.setPayloadVersion("V1");
+                    }
+                    //placeholder to have different versions
+                    if ("V1".equalsIgnoreCase(event.getPayloadVersion())) {
+                        messageProcessingThreads.execute(() -> eventHandlerDelegatorServiceV1.handleEvent(event, message));
+                    }
+                } catch (final Exception e) {
+                    log.error("Exception ", e);
+                }
+            }
+        };
+    }
+
+
+}

api/src/main/java/ca/bc/gov/educ/api/course/messaging/NatsConnection.java

@@ -0,0 +1,81 @@
+package ca.bc.gov.educ.api.course.messaging;
+
+import ca.bc.gov.educ.api.course.properties.ApplicationProperties;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import io.nats.client.Connection;
+import io.nats.client.ConnectionListener;
+import io.nats.client.Nats;
+import lombok.extern.slf4j.Slf4j;
+import org.jboss.threads.EnhancedQueueExecutor;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Profile;
+import org.springframework.stereotype.Component;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.time.Duration;
+
+/**
+ * The type Nats connection.
+ */
+@Component
+@Slf4j
+@Profile("!test")
+public class NatsConnection implements Closeable {
+
+    private final Connection natsCon;
+
+    /**
+     * Instantiates a new Nats connection.
+     *
+     * @param applicationProperties the application properties
+     * @throws IOException          the io exception
+     * @throws InterruptedException the interrupted exception
+     */
+    @Autowired
+    public NatsConnection(final ApplicationProperties applicationProperties) throws IOException, InterruptedException {
+        this.natsCon = connectToNats(applicationProperties.getNatsUrl(), applicationProperties.getNatsMaxReconnect());
+    }
+
+    private Connection connectToNats(String stanUrl, int maxReconnects) throws IOException, InterruptedException {
+        io.nats.client.Options natsOptions = new io.nats.client.Options.Builder()
+                .connectionListener(this::connectionListener)
+                .maxPingsOut(5)
+                .pingInterval(Duration.ofSeconds(2))
+                .connectionName("COREG-API")
+                .connectionTimeout(Duration.ofSeconds(5))
+                .executor(new EnhancedQueueExecutor.Builder()
+                        .setThreadFactory(new ThreadFactoryBuilder().setNameFormat("core-nats-%d").build())
+                        .setCorePoolSize(10).setMaximumPoolSize(50).setKeepAliveTime(Duration.ofSeconds(60)).build())
+                .maxReconnects(maxReconnects)
+                .reconnectWait(Duration.ofSeconds(2))
+                .servers(new String[]{stanUrl})
+                .build();
+        return Nats.connect(natsOptions);
+    }
+
+    private void connectionListener(Connection connection, ConnectionListener.Events events) {
+        log.info("NATS -> {}", events.toString());
+    }
+
+
+    @Override
+    public void close() {
+        if (natsCon != null) {
+            log.info("closing nats connection...");
+            try {
+                natsCon.close();
+            } catch (InterruptedException e) {
+                log.error("error while closing nats connection...", e);
+                Thread.currentThread().interrupt();
+            }
+            log.info("nats connection closed...");
+        }
+    }
+
+    @Bean
+    public Connection connection() {
+        return natsCon;
+    }
+}