Skip to content

Callback mode #3

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

Open
sundbry wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Callback mode #3

Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a18a9bd
Optional callback mode requires acknowledging rabbitmq emssages
Jun 28, 2020
1bbc654
passing tests
Jun 28, 2020
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
23 changes: 18 additions & 5 deletions lib/fluent/plugin/in_amqp.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@ class AMQPInput < Input
config_param :auth_mechanism, :string, default: nil
# milliseconds to delay between messages
config_param :delay, :integer, default: 0
config_param :callback, :bool, default: false

def initialize
super
Expand Down Expand Up @@ -101,13 +102,17 @@ def start
@ack_thread = Thread.new {
@stop = false
loop do
tag = @ack_queue.pop
[tag, success] = @ack_queue.pop
case tag
when :stop
@stop = true
else
begin
@channel.acknowledge(tag, false)
if success
@channel.acknowledge(tag, false)
else
@channel.reject(tag, false)
end
rescue => e
log.error "Failed to acknowledge event", tag: tag, error: e
end
Expand All @@ -121,10 +126,18 @@ def start

q.subscribe(:manual_ack => @manual_ack) do |delivery, meta, msg|
log.debug "Recieved message #{@msg}"
payload = parse_payload(msg)
body = parse_payload(msg)
if @manual_ack && @callback
cb = Proc.new do | success |
@ack_queue.push([delivery.delivery_tag, success])
end
payload = { :payload => payload, :callback => cb }
Copy link

@robot-head robot-head Jun 28, 2020

Choose a reason for hiding this comment

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

confused where the callback is used

Copy link
Author

@sundbry sundbry Jun 28, 2020

Choose a reason for hiding this comment

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

over here: https://github.com/chronicled/fluent-plugin-output-graphql/pull/18

Since we are in the same process should be safe to pass a block (callback) around

obviously this is very implementation specific I would not put this code upstream

else
payload = body
end
router.emit(parse_tag(delivery, meta), parse_time(meta), payload)
if @manual_ack
@ack_queue.push(delivery.delivery_tag)
if @manual_ack && !@callback
@ack_queue.push([delivery.delivery_tag, true])
end
if @delay > 0
sleep(delay / 1000)
Expand Down