-
-
Notifications
You must be signed in to change notification settings - Fork 42
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
Add the 'basicCancel' method to the AMQPClient #134
Conversation
Codecov Report
@@ Coverage Diff @@
## master #134 +/- ##
==========================================
+ Coverage 97.43% 97.47% +0.04%
==========================================
Files 13 13
Lines 117 119 +2
==========================================
+ Hits 114 116 +2
Misses 3 3
Continue to review full report at Codecov.
|
core/src/main/scala/com/github/gvolpe/fs2rabbit/algebra/AMQPClient.scala
Outdated
Show resolved
Hide resolved
@gvolpe the 'basicConsume' method can return G[_] as well. I've added resource management in a scope of this change. |
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.
Looks good so far!
By introducing Stream.bracket(consumerF)(tag => AMQP.basicCancel(channel, tag))
I wonder whether a consumer will re-connect in case a queue gets deleted and by the mechanism of the ResilientStream
it gets re-declared.
Can you verify that these statements hold true?
- Run the
IOAckerConsumer
example. - Go to the RabbitMQ console and delete
testQ
. - You see the following output:
Consumed: AmqpEnvelope(DeliveryTag(1),Hey!,AmqpProperties(None,None,None,None,None,None,None,None,None,None,Map(app -> StringVal(fs2RabbitDemo), demoId -> LongVal(123))))
Consumed: AmqpEnvelope(DeliveryTag(2),{"id":1,"name":"Sherlock","address":{"number":212,"streetName":"Baker St"}},AmqpProperties(None,None,None,None,None,None,None,None,None,None,Map()))
11:06:10.790 [scala-execution-context-global-20] INFO c.g.gvolpe.fs2rabbit.effects.Log$ - Releasing connection: amqp://guest@127.0.0.1:5672/ previously acquired.
11:06:10.806 [scala-execution-context-global-20] ERROR c.g.gvolpe.fs2rabbit.effects.Log$ - Queue might have been DELETED! amq.ctag-EJLX41gM42B3bKMJFJGlbg
java.lang.Exception: Queue might have been DELETED! amq.ctag-EJLX41gM42B3bKMJFJGlbg
at com.github.gvolpe.fs2rabbit.interpreter.AMQPClientStream$$anon$1.$anonfun$handleCancel$2(AMQPClientStream.scala:45) ~[classes/:na]
at com.github.gvolpe.fs2rabbit.interpreter.AMQPClientStream$$anon$1.$anonfun$handleCancel$2$adapted(AMQPClientStream.scala:43) ~[classes/:na]
at scala.Option.fold(Option.scala:158) ~[scala-library-2.12.7.jar:na]
at com.github.gvolpe.fs2rabbit.interpreter.AMQPClientStream$$anon$1.handleCancel(AMQPClientStream.scala:43) ~[classes/:na]
at com.rabbitmq.client.impl.ConsumerDispatcher$3.run(ConsumerDispatcher.java:115) ~[amqp-client-5.5.0.jar:5.5.0]
at com.rabbitmq.client.impl.ConsumerWorkService$WorkPoolRunnable.run(ConsumerWorkService.java:104) ~[amqp-client-5.5.0.jar:5.5.0]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[na:1.8.0_161]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[na:1.8.0_161]
at java.lang.Thread.run(Thread.java:748) ~[na:1.8.0_161]
11:06:10.806 [scala-execution-context-global-20] INFO c.g.gvolpe.fs2rabbit.effects.Log$ - Restarting in 5 seconds...
Consumed: AmqpEnvelope(DeliveryTag(1),Hey!,AmqpProperties(None,None,None,None,None,None,None,None,None,None,Map(app -> StringVal(fs2RabbitDemo), demoId -> LongVal(123))))
Consumed: AmqpEnvelope(DeliveryTag(2),{"id":1,"name":"Sherlock","address":{"number":212,"streetName":"Baker St"}},AmqpProperties(None,None,None,None,None,None,None,None,None,None,Map()))
core/src/main/scala/com/github/gvolpe/fs2rabbit/interpreter/AMQPClientStream.scala
Outdated
Show resolved
Hide resolved
My output:
The only difference that there are two exceptions:
But the queue was recreated and the message was published and consumed successfully. |
Great, I think that's the expected behavior. Once you sync with master we can get it merged, thanks! |
…re/basic-cancel # Conflicts: # core/src/main/scala/com/github/gvolpe/fs2rabbit/algebra/AckConsuming.scala # core/src/main/scala/com/github/gvolpe/fs2rabbit/interpreter/AMQPClientStream.scala # core/src/main/scala/com/github/gvolpe/fs2rabbit/program/ConsumerProgram.scala
# Conflicts: # core/src/main/scala/com/github/gvolpe/fs2rabbit/algebra/AMQPClient.scala # core/src/main/scala/com/github/gvolpe/fs2rabbit/interpreter/AMQPClientStream.scala # core/src/main/scala/com/github/gvolpe/fs2rabbit/program/ConsumingProgram.scala # core/src/test/scala/com/github/gvolpe/fs2rabbit/interpreter/AMQPClientInMemory.scala # core/src/test/scala/com/github/gvolpe/fs2rabbit/interpreter/Fs2RabbitSpec.scala
Hi @gvolpe, I solved all conflicts.
|
A consumer can be canceled even if the connection is still open. So, I have added this missing method.