RabbitMQ Message Publisher Module: Publish Task/Workflow Status Changes

community-server/build.gradle

@@ -36,6 +36,7 @@ dependencies {
     implementation project(':task:conductor-kafka')
 
     implementation project(':conductor-workflow-event-listener')
+    implementation project(':message-publisher:conductor-rabbitmq-message-publisher')
 
     implementation 'org.springframework.boot:spring-boot-starter'
     implementation 'org.springframework.boot:spring-boot-starter-validation'

community-server/src/main/resources/application.properties

@@ -127,3 +127,21 @@ management.metrics.export.datadog.enabled=false
 #optional - default Datadog instance is https://app.datadoghq.com/ 
 # management.metrics.export.datadog.uri=<your DD instance>
 #management.metrics.export.datadog.step=10s
+
+# Following properties set for using for publishing workflow/task status change event to RabbitMQ:
+# (To enable support of RabbitMQ publishers)
+
+#conductor.message-publisher.type=rabbitmq
+
+#conductor.workflow-status-listener.type=rabbitmq
+#conductor.task-status-listener.type=rabbitmq
+#conductor.message-publisher.rabbitmq.hosts=<rabbitmq serverip>
+#conductor.message-publisher.rabbitmq.username=<username>
+#conductor.message-publisher.rabbitmq.password=<password>
+#conductor.message-publisher.rabbitmq.port=5672
+#conductor.message-publisher.rabbitmq.workflowStatusExchange=workflow.status.exchange
+#conductor.message-publisher.rabbitmq.taskStatusExchange=task.status.exchange
+#conductor.message-publisher.rabbitmq.workflowStatusListenerEnabled=true
+#conductor.message-publisher.rabbitmq.allowedTaskStatuses=COMPLETED,IN_PROGRESS,FAILED
+# Task statuses
+# IN_PROGRESS,CANCELED,FAILED,FAILED_WITH_TERMINAL_ERROR,COMPLETED,COMPLETED_WITH_ERRORS,SCHEDULED,TIMED_OUT,SKIPPED

message-publisher/rabbitmq-message-publisher/build.gradle

@@ -0,0 +1,23 @@
+plugins {
+    id 'groovy'
+}
+dependencies {
+
+    implementation "com.netflix.conductor:conductor-common:${revConductor}"
+    implementation "com.netflix.conductor:conductor-core:${revConductor}"
+    implementation "com.netflix.conductor:conductor-amqp:${revConductor}"
+
+    implementation "com.rabbitmq:amqp-client:${revAmqpClient}"
+
+    compileOnly 'org.springframework.boot:spring-boot-starter'
+    compileOnly 'org.springframework.boot:spring-boot-starter-web'
+
+    testImplementation "org.codehaus.groovy:groovy-all:${revGroovy}"
+    testImplementation "org.spockframework:spock-core:${revSpock}"
+    testImplementation "org.spockframework:spock-spring:${revSpock}"
+
+
+    testImplementation "com.netflix.conductor:conductor-server:${revConductor}"
+    testImplementation 'org.springframework.boot:spring-boot-starter-web'
+    testImplementation project(':conductor-test-util').sourceSets.test.output
+}

message-publisher/rabbitmq-message-publisher/dependencies.lock

@@ -0,0 +1,261 @@
+{
+    "annotationProcessor": {
+        "org.springframework.boot:spring-boot-configuration-processor": {
+            "locked": "2.7.16"
+        }
+    },
+    "compileClasspath": {
+        "com.netflix.conductor:conductor-amqp": {
+            "project": true
+        },
+        "com.netflix.conductor:conductor-common": {
+            "locked": "3.15.0"
+        },
+        "com.netflix.conductor:conductor-core": {
+            "locked": "3.15.0"
+        },
+        "org.apache.logging.log4j:log4j-api": {
+            "locked": "2.17.2"
+        },
+        "org.apache.logging.log4j:log4j-core": {
+            "locked": "2.17.2"
+        },
+        "org.apache.logging.log4j:log4j-jul": {
+            "locked": "2.17.2"
+        },
+        "org.apache.logging.log4j:log4j-slf4j-impl": {
+            "locked": "2.17.2"
+        },
+        "org.apache.logging.log4j:log4j-web": {
+            "locked": "2.17.2"
+        },
+        "org.springframework.boot:spring-boot-starter": {
+            "locked": "2.7.16"
+        },
+        "org.springframework.boot:spring-boot-starter-web": {
+            "locked": "2.7.16"
+        }
+    },
+    "runtimeClasspath": {
+        "com.google.guava:guava": {
+            "firstLevelTransitive": [
+                "com.netflix.conductor:conductor-amqp"
+            ],
+            "locked": "32.1.2-jre"
+        },
+        "com.netflix.conductor:conductor-amqp": {
+            "project": true
+        },
+        "com.netflix.conductor:conductor-common": {
+            "firstLevelTransitive": [
+                "com.netflix.conductor:conductor-amqp"
+            ],
+            "locked": "3.15.0"
+        },
+        "com.netflix.conductor:conductor-core": {
+            "firstLevelTransitive": [
+                "com.netflix.conductor:conductor-amqp"
+            ],
+            "locked": "3.15.0"
+        },
+        "com.rabbitmq:amqp-client": {
+            "firstLevelTransitive": [
+                "com.netflix.conductor:conductor-amqp"
+            ],
+            "locked": "5.14.3"
+        },
+        "io.reactivex:rxjava": {
+            "firstLevelTransitive": [
+                "com.netflix.conductor:conductor-amqp"
+            ],
+            "locked": "1.3.8"
+        },
+        "org.apache.commons:commons-lang3": {
+            "firstLevelTransitive": [
+                "com.netflix.conductor:conductor-amqp"
+            ],
+            "locked": "3.12.0"
+        },
+        "org.apache.logging.log4j:log4j-api": {
+            "firstLevelTransitive": [
+                "com.netflix.conductor:conductor-amqp"
+            ],
+            "locked": "2.17.2"
+        },
+        "org.apache.logging.log4j:log4j-core": {
+            "firstLevelTransitive": [
+                "com.netflix.conductor:conductor-amqp"
+            ],
+            "locked": "2.17.2"
+        },
+        "org.apache.logging.log4j:log4j-jul": {
+            "firstLevelTransitive": [
+                "com.netflix.conductor:conductor-amqp"
+            ],
+            "locked": "2.17.2"
+        },
+        "org.apache.logging.log4j:log4j-slf4j-impl": {
+            "firstLevelTransitive": [
+                "com.netflix.conductor:conductor-amqp"
+            ],
+            "locked": "2.17.2"
+        },
+        "org.apache.logging.log4j:log4j-web": {
+            "firstLevelTransitive": [
+                "com.netflix.conductor:conductor-amqp"
+            ],
+            "locked": "2.17.2"
+        }
+    },
+    "testCompileClasspath": {
+        "com.netflix.conductor:conductor-amqp": {
+            "project": true
+        },
+        "com.netflix.conductor:conductor-common": {
+            "locked": "3.15.0"
+        },
+        "com.netflix.conductor:conductor-core": {
+            "locked": "3.15.0"
+        },
+        "com.netflix.conductor:conductor-server": {
+            "locked": "3.15.0"
+        },
+        "junit:junit": {
+            "locked": "4.13.2"
+        },
+        "org.apache.logging.log4j:log4j-api": {
+            "locked": "2.17.2"
+        },
+        "org.apache.logging.log4j:log4j-core": {
+            "locked": "2.17.2"
+        },
+        "org.apache.logging.log4j:log4j-jul": {
+            "locked": "2.17.2"
+        },
+        "org.apache.logging.log4j:log4j-slf4j-impl": {
+            "locked": "2.17.2"
+        },
+        "org.apache.logging.log4j:log4j-web": {
+            "locked": "2.17.2"
+        },
+        "org.codehaus.groovy:groovy-all": {
+            "locked": "3.0.19"
+        },
+        "org.junit.vintage:junit-vintage-engine": {
+            "locked": "5.8.2"
+        },
+        "org.spockframework:spock-core": {
+            "locked": "2.3-groovy-3.0"
+        },
+        "org.spockframework:spock-spring": {
+            "locked": "2.3-groovy-3.0"
+        },
+        "org.springframework.boot:spring-boot-starter-log4j2": {
+            "locked": "2.7.16"
+        },
+        "org.springframework.boot:spring-boot-starter-test": {
+            "locked": "2.7.16"
+        },
+        "org.springframework.boot:spring-boot-starter-web": {
+            "locked": "2.7.16"
+        }
+    },
+    "testRuntimeClasspath": {
+        "com.google.guava:guava": {
+            "firstLevelTransitive": [
+                "com.netflix.conductor:conductor-amqp"
+            ],
+            "locked": "32.1.2-jre"
+        },
+        "com.netflix.conductor:conductor-amqp": {
+            "project": true
+        },
+        "com.netflix.conductor:conductor-common": {
+            "firstLevelTransitive": [
+                "com.netflix.conductor:conductor-amqp"
+            ],
+            "locked": "3.15.0"
+        },
+        "com.netflix.conductor:conductor-core": {
+            "firstLevelTransitive": [
+                "com.netflix.conductor:conductor-amqp"
+            ],
+            "locked": "3.15.0"
+        },
+        "com.netflix.conductor:conductor-server": {
+            "locked": "3.15.0"
+        },
+        "com.rabbitmq:amqp-client": {
+            "firstLevelTransitive": [
+                "com.netflix.conductor:conductor-amqp"
+            ],
+            "locked": "5.14.3"
+        },
+        "io.reactivex:rxjava": {
+            "firstLevelTransitive": [
+                "com.netflix.conductor:conductor-amqp"
+            ],
+            "locked": "1.3.8"
+        },
+        "junit:junit": {
+            "locked": "4.13.2"
+        },
+        "org.apache.commons:commons-lang3": {
+            "firstLevelTransitive": [
+                "com.netflix.conductor:conductor-amqp"
+            ],
+            "locked": "3.12.0"
+        },
+        "org.apache.logging.log4j:log4j-api": {
+            "firstLevelTransitive": [
+                "com.netflix.conductor:conductor-amqp"
+            ],
+            "locked": "2.17.2"
+        },
+        "org.apache.logging.log4j:log4j-core": {
+            "firstLevelTransitive": [
+                "com.netflix.conductor:conductor-amqp"
+            ],
+            "locked": "2.17.2"
+        },
+        "org.apache.logging.log4j:log4j-jul": {
+            "firstLevelTransitive": [
+                "com.netflix.conductor:conductor-amqp"
+            ],
+            "locked": "2.17.2"
+        },
+        "org.apache.logging.log4j:log4j-slf4j-impl": {
+            "firstLevelTransitive": [
+                "com.netflix.conductor:conductor-amqp"
+            ],
+            "locked": "2.17.2"
+        },
+        "org.apache.logging.log4j:log4j-web": {
+            "firstLevelTransitive": [
+                "com.netflix.conductor:conductor-amqp"
+            ],
+            "locked": "2.17.2"
+        },
+        "org.codehaus.groovy:groovy-all": {
+            "locked": "3.0.19"
+        },
+        "org.junit.vintage:junit-vintage-engine": {
+            "locked": "5.8.2"
+        },
+        "org.spockframework:spock-core": {
+            "locked": "2.3-groovy-3.0"
+        },
+        "org.spockframework:spock-spring": {
+            "locked": "2.3-groovy-3.0"
+        },
+        "org.springframework.boot:spring-boot-starter-log4j2": {
+            "locked": "2.7.16"
+        },
+        "org.springframework.boot:spring-boot-starter-test": {
+            "locked": "2.7.16"
+        },
+        "org.springframework.boot:spring-boot-starter-web": {
+            "locked": "2.7.16"
+        }
+    }
+}

message-publisher/rabbitmq-message-publisher/src/main/java/com/netflix/conductor/rabbitmq/config/RabbitMQConfiguration.java

@@ -0,0 +1,81 @@
+/*
+ * <p>
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 com.netflix.conductor.rabbitmq.config;
+
+import com.netflix.conductor.core.listener.TaskStatusListener;
+import com.netflix.conductor.core.listener.WorkflowStatusListener;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import com.netflix.conductor.contribs.queue.amqp.AMQPConnection;
+import com.netflix.conductor.contribs.queue.amqp.config.AMQPRetryPattern;
+import com.netflix.conductor.rabbitmq.listener.TaskStatusPublisherRabbitMQ;
+import com.netflix.conductor.rabbitmq.listener.WorkflowStatusPublisherRabbitMQ;
+import com.netflix.conductor.rabbitmq.services.RabbitMQService;
+import com.netflix.conductor.rabbitmq.services.RabbitMQServiceImpl;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.rabbitmq.client.Address;
+import com.rabbitmq.client.ConnectionFactory;
+
+@Configuration
+@EnableConfigurationProperties(RabbitMQProperties.class)
+@ConditionalOnProperty(name = "conductor.message-publisher.type", havingValue = "rabbitmq")
+public class RabbitMQConfiguration {
+
+    @Bean
+    public AMQPConnection amqpConnection(RabbitMQProperties rabbitMQProperties) {
+        ConnectionFactory connectionFactory = new ConnectionFactory();
+
+        connectionFactory.setHost(rabbitMQProperties.getHosts());
+        connectionFactory.setPort(rabbitMQProperties.getPort());
+        connectionFactory.setUsername(rabbitMQProperties.getUsername());
+        connectionFactory.setPassword(rabbitMQProperties.getPassword());
+
+        Address[] addresses =
+                new Address[] {
+                    new Address(rabbitMQProperties.getHosts(), rabbitMQProperties.getPort())
+                };
+
+        AMQPRetryPattern retryPattern = new AMQPRetryPattern();
+
+        return AMQPConnection.getInstance(connectionFactory, addresses, retryPattern);
+    }
+
+    @Bean
+    public RabbitMQService rabbitMQService(
+            AMQPConnection amqpConnection, ObjectMapper objectMapper) {
+        return new RabbitMQServiceImpl(amqpConnection, objectMapper);
+    }
+
+    @ConditionalOnProperty(
+            name = "conductor.workflow-status-listener.type",
+            havingValue = "rabbitmq",
+            matchIfMissing = false)
+    @Bean
+    public WorkflowStatusListener workflowStatusListenerRabbitMQ(
+            RabbitMQService rabbitMQService, RabbitMQProperties rabbitMQProperties) {
+        return new WorkflowStatusPublisherRabbitMQ(rabbitMQService, rabbitMQProperties);
+    }
+
+    @ConditionalOnProperty(
+            name = "conductor.task-status-listener.type",
+            havingValue = "rabbitmq",
+            matchIfMissing = false)
+    @Bean
+    public TaskStatusListener taskStatusPublisherRabbitMQ(
+            RabbitMQService rabbitMQService, RabbitMQProperties rabbitMQProperties) {
+        return new TaskStatusPublisherRabbitMQ(rabbitMQService, rabbitMQProperties);
+    }
+}