Enable listen command with pull event source

crates/relayer-cli/src/commands/listen.rs

@@ -17,7 +17,7 @@ use tracing::{error, info, instrument};
 use ibc_relayer::{
     chain::handle::Subscription,
     config::{ChainConfig, EventSourceMode},
-    event::source::websocket::EventSource,
+    event::source::EventSource,
 };
 use ibc_relayer_types::{core::ics24_host::identifier::ChainId, events::IbcEvent};
 
@@ -144,29 +144,25 @@ fn subscribe(
     compat_mode: CompatMode,
     rt: Arc<TokioRuntime>,
 ) -> eyre::Result<Subscription> {
-    let EventSourceMode::Push { url, batch_delay } = &chain_config.event_source else {
-        return Err(eyre!("unsupported event source mode, only 'push' is supported for listening to events"));
-    };
-
-    let (mut event_source, tx_cmd) = EventSource::new(
-        chain_config.id.clone(),
-        url.clone(),
-        compat_mode,
-        *batch_delay,
-        rt,
-    )
-    .map_err(|e| eyre!("could not initialize event source: {}", e))?;
-
-    event_source
-        .init_subscriptions()
-        .map_err(|e| eyre!("could not initialize subscriptions: {}", e))?;
-
-    let queries = event_source.queries();
-    info!("listening for queries: {}", queries.iter().format(", "),);
-
-    thread::spawn(|| event_source.run());
-
-    let subscription = tx_cmd.subscribe()?;
+    let (event_source, monitor_tx) = match &chain_config.event_source {
+        EventSourceMode::Push { url, batch_delay } => EventSource::websocket(
+            chain_config.id.clone(),
+            url.clone(),
+            compat_mode,
+            *batch_delay,
+            rt.clone(),
+        ),
+        EventSourceMode::Pull { interval } => EventSource::rpc(
+            chain_config.id.clone(),
+            HttpClient::new(chain_config.rpc_addr.clone())?,
+            *interval,
+            rt.clone(),
+        ),
+    }?;
+
+    thread::spawn(move || event_source.run());
+
+    let subscription = monitor_tx.subscribe()?;
     Ok(subscription)
 }