Open-sourced generic, dynamic POC, RESTful alerting API

amqp/build.gradle.kts

@@ -0,0 +1,19 @@
+import org.springframework.boot.gradle.tasks.bundling.BootJar
+
+dependencies {
+    implementation("org.springframework.amqp:spring-amqp")
+    implementation("org.springframework.amqp:spring-rabbit")
+    implementation("com.google.code.gson:gson")
+    implementation("org.projectlombok:lombok")
+    implementation("org.apache.commons:commons-lang3")
+
+    annotationProcessor("org.projectlombok:lombok")
+}
+
+tasks.getByName<BootJar>("bootJar") {
+    enabled = false
+}
+
+tasks.getByName<Jar>("jar") {
+    enabled = true
+}

amqp/src/main/java/com/poc/alerting/amqp/AmqpMessage.java

@@ -0,0 +1,11 @@
+package com.poc.alerting.amqp;
+
+import java.io.Serializable;
+
+public interface AmqpMessage extends Serializable {
+    String correlationId = "correlation-id";
+
+    String getRoutingKey();
+
+    String getExchange();
+}

amqp/src/main/java/com/poc/alerting/amqp/AmqpResponse.java

@@ -0,0 +1,16 @@
+package com.poc.alerting.amqp;
+
+import java.io.Serializable;
+
+import lombok.Getter;
+import lombok.Setter;
+
+@Getter
+@Setter
+public abstract class AmqpResponse<T> implements Serializable {
+    private T value;
+    private ExceptionType exceptionType;
+    private String exceptionMessage;
+    private Integer errorCode;
+    private String errorDescription;
+}

amqp/src/main/java/com/poc/alerting/amqp/ExceptionType.java

@@ -0,0 +1,21 @@
+package com.poc.alerting.amqp;
+
+public enum ExceptionType {
+    INVALID_REQUEST_EXCEPTION(400),
+    INVALID_ACCOUNT_EXCEPTION(403),
+    NOT_FOUND_EXCEPTION(404),
+    REQUEST_TIMEOUT_EXCEPTION(408),
+    CONFLICT_EXCEPTION(409),
+    UNPROCESSABLE_ENTITY_EXCEPTION(422),
+    INTERNAL_SERVER_EXCEPTION(500);
+
+    private final int status;
+
+    ExceptionType(final int status) {
+        this.status = status;
+    }
+
+    public int getStatus() {
+        return status;
+    }
+}

amqp/src/main/java/com/poc/alerting/amqp/GsonMessageConverter.java

@@ -0,0 +1,79 @@
+package com.poc.alerting.amqp;
+
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.util.Date;
+import java.util.UUID;
+
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.support.converter.MessageConversionException;
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.stereotype.Component;
+import org.springframework.util.StringUtils;
+
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Component
+@Slf4j
+public class GsonMessageConverter implements MessageConverter {
+    private static final Gson GSON = new GsonBuilder().setDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'").create();
+
+    @Override
+    public Message toMessage(final Object object, final MessageProperties messageProperties) throws MessageConversionException {
+        if (!(object instanceof Serializable)) {
+            throw new MessageConversionException("Message object is not serializable");
+        }
+
+        try {
+            final String json = GSON.toJson(object);
+            if (json == null) {
+                throw new MessageConversionException("Unable to serialize the message to JSON");
+            }
+
+            LOG.debug("json = {}", json);
+
+            final byte[] bytes = json.getBytes(StandardCharsets.UTF_8);
+            messageProperties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
+            messageProperties.setContentEncoding("UTF-8");
+            messageProperties.setContentLength(bytes.length);
+            messageProperties.setTimestamp(new Date());
+            messageProperties.setType(object.getClass().getName());
+            if (messageProperties.getMessageId() == null) {
+                messageProperties.setMessageId(UUID.randomUUID().toString());
+            }
+
+            return new Message(bytes, messageProperties);
+        } catch (final Exception e) {
+            throw new MessageConversionException(e.getMessage(), e);
+        }
+    }
+
+    @Override
+    public Object fromMessage(final Message message) throws MessageConversionException {
+        final byte[] messageBody = message.getBody();
+        if (messageBody == null) {
+            LOG.warn("No message body found for message: {}", message);
+            return null;
+        }
+
+        final MessageProperties messageProperties = message.getMessageProperties();
+        final String className = StringUtils.trimAllWhitespace(messageProperties.getType());
+        if (StringUtils.isEmpty(className)) {
+            LOG.error("Could not determine class from message: {}", message);
+            return null;
+        }
+
+        try {
+            final String json = new String(messageBody, StandardCharsets.UTF_8);
+            LOG.debug("json = {}", json);
+            return GSON.fromJson(json, Class.forName(className));
+        } catch (final Exception e) {
+            LOG.error("Could not deserialize message: " + message, e);
+            throw new MessageConversionException(e.getMessage(), e);
+        }
+    }
+}

amqp/src/main/java/com/poc/alerting/amqp/MessageDataTransferObject.java

@@ -0,0 +1,16 @@
+package com.poc.alerting.amqp;
+
+import java.io.Serializable;
+
+import org.springframework.amqp.core.Message;
+
+import lombok.Getter;
+import lombok.Setter;
+
+@Getter
+@Setter
+public class MessageDataTransferObject implements Serializable {
+    private String routingKey;
+    private String exchange;
+    private Message message;
+}

amqp/src/main/java/com/poc/alerting/amqp/RabbitSender.java

@@ -0,0 +1,50 @@
+package com.poc.alerting.amqp;
+
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+@Component
+@Slf4j
+@RequiredArgsConstructor
+public class RabbitSender {
+    private final RabbitTemplate rabbitTemplate;
+
+    public void send(final AmqpMessage amqpMessage) {
+        try {
+            LOG.info("Sending message: {}", amqpMessage.toString());
+            rabbitTemplate.convertAndSend(amqpMessage.getExchange(), amqpMessage.getRoutingKey(), amqpMessage);
+        } catch (final Exception e) {
+            LOG.error("Error sending message, serializing to disk", e);
+        }
+    }
+
+    public <T> T sendAndReceive(final AmqpMessage amqpMessage) {
+        final AmqpResponse<T> amqpResponse = (AmqpResponse) rabbitTemplate.convertSendAndReceive(amqpMessage.getExchange(), amqpMessage.getRoutingKey(), amqpMessage);
+        final String errorMessage = "Something went wrong";
+        if (amqpResponse == null) {
+            LOG.error(errorMessage);
+        }
+        final ExceptionType exceptionType = amqpResponse.getExceptionType();
+        if (exceptionType != null) {
+            final String exceptionMessage = amqpResponse.getExceptionMessage();
+            final int statusCode = exceptionType.getStatus();
+            if (amqpResponse.getErrorCode() != null) {
+                final Integer errorCode = amqpResponse.getErrorCode();
+                final String errorDescription = amqpResponse.getErrorDescription();
+                LOG.error(errorMessage);
+            } else {
+                LOG.error(errorMessage);
+            }
+        }
+        return amqpResponse.getValue();
+    }
+
+    public void sendWithoutBackup(final AmqpMessage amqpMessage) {
+        LOG.info("Sending message: {}", amqpMessage.toString());
+        rabbitTemplate.convertAndSend(amqpMessage.getExchange(), amqpMessage.getRoutingKey(), amqpMessage);
+    }
+}

amqp/src/main/kotlin/com/poc/alerting/amqp/AlertingAmqpMessage.kt

@@ -0,0 +1,20 @@
+package com.poc.alerting.amqp
+
+sealed class AlertingAmqpMessage(
+    val alertId: String,
+    val accountId: String
+): AmqpMessage {
+    override fun getRoutingKey(): String {
+        return "account_$accountId"
+    }
+
+    override fun getExchange(): String {
+        return "poc_alerting"
+    }
+
+    class Add(val frequency: String, alertId: String, accountId: String): AlertingAmqpMessage(alertId, accountId)
+    class Update(val frequency: String, alertId: String, accountId: String): AlertingAmqpMessage(alertId, accountId)
+    class Delete(alertId: String, accountId: String): AlertingAmqpMessage(alertId, accountId)
+    class Pause(alertId: String, accountId: String): AlertingAmqpMessage(alertId, accountId)
+    class Resume(alertId: String, accountId: String): AlertingAmqpMessage(alertId, accountId)
+}

amqp/src/main/kotlin/com/poc/alerting/amqp/AmqpConfiguration.kt

@@ -0,0 +1,79 @@
+package com.poc.alerting.amqp
+
+import org.apache.commons.lang3.time.DateUtils.MILLIS_PER_MINUTE
+import org.springframework.amqp.core.Binding
+import org.springframework.amqp.core.BindingBuilder
+import org.springframework.amqp.core.DirectExchange
+import org.springframework.amqp.core.Exchange
+import org.springframework.amqp.core.ExchangeBuilder
+import org.springframework.amqp.core.FanoutExchange
+import org.springframework.amqp.core.Queue
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory
+import org.springframework.amqp.rabbit.connection.ConnectionFactory
+import org.springframework.amqp.rabbit.core.RabbitTemplate
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.ComponentScan
+import org.springframework.context.annotation.Configuration
+
+@Configuration
+@ComponentScan(basePackages = ["com.poc.alerting.amqp"])
+open class AmqpConfiguration {
+    companion object {
+        const val AMQP_NAME = "poc_alerting"
+        const val NEW_ACCOUNT = "new_account"
+        const val FANOUT_NAME = "${AMQP_NAME}_fanout"
+    }
+
+    @Bean
+    open fun connectionFactory(): ConnectionFactory {
+        return CachingConnectionFactory().apply {
+            virtualHost = "poc"
+            username = "poc_user"
+            setPassword("s!mpleP@ssw0rd")
+            setAddresses("localhost")
+        }
+    }
+
+    @Bean
+    open fun newAccountQueue(): Queue {
+        return Queue(NEW_ACCOUNT, true)
+    }
+
+    @Bean
+    open fun newAccountExchange(): FanoutExchange {
+        return FanoutExchange(NEW_ACCOUNT)
+    }
+
+    @Bean
+    open fun newAccountBinding(newAccountQueue: Queue, newAccountExchange: FanoutExchange): Binding {
+        return BindingBuilder.bind(newAccountQueue).to(newAccountExchange)
+    }
+
+    @Bean
+    open fun pocAlertingExchange(): FanoutExchange {
+        return FanoutExchange(FANOUT_NAME)
+    }
+
+    @Bean
+    open fun directExchange(): DirectExchange {
+        return ExchangeBuilder.directExchange(AMQP_NAME)
+            .durable(false)
+            .withArgument("alternate-exchange", NEW_ACCOUNT)
+            .build()
+    }
+
+    @Bean
+    open fun exchangeBinding(directExchange: Exchange, pocAlertingExchange: FanoutExchange): Binding {
+        return BindingBuilder.bind(directExchange).to(pocAlertingExchange)
+    }
+
+    @Bean
+    open fun rabbitTemplate(connectionFactory: ConnectionFactory, gsonMessageConverter: GsonMessageConverter): RabbitTemplate {
+        return RabbitTemplate(connectionFactory).apply {
+            setExchange(FANOUT_NAME)
+            messageConverter = gsonMessageConverter
+            setReplyTimeout(MILLIS_PER_MINUTE)
+            setMandatory(true)
+        }
+    }
+}