Add support for username and password in ActiveMQ connection

Author: albanf

README.md

@@ -41,4 +41,6 @@ After that step is completed, restart your server and you should have the new in
 |--------------------|--------------------------------------------------------------------------------------------------------------|-------------------------------------------------------------------------------------|
 | brokerUrl          | failover:(ssl://activemq-1.example.com:61616,ssl://activemq-2.example.com:61616)?randomize=false&backup=true | The ActiveMQ client URL. Any valid ActiveMQ client URL can be used.                 |
 | queueName          | ch.qos.logback                                                                                               | The Queue name to read logs off of.                                                 |
+| username           | loguser                                                                                                      | The username to use to establish the connection to ActiveMQ.                        |
+| password           | myPassw0rd                                                                                                   | The password to use to establish the connection to ActiveMQ.                        |
 | throttling_allowed | true                                                                                                         | If the graylog server gets busy, slow down this input. Recommend checking this box. |

src/main/java/com/github/exabrial/graylog/OpenwireConsumer.java

@@ -24,6 +24,8 @@
 
 public class OpenwireConsumer {
 	private final String brokerUrl;
+	private final String username;
+	private final String password;
 	private final String queue;
 	private final MessageInput sourceInput;
 	private final OpenwireTransport openwireTransport;
@@ -33,9 +35,11 @@ public class OpenwireConsumer {
 	private AtomicLong lastSecBytesReadTmp = new AtomicLong(0);
 	private Connection connection;
 
-	OpenwireConsumer(String brokerUrl, String queue, MessageInput sourceInput, ScheduledExecutorService scheduler,
-			OpenwireTransport openwireTransport) {
+	OpenwireConsumer(String brokerUrl, String username, String password, String queue, MessageInput sourceInput,
+			ScheduledExecutorService scheduler, OpenwireTransport openwireTransport) {
 		this.brokerUrl = brokerUrl;
+		this.username = username;
+		this.password = password;
 		this.queue = queue;
 		this.sourceInput = sourceInput;
 		this.openwireTransport = openwireTransport;
@@ -88,7 +92,12 @@ public void onMessage(Message message) {
 	}
 
 	public void connect() throws JMSException {
-		ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(brokerUrl);
+		ActiveMQConnectionFactory connectionFactory;
+		if (username != null && !"".equals(username)) {
+			connectionFactory = new ActiveMQConnectionFactory(username, password, brokerUrl);
+		} else {
+			connectionFactory = new ActiveMQConnectionFactory(brokerUrl);
+		}
 		connection = connectionFactory.createConnection();
 		connection.start();
 		connection.setExceptionListener(new ExceptionListener() {

src/main/java/com/github/exabrial/graylog/OpenwireTransport.java

@@ -30,8 +30,16 @@
 import com.google.inject.assistedinject.AssistedInject;
 
 public class OpenwireTransport extends ThrottleableTransport {
-	public static final String brokerUrl = "ActiveMQ Broker URL";
-	public static final String queueName = "Queue Name";
+	private static final String BROKER_URL_CONFIGURATION_KEY = "brokerUrl";
+	private static final String USERNAME_CONFIGURATION_KEY = "username";
+	private static final String PASSWORD_CONFIGURATION_KEY = "password";
+	private static final String QUEUE_NAME_CONFIGURATION_KEY = "queueName";
+
+	private static final String BROKER_URL_CONFIGURATION_HUMAN_NAME = "ActiveMQ Broker URL";
+	private static final String QUEUE_NAME_CONFIGURATION_HUMAN_NAME = "Queue Name";
+	private static final String USERNAME_CONFIGURATION_HUMAN_NAME = "Username";
+	private static final String PASSWORD_CONFIGURATION_HUMAN_NAME = "Password";
+
 	private static final Logger log = LoggerFactory.getLogger(OpenwireTransport.class);
 
 	private final Configuration configuration;
@@ -115,7 +123,10 @@ public void setMessageAggregator(final CodecAggregator aggregator) {
 
 	@Override
 	public void doLaunch(final MessageInput input) throws MisfireException {
-		consumer = new OpenwireConsumer(configuration.getString("brokerUrl"), configuration.getString("queueName"), input, scheduler,
+		consumer = new OpenwireConsumer(configuration.getString(BROKER_URL_CONFIGURATION_KEY),
+				configuration.getString(USERNAME_CONFIGURATION_KEY),
+				configuration.getString(PASSWORD_CONFIGURATION_KEY),
+				configuration.getString(QUEUE_NAME_CONFIGURATION_KEY), input, scheduler,
 				this);
 		eventBus.register(this);
 		try {
@@ -157,11 +168,16 @@ public static class Config extends ThrottleableTransport.Config {
 		@Override
 		public ConfigurationRequest getRequestedConfiguration() {
 			final ConfigurationRequest cr = super.getRequestedConfiguration();
-			cr.addField(new TextField("brokerUrl", brokerUrl, defaultBrokerUrl(),
+			cr.addField(new TextField(BROKER_URL_CONFIGURATION_KEY, BROKER_URL_CONFIGURATION_HUMAN_NAME, defaultBrokerUrl(),
 					"ActiveMQ Broker URL to connect to; Reference the ActiveMQ documentation for help",
 					ConfigurationField.Optional.NOT_OPTIONAL));
-			cr.addField(new TextField("queueName", queueName, defaultQueueName(), "Name of queue to listen on",
+			cr.addField(new TextField(QUEUE_NAME_CONFIGURATION_KEY, QUEUE_NAME_CONFIGURATION_HUMAN_NAME, defaultQueueName(), "Name of queue to listen on",
 					ConfigurationField.Optional.NOT_OPTIONAL));
+			cr.addField(new TextField(USERNAME_CONFIGURATION_KEY, USERNAME_CONFIGURATION_HUMAN_NAME, null,
+					"username for ActiveMQ broker; Reference the ActiveMQ documentation for help",
+					ConfigurationField.Optional.OPTIONAL));
+			cr.addField(new TextField(PASSWORD_CONFIGURATION_KEY, PASSWORD_CONFIGURATION_HUMAN_NAME, null, "password for ActiveMQ broker; Reference the ActiveMQ documentation for help",
+					ConfigurationField.Optional.OPTIONAL));
 			return cr;
 		}
 
@@ -170,7 +186,7 @@ protected String defaultBrokerUrl() {
 		}
 
 		protected String defaultQueueName() {
-			return "com.example.logback";
+			return "ch.qos.logback";
 		}
 	}
 }