Update integration tests

Author: aramperes

src/sinks/amqp/channel.rs

@@ -1,12 +1,13 @@
+use super::config::AmqpSinkConfig;
 use super::service::AmqpError;
 use crate::amqp::AmqpConfig;
 use deadpool::managed::Pool;
 use lapin::options::ConfirmSelectOptions;
 
 pub type AmqpSinkChannels = Pool<AmqpSinkChannelManager>;
 
-pub(super) fn new_channel_pool(config: &AmqpConfig) -> crate::Result<AmqpSinkChannels> {
-    let channel_manager = AmqpSinkChannelManager::new(config);
+pub(super) fn new_channel_pool(config: &AmqpSinkConfig) -> crate::Result<AmqpSinkChannels> {
+    let channel_manager = AmqpSinkChannelManager::new(&config.connection);
     let channels = Pool::builder(channel_manager)
         .max_size(4)
         .runtime(deadpool::Runtime::Tokio1)

src/sinks/amqp/integration_tests.rs

@@ -3,7 +3,7 @@ use crate::{
     amqp::await_connection,
     config::{SinkConfig, SinkContext},
     shutdown::ShutdownSignal,
-    sinks::amqp::channel::AmqpChannel,
+    sinks::amqp::channel::new_channel_pool,
     template::{Template, UnsignedIntTemplate},
     test_util::{
         components::{run_and_assert_sink_compliance, SINK_TAGS},
@@ -13,7 +13,7 @@ use crate::{
 };
 use config::AmqpPropertiesConfig;
 use futures::StreamExt;
-use std::{collections::HashSet, sync::Arc, time::Duration};
+use std::{collections::HashSet, time::Duration};
 use vector_lib::{config::LogNamespace, event::LogEvent};
 
 pub fn make_config() -> AmqpSinkConfig {
@@ -38,8 +38,8 @@ async fn healthcheck() {
     let mut config = make_config();
     config.exchange = Template::try_from(exchange.as_str()).unwrap();
     await_connection(&config.connection).await;
-    let channel = AmqpChannel::new(&config.connection).await.unwrap();
-    super::config::healthcheck(Arc::new(channel)).await.unwrap();
+    let channels = new_channel_pool(&config).unwrap();
+    super::config::healthcheck(channels).await.unwrap();
 }
 
 #[tokio::test]

src/sinks/amqp/sink.rs

@@ -37,11 +37,8 @@ pub(super) struct AmqpSink {
 
 impl AmqpSink {
     pub(super) async fn new(config: AmqpSinkConfig) -> crate::Result<Self> {
-        let channels = super::channel::new_channel_pool(&config.connection).map_err(|e| {
-            BuildError::AmqpCreateFailed {
-                source: e,
-            }
-        })?;
+        let channels = super::channel::new_channel_pool(&config)
+            .map_err(|e| BuildError::AmqpCreateFailed { source: e })?;
 
         let transformer = config.encoding.transformer();
         let serializer = config.encoding.build()?;