Test for connection lost
@@ -21,12 +21,15 @@ use googapis::google::pubsub::v1::subscriber_client::SubscriberClient;
use googapis::google::pubsub::v1::{PullRequest, Subscription, Topic};
use serial_test::serial;
use std::collections::HashSet;
+use std::time::Duration;
+use async_std::prelude::FutureExt;
use testcontainers::clients::Cli;
use testcontainers::RunnableImage;
use tonic::transport::Channel;
use tremor_common::ports::IN;
use tremor_pipeline::{Event, EventId};
use tremor_value::{literal, Value};
+use crate::instance::State;
#[async_std::test]
#[serial(gpubsub)]
@@ -208,3 +211,82 @@ async fn simple_publish() -> Result<()> {
Ok(())
}
+
+#[async_std::test]
+#[serial(gpubsub)]
+async fn simple_publish_with_timeout() -> Result<()> {
+ let _ = env_logger::try_init();
+
+ let runner = Cli::docker();
+
+ let (pubsub, pubsub_args) =
+ testcontainers::images::google_cloud_sdk_emulators::CloudSdk::pubsub();
+ let runnable_image = RunnableImage::from((pubsub, pubsub_args));
+ let container = runner.run(runnable_image);
+
+ let port =
+ container.get_host_port(testcontainers::images::google_cloud_sdk_emulators::PUBSUB_PORT);
+ let endpoint = format!("http://localhost:{}", port);
+ let endpoint_clone = endpoint.clone();
+
+ let connector_yaml: Value = literal!({
+ "codec": "binary",
+ "config":{
+ "endpoint": endpoint,
+ "connect_timeout": 30000000000u64,
+ "topic": "projects/test/topics/test",
+ "skip_authentication": true
+ }
+ });
+
+ let channel = Channel::from_shared(endpoint_clone)?.connect().await?;
+ let mut subscriber = SubscriberClient::new(channel.clone());
+ let mut publisher = PublisherClient::new(channel.clone());
+ publisher
+ .create_topic(Topic {
+ name: "projects/test/topics/test".to_string(),
+ labels: Default::default(),
+ message_storage_policy: None,
+ kms_key_name: "".to_string(),
+ schema_settings: None,
+ satisfies_pzs: false,
+ message_retention_duration: None,
+ })
+ .await?;
+ subscriber
+ .create_subscription(Subscription {
+ name: "projects/test/subscriptions/test-subscription-a".to_string(),
+ topic: "projects/test/topics/test".to_string(),
+ push_config: None,
+ ack_deadline_seconds: 0,
+ retain_acked_messages: false,
+ message_retention_duration: None,
+ labels: Default::default(),
+ enable_message_ordering: false,
+ expiration_policy: None,
+ filter: "".to_string(),
+ dead_letter_policy: None,
+ retry_policy: None,
+ detached: false,
+ topic_message_retention_duration: None,
+ })
+ .await?;
+
+ let harness =
+ ConnectorHarness::new(function_name!(), &Builder::default(), &connector_yaml).await?;
+ harness.start().await?;
+ harness.wait_for_connected().await?;
+ harness.consume_initial_sink_contraflow().await?;
+
+ drop(container);
+
+ let event = Event {
+ id: EventId::default(),
+ data: (Value::String(format!("Event X").into()), literal!({})).into(),
+ ..Event::default()
+ };
+ harness.send_to_sink(event, IN).await?;
+ harness.wait_for_state(State::Failed).timeout(Duration::from_secs(10)).await?.unwrap();
+
+ Ok(())
+}
