You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by pi...@apache.org on 2025/06/25 07:30:46 UTC
(iggy) branch connectors_config updated (f1176e40 -> 68992f79)
This is an automated email from the ASF dual-hosted git repository.
piotr pushed a change to branch connectors_config
in repository https://gitbox.apache.org/repos/asf/iggy.git
discard f1176e40 Init work on connectors config
add 80024ee3 style(repo): add various brand assets (#1910)
add ab9ca614 chore(repo): fix existing Iggy names in whole repo (#1908)
new 68992f79 Init work on connectors config
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (f1176e40)
\
N -- N -- N refs/heads/connectors_config (68992f79)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
README.md | 4 +-
assets/logo/0.5x/iggy-apache-b&wdark@0.5x.png | Bin 0 -> 36299 bytes
assets/logo/0.5x/iggy-apache-b&wlight@0.5x.png | Bin 0 -> 36828 bytes
assets/logo/0.5x/iggy-apache-color-darkbg@0.5x.png | Bin 0 -> 96024 bytes
.../logo/0.5x/iggy-apache-color-lightbg@0.5x.png | Bin 0 -> 97130 bytes
.../0.5x/iggy-apache-logo-wo-sygnet-dark@0.5x.png | Bin 0 -> 14714 bytes
.../0.5x/iggy-apache-logo-wo-sygnet-light@0.5x.png | Bin 0 -> 13915 bytes
.../logo/0.5x/iggy-apache-sygnet-b&wdark@0.5x.png | Bin 0 -> 23012 bytes
.../logo/0.5x/iggy-apache-sygnet-b&wlight@0.5x.png | Bin 0 -> 23210 bytes
.../0.5x/iggy-apache-sygnet-color-darkbg@0.5x.png | Bin 0 -> 68197 bytes
.../0.5x/iggy-apache-sygnet-color-lightbg@0.5x.png | Bin 0 -> 68525 bytes
assets/logo/4x/iggy-apache-b&wdark@4x.png | Bin 0 -> 440202 bytes
assets/logo/4x/iggy-apache-b&wlight@4x.png | Bin 0 -> 430715 bytes
assets/logo/4x/iggy-apache-color-darkbg@4x.png | Bin 0 -> 1312915 bytes
assets/logo/4x/iggy-apache-color-lightbg@4x.png | Bin 0 -> 1340584 bytes
.../logo/4x/iggy-apache-logo-wo-sygnet-dark@4x.png | Bin 0 -> 207686 bytes
.../4x/iggy-apache-logo-wo-sygnet-light@4x.png | Bin 0 -> 197503 bytes
assets/logo/4x/iggy-apache-sygnet-b&wdark@4x.png | Bin 0 -> 290806 bytes
assets/logo/4x/iggy-apache-sygnet-b&wlight@4x.png | Bin 0 -> 288572 bytes
.../logo/4x/iggy-apache-sygnet-color-darkbg@4x.png | Bin 0 -> 863513 bytes
.../4x/iggy-apache-sygnet-color-lightbg@4x.png | Bin 0 -> 869788 bytes
assets/logo/SVG/iggy-apache-b&wdark.svg | 63 +++++++++++++++
assets/logo/SVG/iggy-apache-b&wlight.svg | 77 ++++++++++++++++++
.../logo/SVG/iggy-apache-color-darkbg.svg | 0
.../logo/SVG/iggy-apache-color-lightbg.svg | 0
.../logo/SVG/iggy-apache-logo-wo-sygnet-dark.svg | 11 +++
.../logo/SVG/iggy-apache-logo-wo-sygnet-light.svg | 18 +++++
assets/logo/SVG/iggy-apache-sygnet-b&wdark.svg | 49 ++++++++++++
assets/logo/SVG/iggy-apache-sygnet-b&wlight.svg | 61 ++++++++++++++
.../logo/SVG/iggy-apache-sygnet-color-darkbg.svg | 88 ++++++++++-----------
.../logo/SVG/iggy-apache-sygnet-color-lightbg.svg | 16 +---
.../dashboard/frontend/src/components/footer.rs | 6 +-
core/bench/dashboard/server/src/github/client.rs | 2 +-
foreign/cpp/README.md | 4 +-
foreign/cpp/sdk/binary.h | 2 +-
foreign/cpp/sdk/model.h | 4 +-
foreign/cpp/tests/CMakeLists.txt | 4 +-
foreign/cpp/tests/e2e/e2e_testutils.cc | 2 +-
foreign/cpp/tests/e2e/server.toml | 12 +--
.../Iggy_SDK_Tests/E2ETests/ConsumerGroupE2E.cs | 34 ++++----
.../Iggy_SDK_Tests/E2ETests/FetchMessagesE2E.cs | 20 ++---
.../Iggy_SDK_Tests/E2ETests/FlushMessagesE2E.cs | 26 +++---
.../csharp/Iggy_SDK_Tests/E2ETests/OffsetE2E.cs | 8 +-
foreign/csharp/Iggy_SDK_Tests/E2ETests/PATE2E.cs | 26 +++---
.../Iggy_SDK_Tests/E2ETests/SendMessagesE2E.cs | 20 ++---
foreign/csharp/README.md | 7 +-
.../go/benchmarks/send_messages_benchmark_test.go | 4 +-
.../binary_request_serializer.go | 2 +-
.../binary_response_deserializer.go | 4 +-
.../create_stream_serializer.go | 2 +-
.../create_stream_serializer_test.go | 2 +-
.../create_topic_serializer.go | 2 +-
.../fetch_messages_request_serializer.go | 2 +-
.../fetch_messages_request_serializer_test.go | 2 +-
.../binary_serialization/identifier_serializer.go | 2 +-
.../identifier_serializer_test.go | 2 +-
.../log_in_request_serializer.go | 2 +-
.../send_messages_request_serializer.go | 2 +-
.../send_messages_request_serializer_test.go | 2 +-
.../go/binary_serialization/stats_serializer.go | 2 +-
.../update_stream_serializer.go | 2 +-
.../update_stream_serializer_test.go | 2 +-
.../update_topic_serializer.go | 2 +-
.../update_topic_serializer_test.go | 2 +-
foreign/go/cli/cli.go | 4 +-
foreign/go/contracts/messages.go | 2 +-
.../go/e2e/tcp_test/consumers_feature_create.go | 2 +-
.../go/e2e/tcp_test/consumers_feature_delete.go | 4 +-
.../go/e2e/tcp_test/consumers_feature_get_all.go | 2 +-
.../go/e2e/tcp_test/consumers_feature_get_by_id.go | 4 +-
foreign/go/e2e/tcp_test/consumers_feature_join.go | 2 +-
foreign/go/e2e/tcp_test/consumers_feature_leave.go | 2 +-
foreign/go/e2e/tcp_test/consumers_steps.go | 4 +-
foreign/go/e2e/tcp_test/messages_feature_send.go | 2 +-
foreign/go/e2e/tcp_test/messages_steps.go | 4 +-
.../go/e2e/tcp_test/partitions_feature_create.go | 2 +-
.../go/e2e/tcp_test/partitions_feature_delete.go | 2 +-
foreign/go/e2e/tcp_test/partitions_steps.go | 4 +-
foreign/go/e2e/tcp_test/pat_feature_create.go | 2 +-
foreign/go/e2e/tcp_test/pat_feature_delete.go | 2 +-
foreign/go/e2e/tcp_test/pat_steps.go | 4 +-
foreign/go/e2e/tcp_test/session_feature_login.go | 2 +-
foreign/go/e2e/tcp_test/stream_feature_create.go | 2 +-
foreign/go/e2e/tcp_test/stream_feature_delete.go | 2 +-
.../go/e2e/tcp_test/stream_feature_get_by_id.go | 2 +-
foreign/go/e2e/tcp_test/stream_feature_update.go | 2 +-
foreign/go/e2e/tcp_test/stream_steps.go | 6 +-
foreign/go/e2e/tcp_test/test_helpers.go | 4 +-
foreign/go/e2e/tcp_test/test_shared_steps.go | 2 +-
foreign/go/e2e/tcp_test/topic_feature_create.go | 2 +-
foreign/go/e2e/tcp_test/topic_feature_delete.go | 4 +-
foreign/go/e2e/tcp_test/topic_feature_get_all.go | 2 +-
foreign/go/e2e/tcp_test/topic_feature_get_by_id.go | 4 +-
foreign/go/e2e/tcp_test/topic_feature_update.go | 2 +-
foreign/go/e2e/tcp_test/topic_steps.go | 6 +-
foreign/go/e2e/tcp_test/users_feature_create.go | 2 +-
foreign/go/e2e/tcp_test/users_feature_delete.go | 2 +-
foreign/go/e2e/tcp_test/users_feature_get_by_id.go | 2 +-
foreign/go/e2e/tcp_test/users_feature_password.go | 2 +-
.../go/e2e/tcp_test/users_feature_permissions.go | 2 +-
foreign/go/e2e/tcp_test/users_feature_update.go | 2 +-
foreign/go/e2e/tcp_test/users_steps.go | 4 +-
foreign/go/go.mod | 2 +-
foreign/go/message_stream.go | 2 +-
foreign/go/message_stream_factory.go | 4 +-
foreign/go/samples/README.md | 4 +-
foreign/go/samples/consumer/README.md | 2 +-
foreign/go/samples/consumer/consumer.go | 6 +-
foreign/go/samples/producer/README.md | 2 +-
foreign/go/samples/producer/message_generator.go | 2 +-
foreign/go/samples/producer/producer.go | 6 +-
foreign/go/tcp/tcp_access_token_managament.go | 4 +-
foreign/go/tcp/tcp_clients_managament.go | 4 +-
foreign/go/tcp/tcp_consumer_group_managament.go | 6 +-
foreign/go/tcp/tcp_core.go | 10 +--
foreign/go/tcp/tcp_messaging.go | 6 +-
foreign/go/tcp/tcp_offset_managament.go | 4 +-
foreign/go/tcp/tcp_partition_managament.go | 4 +-
foreign/go/tcp/tcp_session_managament.go | 4 +-
foreign/go/tcp/tcp_stream_managament.go | 6 +-
foreign/go/tcp/tcp_topic_managament.go | 6 +-
foreign/go/tcp/tcp_user_managament.go | 6 +-
foreign/go/tcp/tcp_utilities.go | 4 +-
foreign/node/README.md | 4 +-
foreign/node/docker-compose.yml | 2 +-
foreign/node/package.json | 3 +-
foreign/python/.devcontainer/docker-compose.yml | 2 +-
foreign/python/Cargo.toml | 2 +-
foreign/python/README.md | 2 +-
foreign/python/pyproject.toml | 10 +--
web/README.md | 8 +-
131 files changed, 545 insertions(+), 292 deletions(-)
create mode 100644 assets/logo/0.5x/iggy-apache-b&wdark@0.5x.png
create mode 100644 assets/logo/0.5x/iggy-apache-b&wlight@0.5x.png
create mode 100644 assets/logo/0.5x/iggy-apache-color-darkbg@0.5x.png
create mode 100644 assets/logo/0.5x/iggy-apache-color-lightbg@0.5x.png
create mode 100644 assets/logo/0.5x/iggy-apache-logo-wo-sygnet-dark@0.5x.png
create mode 100644 assets/logo/0.5x/iggy-apache-logo-wo-sygnet-light@0.5x.png
create mode 100644 assets/logo/0.5x/iggy-apache-sygnet-b&wdark@0.5x.png
create mode 100644 assets/logo/0.5x/iggy-apache-sygnet-b&wlight@0.5x.png
create mode 100644 assets/logo/0.5x/iggy-apache-sygnet-color-darkbg@0.5x.png
create mode 100644 assets/logo/0.5x/iggy-apache-sygnet-color-lightbg@0.5x.png
create mode 100644 assets/logo/4x/iggy-apache-b&wdark@4x.png
create mode 100644 assets/logo/4x/iggy-apache-b&wlight@4x.png
create mode 100644 assets/logo/4x/iggy-apache-color-darkbg@4x.png
create mode 100644 assets/logo/4x/iggy-apache-color-lightbg@4x.png
create mode 100644 assets/logo/4x/iggy-apache-logo-wo-sygnet-dark@4x.png
create mode 100644 assets/logo/4x/iggy-apache-logo-wo-sygnet-light@4x.png
create mode 100644 assets/logo/4x/iggy-apache-sygnet-b&wdark@4x.png
create mode 100644 assets/logo/4x/iggy-apache-sygnet-b&wlight@4x.png
create mode 100644 assets/logo/4x/iggy-apache-sygnet-color-darkbg@4x.png
create mode 100644 assets/logo/4x/iggy-apache-sygnet-color-lightbg@4x.png
create mode 100644 assets/logo/SVG/iggy-apache-b&wdark.svg
create mode 100644 assets/logo/SVG/iggy-apache-b&wlight.svg
copy web/static/iggy-apache-darkbg.svg => assets/logo/SVG/iggy-apache-color-darkbg.svg (100%)
copy web/static/iggy-apache-lightbg.svg => assets/logo/SVG/iggy-apache-color-lightbg.svg (100%)
create mode 100644 assets/logo/SVG/iggy-apache-logo-wo-sygnet-dark.svg
create mode 100644 assets/logo/SVG/iggy-apache-logo-wo-sygnet-light.svg
create mode 100644 assets/logo/SVG/iggy-apache-sygnet-b&wdark.svg
create mode 100644 assets/logo/SVG/iggy-apache-sygnet-b&wlight.svg
copy web/src/lib/components/Logo/LogoMark.svelte => assets/logo/SVG/iggy-apache-sygnet-color-darkbg.svg (83%)
copy web/static/iggy-apache-lightbg.svg => assets/logo/SVG/iggy-apache-sygnet-color-lightbg.svg (83%)
(iggy) 01/01: Init work on connectors config
Posted by pi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
piotr pushed a commit to branch connectors_config
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 68992f79415b5d7c35bdc04c6cfde85aae44c7b6
Author: spetz <pi...@gmail.com>
AuthorDate: Tue Jun 24 21:24:44 2025 +0200
Init work on connectors config
---
Cargo.lock | 2 +
Cargo.toml | 1 +
core/connectors/README.md | 2 +-
core/connectors/runtime/Cargo.toml | 2 +
core/connectors/runtime/README.md | 2 +-
core/connectors/runtime/src/main.rs | 100 ++++++++++------------------------
core/connectors/runtime/src/stream.rs | 70 ++++++++++++++++++++++++
core/server/Cargo.toml | 2 +-
8 files changed, 106 insertions(+), 75 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index ea6eb0e1..34971091 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3770,6 +3770,8 @@ dependencies = [
"config",
"dashmap",
"dlopen2",
+ "dotenvy",
+ "figlet-rs",
"flume",
"futures",
"iggy",
diff --git a/Cargo.toml b/Cargo.toml
index 479e12c3..290f5bed 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -84,6 +84,7 @@ derive_more = { version = "2.0.1", features = ["full"] }
derive-new = "0.7.0"
dirs = "6.0.0"
dlopen2 = "0.8.0"
+dotenvy = "0.15.7"
enum_dispatch = "0.3.13"
figlet-rs = "0.1.5"
flume = "0.11.1"
diff --git a/core/connectors/README.md b/core/connectors/README.md
index f7dd95dc..8bb81bec 100644
--- a/core/connectors/README.md
+++ b/core/connectors/README.md
@@ -21,7 +21,7 @@ The highly performant and modular runtime for statically typed, yet dynamically
2. Run `docker compose up -d` from `/examples/rust/src/sink-data-producer` which will start the Quickwit server to be used by an example sink connector. At this point, you can access the Quickwit UI at [http://localhost:7280](http://localhost:7280) - check this dashboard again later on, after the `events` index will be created.
-3. Set environment variable `IGGY_CONNECTORS_RUNTIME_CONFIG_PATH=core/connectors/runtime/config` (adjust the path as needed) pointing to the runtime configuration file.
+3. Set environment variable `IGGY_CONNECTORS_CONFIG_PATH=core/connectors/runtime/config` (adjust the path as needed) pointing to the runtime configuration file.
4. Start the Iggy server and invoke the following commands via Iggy CLI to create the example streams and topics used by the sample connectors.
diff --git a/core/connectors/runtime/Cargo.toml b/core/connectors/runtime/Cargo.toml
index c3c3e15c..84912851 100644
--- a/core/connectors/runtime/Cargo.toml
+++ b/core/connectors/runtime/Cargo.toml
@@ -34,6 +34,8 @@ axum-server = { workspace = true }
config = { workspace = true }
dashmap = { workspace = true }
dlopen2 = { workspace = true }
+dotenvy = { workspace = true }
+figlet-rs = { workspace = true }
flume = { workspace = true }
futures = { workspace = true }
iggy = { workspace = true }
diff --git a/core/connectors/runtime/README.md b/core/connectors/runtime/README.md
index 18da28d5..ce5b2a82 100644
--- a/core/connectors/runtime/README.md
+++ b/core/connectors/runtime/README.md
@@ -22,7 +22,7 @@ password = "iggy"
All the other config sections start either with `sources` or `sinks` depending on the connector type.
-Keep in mind that either of `toml`, `yaml`, or `json` formats are supported for the configuration file. The path to the configuration can be overriden by `IGGY_CONNECTORS_RUNTIME_CONFIG_PATH` environment variable. Each configuration section can be also additionally updated by using the following convention `IGGY_CONNECTORS_SECTION_NAME.KEY_NAME` e.g. `IGGY_CONNECTORS_IGGY_USERNAME` and so on.
+Keep in mind that either of `toml`, `yaml`, or `json` formats are supported for the configuration file. The path to the configuration can be overriden by `IGGY_CONNECTORS_CONFIG_PATH` environment variable. Each configuration section can be also additionally updated by using the following convention `IGGY_CONNECTORS_SECTION_NAME.KEY_NAME` e.g. `IGGY_CONNECTORS_IGGY_USERNAME` and so on.
## HTTP API
diff --git a/core/connectors/runtime/src/main.rs b/core/connectors/runtime/src/main.rs
index 67bf303a..9da57d52 100644
--- a/core/connectors/runtime/src/main.rs
+++ b/core/connectors/runtime/src/main.rs
@@ -19,8 +19,10 @@
use config::{Config, Environment, File};
use configs::{ConfigFormat, RuntimeConfig};
use dlopen2::wrapper::{Container, WrapperApi};
+use dotenvy::dotenv;
use error::RuntimeError;
-use iggy::prelude::{Client, IggyClient, IggyClientBuilder, IggyConsumer, IggyProducer};
+use figlet_rs::FIGfont;
+use iggy::prelude::{Client, IggyConsumer, IggyProducer};
use iggy_connector_sdk::{
StreamDecoder, StreamEncoder,
sink::ConsumeCallback,
@@ -33,7 +35,7 @@ use std::{
env,
sync::{Arc, atomic::AtomicU32},
};
-use tracing::{debug, error, info};
+use tracing::{debug, info};
use tracing_subscriber::{EnvFilter, Registry, layer::SubscriberExt, util::SubscriberInitExt};
mod api;
@@ -43,6 +45,7 @@ pub(crate) mod error;
mod manager;
mod sink;
mod source;
+mod stream;
mod transform;
#[global_allocator]
@@ -77,13 +80,29 @@ struct SinkApi {
#[tokio::main]
async fn main() -> Result<(), RuntimeError> {
+ let standard_font = FIGfont::standard().unwrap();
+ let figure = standard_font.convert("Iggy Connectors");
+ println!("{}", figure.unwrap());
+
+ if let Ok(env_path) = std::env::var("IGGY_CONNECTORS_ENV_PATH") {
+ if dotenvy::from_path(&env_path).is_ok() {
+ println!("Loaded environment variables from path: {env_path}");
+ }
+ } else if let Ok(path) = dotenv() {
+ println!(
+ "Loaded environment variables from .env file at path: {}",
+ path.display()
+ );
+ }
+
Registry::default()
.with(tracing_subscriber::fmt::layer())
.with(EnvFilter::try_from_default_env().unwrap_or(EnvFilter::new("INFO")))
.init();
+
let config_path =
- env::var("IGGY_CONNECTORS_RUNTIME_CONFIG_PATH").unwrap_or_else(|_| "config".to_string());
- info!("Starting Iggy Connector Runtime, loading configuration from: {config_path}...");
+ env::var("IGGY_CONNECTORS_CONFIG_PATH").unwrap_or_else(|_| "config".to_string());
+ info!("Starting Iggy Connectors Runtime, loading configuration from: {config_path}...");
let builder = Config::builder()
.add_source(File::with_name(&config_path))
.add_source(Environment::with_prefix("IGGY_CONNECTORS").separator("_"));
@@ -94,30 +113,9 @@ async fn main() -> Result<(), RuntimeError> {
.try_deserialize()
.expect("Failed to deserialize runtime config");
- let iggy_config = config.iggy.clone();
- let iggy_address = iggy_config.address;
- let iggy_username = iggy_config.username;
- let iggy_password = iggy_config.password;
- let iggy_token = iggy_config.token;
- let consumer_client = create_iggy_client(
- &iggy_address,
- iggy_username.as_deref(),
- iggy_password.as_deref(),
- iggy_token.as_deref(),
- )
- .await?;
- consumer_client.connect().await?;
- let producer_client = create_iggy_client(
- &iggy_address,
- iggy_username.as_deref(),
- iggy_password.as_deref(),
- iggy_token.as_deref(),
- )
- .await?;
- producer_client.connect().await?;
-
- let sources = source::init(config.sources.clone(), &producer_client).await?;
- let sinks = sink::init(config.sinks.clone(), &consumer_client).await?;
+ let (stream_consumer, stream_producer) = stream::init(config.iggy.clone()).await?;
+ let sources = source::init(config.sources.clone(), &stream_producer).await?;
+ let sinks = sink::init(config.sinks.clone(), &stream_consumer).await?;
let mut sink_wrappers = vec![];
let mut sink_with_plugins = HashMap::new();
@@ -196,8 +194,8 @@ async fn main() -> Result<(), RuntimeError> {
}
}
- producer_client.shutdown().await?;
- consumer_client.shutdown().await?;
+ stream_producer.shutdown().await?;
+ stream_consumer.shutdown().await?;
info!("All connectors closed. Runtime shutdown complete.");
Ok(())
@@ -220,48 +218,6 @@ pub fn resolve_plugin_path(path: &str) -> String {
}
}
-async fn create_iggy_client(
- address: &str,
- username: Option<&str>,
- password: Option<&str>,
- token: Option<&str>,
-) -> Result<IggyClient, RuntimeError> {
- let connection_string = if let Some(token) = token {
- if token.is_empty() {
- error!("Iggy token cannot be empty (if username and password are not provided)");
- return Err(RuntimeError::MissingIggyCredentials);
- }
-
- let redacted_token = token.chars().take(3).collect::<String>();
- info!("Using token: {redacted_token}*** for Iggy authentication");
- format!("iggy://{token}@{address}")
- } else {
- info!("Using username and password for Iggy authentication");
- let username = username.ok_or(RuntimeError::MissingIggyCredentials)?;
- if username.is_empty() {
- error!("Iggy password cannot be empty (if token is not provided)");
- return Err(RuntimeError::MissingIggyCredentials);
- }
-
- let password = password.ok_or(RuntimeError::MissingIggyCredentials)?;
- if password.is_empty() {
- error!("Iggy password cannot be empty (if token is not provided)");
- return Err(RuntimeError::MissingIggyCredentials);
- }
-
- let redacted_username = username.chars().take(3).collect::<String>();
- let redacted_password = password.chars().take(3).collect::<String>();
- info!(
- "Using username: {redacted_username}***, password: {redacted_password}*** for Iggy authentication"
- );
- format!("iggy://{username}:{password}@{address}")
- };
-
- let client = IggyClientBuilder::from_connection_string(&connection_string)?.build()?;
- client.connect().await?;
- Ok(client)
-}
-
struct SinkConnector {
container: Container<SinkApi>,
plugins: Vec<SinkConnectorPlugin>,
diff --git a/core/connectors/runtime/src/stream.rs b/core/connectors/runtime/src/stream.rs
new file mode 100644
index 00000000..f67afd31
--- /dev/null
+++ b/core/connectors/runtime/src/stream.rs
@@ -0,0 +1,70 @@
+use iggy::prelude::{Client, IggyClient, IggyClientBuilder};
+use tracing::{error, info};
+
+use crate::{configs::IggyConfig, error::RuntimeError};
+
+pub async fn init(config: IggyConfig) -> Result<(IggyClient, IggyClient), RuntimeError> {
+ let iggy_address = config.address;
+ let iggy_username = config.username;
+ let iggy_password = config.password;
+ let iggy_token = config.token;
+ let consumer_client = create_client(
+ &iggy_address,
+ iggy_username.as_deref(),
+ iggy_password.as_deref(),
+ iggy_token.as_deref(),
+ )
+ .await?;
+ consumer_client.connect().await?;
+ let producer_client = create_client(
+ &iggy_address,
+ iggy_username.as_deref(),
+ iggy_password.as_deref(),
+ iggy_token.as_deref(),
+ )
+ .await?;
+ producer_client.connect().await?;
+ Ok((consumer_client, producer_client))
+}
+
+async fn create_client(
+ address: &str,
+ username: Option<&str>,
+ password: Option<&str>,
+ token: Option<&str>,
+) -> Result<IggyClient, RuntimeError> {
+ let connection_string = if let Some(token) = token {
+ if token.is_empty() {
+ error!("Iggy token cannot be empty (if username and password are not provided)");
+ return Err(RuntimeError::MissingIggyCredentials);
+ }
+
+ let redacted_token = token.chars().take(3).collect::<String>();
+ info!("Using token: {redacted_token}*** for Iggy authentication");
+ format!("iggy://{token}@{address}")
+ } else {
+ info!("Using username and password for Iggy authentication");
+ let username = username.ok_or(RuntimeError::MissingIggyCredentials)?;
+ if username.is_empty() {
+ error!("Iggy password cannot be empty (if token is not provided)");
+ return Err(RuntimeError::MissingIggyCredentials);
+ }
+
+ let password = password.ok_or(RuntimeError::MissingIggyCredentials)?;
+ if password.is_empty() {
+ error!("Iggy password cannot be empty (if token is not provided)");
+ return Err(RuntimeError::MissingIggyCredentials);
+ }
+
+ let redacted_username = username.chars().take(3).collect::<String>();
+ let redacted_password = password.chars().take(3).collect::<String>();
+ info!(
+ "Using username: {redacted_username}***, password: {redacted_password}*** for Iggy authentication"
+ );
+ format!("iggy://{username}:{password}@{address}")
+ };
+
+ let client = IggyClientBuilder::from_connection_string(&connection_string)?.build()?;
+ client.connect().await?;
+ Ok(client)
+}
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml
index b0e46455..3a0cf94e 100644
--- a/core/server/Cargo.toml
+++ b/core/server/Cargo.toml
@@ -50,7 +50,7 @@ console-subscriber = { workspace = true, optional = true }
crossbeam = { workspace = true }
dashmap = { workspace = true }
derive_more = { workspace = true }
-dotenvy = "0.15.7"
+dotenvy = { workspace = true }
enum_dispatch = { workspace = true }
error_set = { version = "0.8.5", features = ["tracing"] }
figlet-rs = { workspace = true }