feat(RabbitmqRPC): implement basic support for RabbitMQ RPC.

src/main/java/com/cleverthis/authservice/controller/rabbitmq/AbstractEndpoint.java

@@ -0,0 +1,144 @@
+package com.cleverthis.authservice.controller.rabbitmq;
+
+import com.cleverthis.authservice.controller.rabbitmq.exception.EndpointBadRequestException;
+import com.cleverthis.authservice.controller.rabbitmq.exception.EndpointException;
+import com.cleverthis.authservice.controller.rabbitmq.model.EndpointRequest;
+import com.cleverthis.authservice.controller.rabbitmq.model.EndpointResponses;
+import com.cleverthis.clevermicro.client.amqp.CleverMicroAmqpClient;
+import com.cleverthis.clevermicro.client.amqp.handler.HandlerContext;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.rabbitmq.client.BuiltinExchangeType;
+import jakarta.annotation.PreDestroy;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Map;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.web.bind.annotation.RequestParam;
+
+/**
+ * Abstract class for endpoint of auth-service over RabbitMQ.
+ */
+@Slf4j
+public abstract class AbstractEndpoint implements AutoCloseable {
+    public static final String ENDPOINT_EXCHANGE = "cleverthis.clevermicro.auth.endpoints";
+    public static final BuiltinExchangeType ENDPOINT_EXCHANGE_TYPE = BuiltinExchangeType.DIRECT;
+
+    private final String handlerTag;
+    private final CleverMicroAmqpClient client;
+    private final ObjectMapper objectMapper;
+
+    private final String queueName;
+
+    /**
+     * Initialize this abstract class.
+     */
+    public AbstractEndpoint(
+          @RequestParam final CleverMicroAmqpClient client,
+          final ObjectMapper objectMapper,
+          final String routingKey
+    ) throws IOException {
+        this.client = client;
+        this.objectMapper = objectMapper;
+
+        this.queueName = ENDPOINT_EXCHANGE + "." + routingKey;
+        client.getHandlerManager().declareExchange(ENDPOINT_EXCHANGE, ENDPOINT_EXCHANGE_TYPE);
+        client.getHandlerManager().declareQueue(
+              this.queueName, true, false, false, Map.of()
+        );
+        client.getHandlerManager().bindQueue(
+              this.queueName, ENDPOINT_EXCHANGE, routingKey, Map.of()
+        );
+        this.handlerTag = client.getHandlerManager().registerHandler(
+              this.queueName, "", 1,
+              ctx -> new Thread(() -> {
+                  try {
+                      final var req = parseRequest(ctx);
+                      if (req == null) {
+                          throw new EndpointBadRequestException(
+                                "Malformed request, cannot be parsed");
+                      }
+                      handleRequest(ctx, req);
+                  } catch (final EndpointException ex) {
+                      final var resp = EndpointResponses.error(ex);
+                      reply(ctx, resp);
+                  } catch (final Throwable t) {
+                      log.error(
+                            "Error when handling request for queue {}, only refill availability",
+                            this.queueName, t
+                      );
+                      final var resp = EndpointResponses.internalServerError(
+                            "Error when handling the request", t);
+                      reply(ctx, resp);
+                  }
+              }).start(),
+              tag -> log.info("Endpoint canceled for queue {}: handlerTag={}",
+                    this.queueName, tag)
+        );
+    }
+
+    private EndpointRequest parseRequest(final HandlerContext ctx) {
+        InputStream reqStream = null;
+        if (ctx.getMessageBodySingleStream().isPresent()) {
+            reqStream = ctx.getMessageBodySingleStream().get();
+        } else if (ctx.getMessageBodyMultiStream().isPresent()) {
+            final var multibody = ctx.getMessageBodyMultiStream().get();
+            reqStream = multibody.get("request");
+        }
+        // check null
+        if (reqStream == null) {
+            return null;
+        }
+        // parsing, need to convert to final
+        try (final var stream = reqStream) {
+            return this.objectMapper.readValue(
+                  stream,
+                  EndpointRequest.class
+            );
+        } catch (final IOException ex) {
+            log.error("Error when parsing single stream request for queue: {}", this.queueName, ex);
+            return null;
+        }
+    }
+
+    /**
+     * Handle a request.
+     */
+    protected abstract void handleRequest(
+          final HandlerContext ctx,
+          final EndpointRequest request
+    ) throws EndpointException;
+
+    /**
+     * Try to reply. Refill counter if error.
+     */
+    protected final void reply(final HandlerContext ctx, final Object response) {
+        final byte[] bytes;
+        try {
+            bytes = this.objectMapper.writerWithDefaultPrettyPrinter()
+                  .writeValueAsBytes(response);
+        } catch (final JsonProcessingException ex) {
+            log.error("Error when serializing response, ignore reply and refill counter. Queue {}",
+                  this.queueName, ex);
+            ctx.refillAvailability();
+            return;
+        }
+        try {
+            ctx.finish(bytes);
+        } catch (final IOException ex) {
+            // the counter should be refilled before exception
+            log.error("Error when replying request for queue {}", this.queueName, ex);
+        }
+    }
+
+    /**
+     * Close the endpoint and release resources.
+     * Should be called in {@link jakarta.annotation.PreDestroy} annotated methods.
+     */
+    @Override
+    @PreDestroy
+    public void close() {
+        log.info("Closing Endpoint for queue: {}, consumerTag={}", this.queueName, this.handlerTag);
+        this.client.getHandlerManager().cancelHandler(this.handlerTag);
+    }
+}

src/main/java/com/cleverthis/authservice/controller/rabbitmq/exception/EndpointBadRequestException.java

@@ -0,0 +1,21 @@
+package com.cleverthis.authservice.controller.rabbitmq.exception;
+
+/**
+ * Exception to signal the request is malformed or invalid.
+ */
+public class EndpointBadRequestException extends EndpointException {
+    private static final String ENDPOINT_EXCEPTION = "cleverthis.clevermicro.bad_request";
+
+    public EndpointBadRequestException(
+          final String message,
+          final Throwable cause
+    ) {
+        super(ENDPOINT_EXCEPTION, message, cause);
+    }
+
+    public EndpointBadRequestException(
+          final String message
+    ) {
+        super(ENDPOINT_EXCEPTION, message);
+    }
+}

src/main/java/com/cleverthis/authservice/controller/rabbitmq/exception/EndpointException.java

@@ -0,0 +1,28 @@
+package com.cleverthis.authservice.controller.rabbitmq.exception;
+
+import lombok.Getter;
+
+/**
+ * Exception for endpoints.
+ */
+public class EndpointException extends Exception {
+    @Getter
+    private final String endpointException;
+
+    public EndpointException(
+          final String endpointException,
+          final String message,
+          final Throwable cause
+    ) {
+        super(message, cause);
+        this.endpointException = endpointException;
+    }
+
+    public EndpointException(
+          final String endpointException,
+          final String message
+    ) {
+        super(message);
+        this.endpointException = endpointException;
+    }
+}

src/main/java/com/cleverthis/authservice/controller/rabbitmq/exception/EndpointInternalServerErrorException.java

@@ -0,0 +1,15 @@
+package com.cleverthis.authservice.controller.rabbitmq.exception;
+
+/**
+ * Exception to signal the handler is failed abnormally.
+ */
+public class EndpointInternalServerErrorException extends EndpointException {
+    private static final String ENDPOINT_EXCEPTION = "cleverthis.clevermicro.server_internal_error";
+
+    public EndpointInternalServerErrorException(
+          final String message,
+          final Throwable cause
+    ) {
+        super(ENDPOINT_EXCEPTION, message, cause);
+    }
+}

src/main/java/com/cleverthis/authservice/controller/rabbitmq/model/EndpointErrorResponse.java

@@ -0,0 +1,21 @@
+package com.cleverthis.authservice.controller.rabbitmq.model;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.Builder;
+import lombok.Data;
+
+/**
+ * The model for replied response.
+ */
+@Data
+@Builder
+public class EndpointErrorResponse {
+    @JsonProperty("type")
+    private final String type = "ERROR";
+    @JsonProperty("exception")
+    private final String exception;
+    @JsonProperty("message")
+    private final String message;
+    @JsonProperty("additional_info")
+    private final Object additionalInfo;
+}

src/main/java/com/cleverthis/authservice/controller/rabbitmq/model/EndpointOkResponse.java

@@ -0,0 +1,19 @@
+package com.cleverthis.authservice.controller.rabbitmq.model;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.List;
+import lombok.Builder;
+import lombok.Data;
+
+/**
+ * The model for replied response.
+ */
+@Data
+@Builder
+public class EndpointOkResponse {
+    @JsonProperty("type")
+    private final String type = "OK";
+    @JsonProperty("result")
+    private final List<?> result;
+
+}

src/main/java/com/cleverthis/authservice/controller/rabbitmq/model/EndpointRequest.java

@@ -0,0 +1,19 @@
+package com.cleverthis.authservice.controller.rabbitmq.model;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.util.LinkedHashMap;
+import lombok.Builder;
+import lombok.Data;
+
+/**
+ * The model for incoming requests.
+ */
+@Data
+@Builder
+public class EndpointRequest {
+    @JsonProperty("method")
+    private final String method;
+    @JsonProperty("args")
+    private final LinkedHashMap<String, JsonNode> args;
+}

src/main/java/com/cleverthis/authservice/controller/rabbitmq/model/EndpointResponses.java

@@ -0,0 +1,84 @@
+package com.cleverthis.authservice.controller.rabbitmq.model;
+
+import com.cleverthis.authservice.controller.rabbitmq.exception.EndpointException;
+import com.cleverthis.authservice.controller.rabbitmq.exception.EndpointInternalServerErrorException;
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.util.Arrays;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * The model for replied response.
+ */
+public class EndpointResponses {
+    private EndpointResponses() {
+    }
+
+    /**
+     * Create an error response from endpoint exception.
+     */
+    public static EndpointErrorResponse error(
+          final EndpointException ex
+    ) {
+        final var sw = new StringWriter();
+        final var pw = new PrintWriter(sw);
+        ex.printStackTrace(pw);
+        pw.flush();
+        pw.close();
+        return EndpointErrorResponse.builder()
+              .exception(ex.getEndpointException())
+              .message(ex.getMessage())
+              .additionalInfo(Map.of(
+                    "stacktrace", sw.toString()
+              ))
+              .build();
+    }
+
+    /**
+     * Create an error response from general exception.
+     */
+    public static EndpointErrorResponse internalServerError(
+          final String message,
+          final Throwable throwable
+    ) {
+        return EndpointResponses.error(
+              new EndpointInternalServerErrorException(
+                    message, throwable
+              )
+        );
+    }
+
+    /**
+     * Create an ok response from a result.
+     */
+    public static <T> EndpointOkResponse ok(final T result) {
+        return EndpointOkResponse.builder()
+              .result(List.of(result))
+              .build();
+    }
+
+    /**
+     * Create an ok response from a list of results.
+     */
+    public static <T> EndpointOkResponse ok(final Iterable<T> result) {
+        final var list = new LinkedList<T>();
+        result.forEach(list::add);
+        return EndpointOkResponse.builder()
+              .result(list)
+              .build();
+    }
+
+    /**
+     * Create an ok response from an array of results.
+     */
+    public static <T> EndpointOkResponse ok(final T[] result) {
+        // here we make a copy so the changes to the input array
+        // will not change the result field.
+        final var list = new LinkedList<>(Arrays.asList(result));
+        return EndpointOkResponse.builder()
+              .result(list)
+              .build();
+    }
+}