Merge broadcaster and pipeline

islandora-connector-broadcast/src/main/java/ca/islandora/alpaca/connector/broadcast/BroadcastRouter.java

@@ -19,6 +19,7 @@
 
 import static org.apache.camel.LoggingLevel.INFO;
 
+import org.apache.camel.ExchangePattern;
 import org.apache.camel.builder.RouteBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -37,14 +38,23 @@ public class BroadcastRouter extends RouteBuilder {
      */
     public void configure() throws Exception {
 
-        // Distribute message based on configured header.
+        // Distribute message based on headers.
         from("{{input.stream}}")
                 .routeId("MessageBroadcaster")
-                .description("Broadcast messages from one queue/topic to other specified queues/topics.")
+                .description("Broadcast messages from one queue/topic to other queues/topics")
                 .log(INFO, LOGGER,
                         "Distributing message: ${headers[JMSMessageID]} with timestamp ${headers[JMSTimestamp]}")
-                .recipientList(simple("${headers[IslandoraBroadcastRecipients]}"))
-                .ignoreInvalidEndpoints();
+                .filter(header("IslandoraExchangePattern"))
+                    .process(exchange -> {
+                        final String patternName = exchange.getIn().getHeader("IslandoraExchangePattern", String.class);
+                        try {
+                            exchange.setPattern(ExchangePattern.asEnum(patternName));
+                        } catch (IllegalArgumentException e) {
+                            LOGGER.warn("Ignoring malformed exchange pattern: " + patternName);
+                        }
+                    })
+                    .end()
+                .routingSlip(header("IslandoraBroadcastRecipients")).ignoreInvalidEndpoints();
     }
 }
 

karaf/src/main/resources/features.xml

@@ -31,15 +31,4 @@
 
   </feature>
 
- <feature name="islandora-connector-pipeline" version="${project.version}" start-level="80">
-    <details>Distributes a message to a list of queues/topics sequentially.</details>
-
-    <feature version="${fcrepoCamelToolboxVersion}">fcrepo-service-activemq</feature>
-
-    <bundle>mvn:ca.islandora.alpaca/islandora-connector-pipeline/${project.version}</bundle>
-
-    <configfile finalname="/etc/ca.islandora.alpaca.connector.pipeline.cfg">mvn:ca.islandora.alpaca/islandora-connector-pipeline/${project.version}/cfg/configuration</configfile>
-
-  </feature>
-
 </features>

settings.gradle

@@ -1,9 +1,7 @@
 include ':islandora-karaf'
 include ':islandora-indexing-triplestore'
 include ':islandora-connector-broadcast'
-include ':islandora-connector-pipeline'
 
 project(':islandora-karaf').projectDir = "$rootDir/karaf" as File
 project(':islandora-indexing-triplestore').projectDir = "$rootDir/islandora-indexing-triplestore" as File
 project(':islandora-connector-broadcast').projectDir = "$rootDir/islandora-connector-broadcast" as File
-project(':islandora-connector-pipeline').projectDir = "$rootDir/islandora-connector-pipeline" as File