Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Add option to AMQP transports to bind the queue to the exchange #1633

Merged
merged 1 commit into from
Dec 15, 2015
Merged

Add option to AMQP transports to bind the queue to the exchange #1633

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion graylog2-inputs/src/main/java/org/graylog2/inputs/transports/AmqpConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@ public class AmqpConsumer {

private final String queue;
private final String exchange;
private final boolean exchangeBind;
private final String routingKey;
private final boolean requeueInvalid;

Expand All @@ -71,7 +72,7 @@ public class AmqpConsumer {
private AtomicLong lastSecBytesReadTmp = new AtomicLong(0);

public AmqpConsumer(String hostname, int port, String virtualHost, String username, String password,
int prefetchCount, String queue, String exchange, String routingKey, int parallelQueues,
int prefetchCount, String queue, String exchange, boolean exchangeBind, String routingKey,int parallelQueues,
boolean tls, boolean requeueInvalid, int heartbeatTimeout, MessageInput sourceInput,
ScheduledExecutorService scheduler, AmqpTransport amqpTransport) {
this.hostname = hostname;
Expand All @@ -83,6 +84,7 @@ public AmqpConsumer(String hostname, int port, String virtualHost, String userna

this.queue = queue;
this.exchange = exchange;
this.exchangeBind = exchangeBind;
this.routingKey = routingKey;
this.heartbeatTimeout = heartbeatTimeout;

Expand All @@ -108,6 +110,9 @@ public void run() throws IOException {
for (int i = 0; i < parallelQueues; i++) {
final String queueName = String.format(queue, i);
channel.queueDeclare(queueName, true, false, false, null);
if (exchangeBind) {
channel.queueBind(queueName, exchange, routingKey);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we check whether the exchange exists and issue a warning if it doesn't?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If the input cannot be started due to a missing exchange, there is a warning in the web interface that is enough to debug the issue. It actually says that the exchange does not exist.

}
channel.basicConsume(queueName, false, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
Expand Down
11 changes: 11 additions & 0 deletions graylog2-inputs/src/main/java/org/graylog2/inputs/transports/AmqpTransport.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,7 @@ public class AmqpTransport extends ThrottleableTransport {
public static final String CK_PASSWORD = "broker_password";
public static final String CK_PREFETCH = "prefetch";
public static final String CK_EXCHANGE = "exchange";
public static final String CK_EXCHANGE_BIND = "exchange_bind";
public static final String CK_QUEUE = "queue";
public static final String CK_ROUTING_KEY = "routing_key";
public static final String CK_PARALLEL_QUEUES = "parallel_queues";
Expand Down Expand Up @@ -160,6 +161,7 @@ public void doLaunch(MessageInput input) throws MisfireException {
configuration.getInt(CK_PREFETCH),
configuration.getString(CK_QUEUE),
configuration.getString(CK_EXCHANGE),
configuration.getBoolean(CK_EXCHANGE_BIND),
configuration.getString(CK_ROUTING_KEY),
configuration.getInt(CK_PARALLEL_QUEUES),
configuration.getBoolean(CK_TLS),
Expand Down Expand Up @@ -291,6 +293,15 @@ public ConfigurationRequest getRequestedConfiguration() {
)
);

cr.addField(
new BooleanField(
CK_EXCHANGE_BIND,
"Bind to exchange",
false,
"Binds the queue to the configured exchange. The exchange must already exist."
)
);

cr.addField(
new TextField(
CK_ROUTING_KEY,
Expand Down