ActiveMQ with persistent queues PoC

customer-solutions/supply-chain/evrythng.eif/.gitignore

@@ -1,7 +1,9 @@
 .gradle
 /build/
 /out/
+
 /src/data
+activemq-data/
 
 # Ignore Gradle GUI config
 gradle-app.setting

customer-solutions/supply-chain/evrythng.eif/build.gradle

@@ -1,5 +1,5 @@
 group 'com.evrythng'
-version '0.18.80'
+version '0.18.81'
 
 apply plugin: 'java'
 
@@ -23,6 +23,9 @@ dependencies {
     compile 'org.xmlunit:xmlunit-core:2.5.1'
     compile 'com.google.code.gson:gson:2.8.1'
 
+    compile group: 'org.apache.activemq', name: 'activemq-all', version: '5.4.2'
+    compile group: 'org.apache.activemq', name: 'activemq-camel', version: '5.15.3'
+
     testCompile group: 'junit', name: 'junit', version: '4.12'
 
     testCompile group: 'org.apache.camel', name: 'camel-test', version: '2.18.2'

customer-solutions/supply-chain/evrythng.eif/src/main/java/com/evrythng/demo/mq/MQBroker.java

@@ -0,0 +1,19 @@
+package com.evrythng.demo.mq;
+
+import org.apache.activemq.broker.BrokerService;
+
+/**
+ *
+ */
+public class MQBroker implements Runnable {
+    @Override
+    public void run() {
+        BrokerService broker = new BrokerService();
+        try {
+            broker.addConnector("tcp://localhost:61616");
+            broker.start();
+        } catch (Exception e) {
+            throw new RuntimeException("Unable to start ActiveMQ Broker", e);
+        }
+    }
+}

customer-solutions/supply-chain/evrythng.eif/src/main/java/com/evrythng/demo/supplychain/EVTLoader.java

@@ -12,12 +12,16 @@
 public abstract class EVTLoader {
 
     public static ApiManager newApiManager() {
-        try {
-            URL url = new URL(System.getenv("EVT_URL"));
+            URL url = readEVTUrl();
             String EVT_KEY = System.getenv("EVT_KEY");
             ApiConfiguration apiConfiguration = new ApiConfiguration(EVT_KEY);
             apiConfiguration.setUrl(url.toString());
             return new ApiManager(apiConfiguration);
+    }
+
+    private static URL readEVTUrl() {
+        try {
+            return new URL(System.getenv("EVT_URL"));
         } catch (MalformedURLException e) {
             throw new RuntimeException("Malformed URL in ENV variable EVT_URL", e);
         }

customer-solutions/supply-chain/evrythng.eif/src/main/java/com/evrythng/demo/supplychain/XMLProductsLoaderPipeline.java

@@ -1,5 +1,6 @@
 package com.evrythng.demo.supplychain;
 
+import com.evrythng.demo.mq.MQBroker;
 import com.evrythng.demo.supplychain.products.ProductProcessor;
 import com.evrythng.demo.supplychain.products.UnreliableProductLoader;
 import org.apache.camel.builder.RouteBuilder;
@@ -26,27 +27,38 @@ public class XMLProductsLoaderPipeline extends RouteBuilder implements Runnable
     public void configure() throws Exception {
         from("file:src/data")
                 .choice()
-                    .when(xpath("namespace-uri(/*) = 'http://schema.org/Product'"))
+                    .when(xpath(String.format("namespace-uri(/*) = '%s'", Products.ns)))
                         .log("Received XML file containing Products")
-                        .split(Products.ns.xpath(Products.XPATH_PRODUCTS))
+                        .split(Products.namespaces.xpath(Products.XPATH_PRODUCTS))
                         .unmarshal().jaxb(Products.CONTEXT_PATH)
-                        .to(productsXMLQueue)
+                        .to("activemq:products")
                         .endChoice()
                     .otherwise()
                         .log("Ignoring file");
+        from("activemq:products")
+                .throttle(30)
+                .asyncDelayed()
+                .to(productsXMLQueue);
         from(productsXMLQueueConsumer)
                 .process(new ProductProcessor())
                 .process(new UnreliableProductLoader())
-                .errorHandler(deadLetterChannel("seda:errors"));
-        from("seda:errors")
-                .log("ERROR uploading Product to EVT");
+                .errorHandler(deadLetterChannel("activemq:com.evrythng.retry"));
+        from("activemq:com.evrythng.retry")
+                .log("ERROR uploading Product to EVT")
+                .delayer(4 * 1000)
+                .to(productsXMLQueue);
         from("seda:xml-validation")
                 .to("validator:/org/schema/gs1.products.xsd");
 //                .onException(org.xml.sax.SAXParseException.class);
     }
 
     @Override
     public void run() {
+        startMQBroker();
+        startPipeline();
+    }
+
+    private void startPipeline() {
         Main main = new Main();
         main.addRouteBuilder(new XMLProductsLoaderPipeline());
         try {
@@ -56,6 +68,10 @@ public void run() {
         }
     }
 
+    private void startMQBroker() {
+        new Thread(new MQBroker()).start();
+    }
+
     public static void main(String[] args) {
         new XMLProductsLoaderPipeline().run();
     }

customer-solutions/supply-chain/evrythng.eif/src/main/java/org/schema/Products.java

@@ -9,7 +9,9 @@
 @XmlRootElement(name="Products", namespace = "http://schema.org/Product")
 public class Products {
 
-    public static final Namespaces ns = new Namespaces("p", "http://schema.org/Product");
+    public static final String ns = "http://schema.org/Product";
+
+    public static final Namespaces namespaces = new Namespaces("p", ns);
 
     public static final String CONTEXT_PATH = Products.class.getPackage().getName();