-
Notifications
You must be signed in to change notification settings - Fork 29
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Implemented GRAILSPLUGINS-2525: exchanges can now be declared in the …
…Rabbit configuration and queues bound to them.
- Loading branch information
Showing
4 changed files
with
430 additions
and
17 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,23 +1,111 @@ | ||
| package org.grails.rabbitmq | ||
|
|
||
| import org.springframework.amqp.core.Queue | ||
| import org.springframework.amqp.core.* | ||
| import org.slf4j.LoggerFactory | ||
|
|
||
| class RabbitQueueBuilder { | ||
|
|
||
| private final static log = LoggerFactory.getLogger(RabbitQueueBuilder) | ||
|
|
||
| def bindings = [] | ||
| def exchanges = [] | ||
| def queues = [] | ||
|
|
||
| private currentExchange | ||
|
|
||
| def methodMissing(String methodName, args) { | ||
| def queue = new Queue(methodName) | ||
|
|
||
| if(args) { | ||
| def argsMap = args[0] | ||
| def argsMap = args ? args[0] : [:] | ||
| if(argsMap) { | ||
| queue.autoDelete = Boolean.valueOf(argsMap.autoDelete) | ||
| queue.exclusive = Boolean.valueOf(argsMap.exclusive) | ||
| queue.durable = Boolean.valueOf(argsMap.durable) | ||
| if(argsMap.arguments instanceof Map) { | ||
| queue.arguments = argsMap.arguments | ||
| } | ||
| } | ||
|
|
||
| // If we are nested inside of an exchange definition, create | ||
| // a binding between the queue and the exchange. | ||
| if (currentExchange) { | ||
| def newBinding = [ queue: methodName, exchange: currentExchange.name ] | ||
| bindings << newBinding | ||
|
|
||
| switch (currentExchange.type) { | ||
| case DirectExchange: | ||
| if (argsMap.binding && !(argsMap.binding instanceof CharSequence)) { | ||
| throw new RuntimeException( | ||
| "The binding for queue '${methodName}' to direct " + | ||
| "exchange '${currentExchange.name}' must be a string.") | ||
| } | ||
|
|
||
| // Use the name of the queue as a default binding if no | ||
| // explicit one is declared. | ||
| newBinding.rule = argsMap.binding ?: queue.name | ||
| break | ||
|
|
||
| case FanoutExchange: | ||
| // Any binding will be ignored. | ||
| log.warn "'${currentExchange.name}' is a fanout exchange - binding for queue '${methodName}' ignored" | ||
| break | ||
|
|
||
| /* Enable once we have a version of Spring AMQP with HeadersExchange included | ||
| case HeadersExchange: | ||
| if (!(argsMap.binding instanceof Map)) { | ||
| throw new RuntimeException( | ||
| "The binding for queue '${methodName}' to headers " + | ||
| "exchange '${currentExchange.name}' must be declared " + | ||
| "and must be a map.") | ||
| } | ||
| newBinding.rule = argsMap.binding | ||
| break | ||
| */ | ||
|
|
||
| case TopicExchange: | ||
| if (!(argsMap.binding instanceof CharSequence)) { | ||
| throw new RuntimeException( | ||
| "The binding for queue '${methodName}' to topic " + | ||
| "exchange '${currentExchange.name}' must be declared " + | ||
| "and must be a string.") | ||
| } | ||
|
|
||
| newBinding.rule = argsMap.binding | ||
| break | ||
| } | ||
| } | ||
| queues << queue | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Defines a new exchange. | ||
| * @param args The properties of the exchange, such as its name | ||
| * and whether it is durable or not. | ||
| * @param c An optional closure that includes queue definitions. | ||
| * If provided, the queues are bound to this exchange. | ||
| */ | ||
| def exchange(Map args, Closure c = null) { | ||
| if (currentExchange) throw new RuntimeException("Cannot declare an exchange within another exchange") | ||
| if (!args.name) throw new RuntimeException("A name must be provided for the exchange") | ||
| if (!args.type) throw new RuntimeException("A type must be provided for the exchange '${args.name}'") | ||
| exchanges << [ *:args ] | ||
|
|
||
| if (c) { | ||
| currentExchange = exchanges[-1] | ||
| c = c.clone() | ||
| c.delegate = this | ||
| c() | ||
| } | ||
|
|
||
| // Clear the current exchange regardless of whether there was | ||
| // a closure argument or not. Just an extra safety measure. | ||
| currentExchange = null | ||
| } | ||
|
|
||
| def getDirect() { return DirectExchange } | ||
| def getFanout() { return FanoutExchange } | ||
| def getHeaders() { return HeadersExchange } | ||
| def getTopic() { return TopicExchange } | ||
| } |
Oops, something went wrong.