Bugfix: shovel sources reconnect (#758)

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- Shovel AMQP source didn't reconnect on network failures (partially fixed in prev release) [#758](https://github.com/cloudamqp/lavinmq/pull/758)
+
 ## [2.0.0-rc.4] - 2024-08-21
 
 ### Fixed

shard.lock

@@ -10,7 +10,7 @@ shards:
 
   amqp-client:
     git: https://github.com/cloudamqp/amqp-client.cr.git
-    version: 1.2.5
+    version: 1.2.6
 
   lz4:
     git: https://github.com/84codes/lz4.cr.git

spec/server_spec.cr

@@ -80,6 +80,7 @@ describe LavinMQ::Server do
             msg.reject(requeue: false)
             done.send nil
           end
+        rescue AMQP::Client::Connection::ClosedException
         end
       end
       timeout = false

spec/shovel_spec.cr

@@ -24,24 +24,15 @@ describe LavinMQ::Shovel do
             direct_user: s.users.direct_user
           )
 
-          wg = WaitGroup.new(1)
-          exception : Exception? = nil
-          spawn do
-            source.start
-            wg.done
+          source.start
+
+          s.vhosts["/"].queues["source"].publish(LavinMQ::Message.new("", "", ""))
+          expect_raises(AMQP::Client::Connection::ClosedException) do
             source.each do
+              s.vhosts["/"].connections.each &.close("spec")
             end
-          rescue ex
-            exception = ex
-          ensure
-            wg.done
           end
-          wg.wait # wait for source to start
-          wg.add 1
-          s.vhosts["/"].connections.each &.close("spec")
-          wg.wait # wait for exception to be rescued
           source.started?.should be_false
-          exception.should be_a AMQP::Client::Connection::ClosedException
         end
       end
     end