Merge pull request #3 from nexylan/1-add-the-possibility-to-define-if-it-s-a-permanent-queue-or-not

Add durability management

Author: henri9813

dist/graylog-rabbitmq-1.2.1.jar dist/graylog-rabbitmq-1.3.0.jar

@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>net.nexylan.graylog.plugins</groupId>
     <artifactId>graylog-rabbitmq</artifactId>
-    <version>1.2.1</version>
+    <version>1.3.0</version>
     <packaging>jar</packaging>
 
     <name>${project.artifactId}</name>

pom.xml

@@ -3,10 +3,7 @@
 import org.graylog2.plugin.Message;
 import org.graylog2.plugin.configuration.Configuration;
 import org.graylog2.plugin.configuration.ConfigurationRequest;
-import org.graylog2.plugin.configuration.fields.ConfigurationField;
-import org.graylog2.plugin.configuration.fields.DropdownField;
-import org.graylog2.plugin.configuration.fields.NumberField;
-import org.graylog2.plugin.configuration.fields.TextField;
+import org.graylog2.plugin.configuration.fields.*;
 import org.graylog2.plugin.inputs.annotations.ConfigClass;
 import org.graylog2.plugin.inputs.annotations.FactoryClass;
 import org.graylog2.plugin.outputs.MessageOutput;
@@ -32,6 +29,7 @@ public class RabbitMq implements MessageOutput{
     private static final String RABBIT_USER = "rabbit_user";
     private static final String RABBIT_PASSWORD = "rabbit_password";
     private static final String RABBIT_TTL = "rabbit_ttl";
+    private static final String RABBIT_DURABLE = "rabbit_durable";
 
     private boolean running;
 
@@ -46,7 +44,13 @@ public RabbitMq(@Assisted Configuration configuration) throws MessageOutputConfi
 
         // Set up sender
         sender = new RabbitMQSender(
-                configuration.getString(RABBIT_HOST), configuration.getInt(RABBIT_PORT), configuration.getString(RABBIT_QUEUE), configuration.getString(RABBIT_USER), configuration.getString(RABBIT_PASSWORD), configuration.getInt(RABBIT_TTL)
+                configuration.getString(RABBIT_HOST),
+                configuration.getInt(RABBIT_PORT),
+                configuration.getString(RABBIT_QUEUE),
+                configuration.getString(RABBIT_USER),
+                configuration.getString(RABBIT_PASSWORD),
+                configuration.getInt(RABBIT_TTL),
+                configuration.getBoolean(RABBIT_DURABLE)
         );
 
         running = true;
@@ -145,6 +149,12 @@ public ConfigurationRequest getRequestedConfiguration() {
                     ConfigurationField.Optional.NOT_OPTIONAL)
             );
 
+            configurationRequest.addField(new BooleanField(RABBIT_DURABLE,
+                    "RabbitMQ Durable",
+                    true,
+                    "May this queue must be durable ?"
+            ));
+
             return configurationRequest;
         }
     }

src/main/java/net/nexylan/graylog/rabbitmq/RabbitMq.java

@@ -21,6 +21,7 @@ public class RabbitMQSender implements Sender {
     private String user;
     private String password;
     private int ttl;
+    private boolean durable;
 
 
     //RabbitMQ objects
@@ -34,14 +35,15 @@ public class RabbitMQSender implements Sender {
 
     private boolean is_initialized = false;
 
-    public RabbitMQSender(String host, int port, String queue, String user, String password, int ttl)
+    public RabbitMQSender(String host, int port, String queue, String user, String password, int ttl, boolean durable)
     {
         this.host = host;
         this.port = port;
         this.queue = queue;
         this.user = user;
         this.password = password;
         this.ttl = ttl;
+        this.durable = durable;
         initialize();
     }
 
@@ -79,7 +81,7 @@ public void initialize()
         }
 
         try {
-            this.channel.queueDeclare(this.queue, false, false, false, null);
+            this.channel.queueDeclare(this.queue, false, this.durable, false, null);
             LOG.info("[RabbitMQ] The queue have been successfully created.");
         } catch (IOException e) {
             LOG.error("[RabbitMQ] Impossible to declare the queue.");