-
Notifications
You must be signed in to change notification settings - Fork 35
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.
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
Consumer arg support for shovels #578
Conversation
4926557
to
23b30c1
Compare
src/lavinmq/shovel/shovel.cr
Outdated
@@ -41,7 +41,8 @@ module LavinMQ | |||
|
|||
def initialize(@name : String, @uri : URI, @queue : String?, @exchange : String? = nil, | |||
@exchange_key : String? = nil, | |||
@delete_after = DEFAULT_DELETE_AFTER, @prefetch = DEFAULT_PREFETCH, @ack_mode = DEFAULT_ACK_MODE, | |||
@delete_after = DEFAULT_DELETE_AFTER, @prefetch = DEFAULT_PREFETCH, | |||
@ack_mode = DEFAULT_ACK_MODE, @consumer_args = {} of String => JSON::Any || nil, |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Any reason to have @consumer_args
as a instance variable? Looks like it's only used to populate @args
?
And the || nil
part seems odd. Will it ever do anything? Default value is empty hash. Or do you want @consumer_args : Hash(String, JSON::Any)? = nil
?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yeah, looks much better that way 👍
src/lavinmq/shovel/shovel.cr
Outdated
@@ -60,6 +61,10 @@ module LavinMQ | |||
if @queue.nil? && @exchange.nil? | |||
raise ArgumentError.new("Shovel source requires a queue or an exchange") | |||
end | |||
@args = AMQ::Protocol::Table.new | |||
@consumer_args.try &.each do |k, v| | |||
@args[k] = v |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
How will JSON::Any
be handled from here? Should it be v.as_s?
or something?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I don't think v.as_s?
is needed for it to work, but it's probably best to be specific
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Oh, AMQ::Protocol::Table
accepts JSON::Any
as value. Didn't know that (and I guess it compiled for you 🤦). It should be fine then!
Moved the definitions upload stuff to a separate PR to keep this PR more focused |
40044d2
to
3b3b9df
Compare
WHAT is this pull request doing?
Shovels handle
src-consumer-args
to enable them to handle stream queues.HOW can this pull request be tested?
Create a shovel for a stream queue via API with x-stream-offset