Merge 1967720 into 911e08c

repairnator/repairnator-pipeline/pom.xml

@@ -90,6 +90,20 @@
             <version>1.5.0</version>
             <scope>test</scope>
         </dependency>
+        <!-- https://mvnrepository.com/artifact/javax.jms/javax.jms-api -->
+        <dependency>
+            <groupId>javax.jms</groupId>
+            <artifactId>javax.jms-api</artifactId>
+            <version>2.0</version>
+        </dependency>
+
+        <!-- https://mvnrepository.com/artifact/org.apache.activemq/activemq-all -->
+        <!-- https://mvnrepository.com/artifact/org.apache.activemq/activemq-core -->
+        <dependency>
+            <groupId>org.apache.activemq</groupId>
+            <artifactId>activemq-core</artifactId>
+            <version>5.7.0</version>
+        </dependency>
     </dependencies>
 
     <properties>

repairnator/repairnator-pipeline/src/main/java/fr/inria/spirals/repairnator/pipeline/Launcher.java

@@ -1,5 +1,6 @@
 package fr.inria.spirals.repairnator.pipeline;
 
+import static fr.inria.spirals.repairnator.config.RepairnatorConfig.PIPELINE_MODE;
 import ch.qos.logback.classic.Level;
 import com.martiansoftware.jsap.FlaggedOption;
 import com.martiansoftware.jsap.JSAP;
@@ -213,6 +214,26 @@ private JSAP defineArgs() throws JSAPException {
         opt2.setHelp("The ids, names and urls of all experimental pluginrepos used. Must be a list of length n*3 in the order id, name, url, repeat.");
         jsap.registerParameter(opt2);
 
+        opt2 = new FlaggedOption("pipelinemode");
+        opt2.setLongFlag("pipelinemode");
+        opt2.setStringParser(JSAP.STRING_PARSER);
+        opt2.setDefault(PIPELINE_MODE.NOOP.name());
+        opt2.setHelp("Possible string values DOCKER,KUBERNETES,NOOP . DOCKER is for running DockerPipeline, KUBERNETES is for running ActiveMQPipeline and "+PIPELINE_MODE.NOOP.name()+" is for NoopRunner. The last two options do not use docker during run.");
+        jsap.registerParameter(opt2);
+
+        opt2 = new FlaggedOption("activemqurl");
+        opt2.setLongFlag("activemqurl");
+        opt2.setStringParser(JSAP.STRING_PARSER);
+        opt2.setDefault("tcp://localhost:61616");
+        opt2.setHelp("format: 'tcp://IP_OR_DNSNAME:61616', default as 'tcp://localhost:61616'");
+        jsap.registerParameter(opt2);
+
+        opt2 = new FlaggedOption("activemqqueuename");
+        opt2.setLongFlag("activemqqueuename");
+        opt2.setStringParser(JSAP.STRING_PARSER);
+        opt2.setDefault("pipeline");
+        opt2.setHelp("Just a name, default as 'pipeline'");
+        jsap.registerParameter(opt2);
 
         return jsap;
     }
@@ -282,6 +303,10 @@ private void initConfig(JSAPResult arguments) {
         } else {
             this.getConfig().setExperimentalPluginRepoList(null);
         }
+
+        this.getConfig().setPipelineMode(arguments.getString("pipelinemode"));
+        this.getConfig().setActiveMQUrl(arguments.getString("activemqurl"));
+        this.getConfig().setActiveMQQueueName(arguments.getString("activemqqueuename"));
     }
 
     private void checkToolsLoaded(JSAP jsap) {

repairnator/repairnator-pipeline/src/main/java/fr/inria/spirals/repairnator/pipeline/PipelineBuildListener.java

@@ -0,0 +1,77 @@
+package fr.inria.spirals.repairnator.pipeline;
+
+import fr.inria.spirals.repairnator.config.RepairnatorConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import javax.jms.Message;
+
+import javax.jms.Connection;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import javax.jms.MessageListener;
+
+/*For reciever*/
+import javax.jms.MessageConsumer;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+
+/**
+ * This class fetch build ids from ActiveMQ queue and run the pipeline with it.
+ */
+public class PipelineBuildListener implements MessageListener {
+    private static final Logger LOGGER = LoggerFactory.getLogger(PipelineBuildListener.class);
+    private static final RepairnatorConfig config = RepairnatorConfig.getInstance();
+    private Launcher launcher;
+
+    public PipelineBuildListener(Launcher launcher){
+        this.launcher = launcher;
+        this.runAsConsumerServer();
+    }
+
+    /**
+     * Run this as a listener server and fetch one message as a time
+     */
+    public void runAsConsumerServer() {
+        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(config.getActiveMQUrl() + "?jms.prefetchPolicy.all=1");
+        Connection connection;
+        try {
+            connection = connectionFactory.createConnection();
+            connection.start();
+            Session session = connection.createSession(false,Session.CLIENT_ACKNOWLEDGE);
+            Destination queue = session.createQueue(config.getActiveMQQueueName());
+
+            MessageConsumer consumer = session.createConsumer(queue);
+            consumer.setMessageListener(this);
+            LOGGER.warn("Server is now listening for build ids");
+        } catch (JMSException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    /**
+     * Method implemented from MessageListener and is called 
+     * each time this is done with the previous message
+     *
+     * @param message ActiveMQ message object containing a string message.
+     */
+    public void onMessage(Message message) {
+        String messageText = null;
+        try {
+            message.acknowledge();
+            if (message instanceof TextMessage) {
+                TextMessage textMessage = (TextMessage) message;
+                messageText = textMessage.getText();
+
+                LOGGER.info("A new buildId has arrived: " + messageText);
+
+                config.setBuildId(Integer.parseInt(messageText));
+
+                this.launcher.mainProcess();
+            }
+        } catch (JMSException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}

repairnator/repairnator-realtime/src/main/java/fr/inria/spirals/repairnator/realtime/ActiveMQPipelineRunner.java

@@ -11,6 +11,7 @@
 import javax.jms.MessageProducer;
 import javax.jms.Session;
 import javax.jms.TextMessage;
+import javax.jms.MessageListener;
 
 /*For reciever*/
 import javax.jms.Message;
@@ -19,12 +20,13 @@
 import org.apache.activemq.ActiveMQConnectionFactory;
 import fr.inria.spirals.repairnator.config.RepairnatorConfig;
 
+import java.util.concurrent.TimeUnit;
 /**
  * This class will take the builds qualified by the inspectBuild
  * and submit to an ActiveMQ queue for the repairnator-worker
  * to run on Kubernetes.
  */
-public class ActiveMQPipelineRunner implements PipelineRunner  {
+public class ActiveMQPipelineRunner implements PipelineRunner,MessageListener {
     private static final Logger LOGGER = LoggerFactory.getLogger(ActiveMQPipelineRunner.class);
     private static final int DELAY_BETWEEN_DOCKER_IMAGE_REFRESH = 60; // in minutes
     private static final RepairnatorConfig config = RepairnatorConfig.getInstance();
@@ -71,6 +73,10 @@ public Boolean testConnection() {
         }
     }
 
+    /**
+     * Given a build object, produce am ActiveMQ message and send to queue
+     * @param build object containing a build id.
+     */
     public void submitBuild(Build build) {
         try {
             /*
@@ -112,6 +118,11 @@ public void initRunner() {
         // so far, nothing to set up the connection
     }
 
+    /**
+     * Fetch one message from queue and done
+     * 
+     * @return a String message
+     */
     public String receiveBuildFromQueue() {
         try {
             // Create a ConnectionFactory
@@ -144,4 +155,49 @@ public String receiveBuildFromQueue() {
             throw new RuntimeException(e);
         }
     }
+
+    /**
+     * Run this as a listener server and fetch one message as a time
+     */
+    public void runAsConsumerServer() {
+        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(config.getActiveMQUrl() + "?jms.prefetchPolicy.all=1");
+        Connection connection;
+        try {
+            connection = connectionFactory.createConnection();
+            connection.start();
+            Session session = connection.createSession(false,Session.CLIENT_ACKNOWLEDGE);
+            Destination queue = session.createQueue(config.getActiveMQQueueName());
+
+            MessageConsumer responseConsumer = session.createConsumer(queue);
+            responseConsumer.setMessageListener(this);
+            LOGGER.warn("Listening");
+        } catch (JMSException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    /**
+     * Method implemented from MessageListener and is called 
+     * each time this is done with the previous message
+     *
+     * @param message ActiveMQ message object containing a string message.
+     */
+    public void onMessage(Message message) {
+        String messageText = null;
+        try {
+            message.acknowledge();
+            try {
+                TimeUnit.SECONDS.sleep(10);
+            } catch(java.lang.InterruptedException e) {
+
+            }
+            if (message instanceof TextMessage) {
+                TextMessage textMessage = (TextMessage) message;
+                messageText = textMessage.getText();
+                System.out.println("messageText = " + messageText);
+            }
+        } catch (JMSException e) {
+            throw new RuntimeException(e);
+        }
+    }
 }