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/05/24 19:12:12 UTC
(iggy) 01/01: feat(repo): add connectors runtime
This is an automated email from the ASF dual-hosted git repository.
piotr pushed a commit to branch connectors
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit c32631cf51571ceba0f81ecec0f84391c43a4923
Author: spetz <pi...@gmail.com>
AuthorDate: Sat May 24 21:11:54 2025 +0200
feat(repo): add connectors runtime
---
Cargo.lock | 372 +++++++++++++++++++++
Cargo.toml | 21 +-
core/connectors/README.md | 23 ++
core/connectors/data_producer/Cargo.toml | 48 +++
core/connectors/data_producer/src/main.rs | 146 ++++++++
core/connectors/docker-compose.yml | 38 +++
core/connectors/runtime/Cargo.toml | 50 +++
core/connectors/runtime/config.toml | 131 ++++++++
core/connectors/runtime/src/main.rs | 361 ++++++++++++++++++++
core/connectors/runtime/src/sink.rs | 344 +++++++++++++++++++
core/connectors/runtime/src/source.rs | 299 +++++++++++++++++
core/connectors/runtime/src/transform.rs | 66 ++++
core/connectors/sdk/Cargo.toml | 48 +++
core/connectors/sdk/src/decoders/json.rs | 37 ++
core/connectors/sdk/src/decoders/mod.rs | 21 ++
core/connectors/sdk/src/decoders/raw.rs | 31 ++
core/connectors/sdk/src/decoders/text.rs | 37 ++
core/connectors/sdk/src/encoders/json.rs | 39 +++
core/connectors/sdk/src/encoders/mod.rs | 21 ++
core/connectors/sdk/src/encoders/raw.rs | 34 ++
core/connectors/sdk/src/encoders/text.rs | 37 ++
core/connectors/sdk/src/lib.rs | 239 +++++++++++++
core/connectors/sdk/src/sink.rs | 240 +++++++++++++
core/connectors/sdk/src/source.rs | 218 ++++++++++++
core/connectors/sdk/src/transforms/add_fields.rs | 128 +++++++
.../connectors/sdk/src/transforms/delete_fields.rs | 61 ++++
core/connectors/sdk/src/transforms/mod.rs | 45 +++
core/connectors/sinks/quickwit_sink/Cargo.toml | 45 +++
core/connectors/sinks/quickwit_sink/src/lib.rs | 209 ++++++++++++
core/connectors/sinks/stdout_sink/Cargo.toml | 42 +++
core/connectors/sinks/stdout_sink/src/lib.rs | 88 +++++
core/connectors/sources/test_source/Cargo.toml | 47 +++
core/connectors/sources/test_source/src/lib.rs | 157 +++++++++
33 files changed, 3720 insertions(+), 3 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 96dee5e0..972240a7 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -420,6 +420,12 @@ version = "1.7.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "69f7f8c3906b62b754cd5326047894316021dcfe5a194c8ea52bdd94934a3457"
+[[package]]
+name = "arraydeque"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7d902e3d592a523def97af8f317b08ce16b7ab854c1985a0c671e6f15cebc236"
+
[[package]]
name = "arrayref"
version = "0.3.9"
@@ -595,6 +601,15 @@ dependencies = [
"bytemuck",
]
+[[package]]
+name = "atomic-polyfill"
+version = "1.0.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8cf2bce30dfe09ef0bfaef228b9d414faaf7e563035494d7fe092dba54b300f4"
+dependencies = [
+ "critical-section",
+]
+
[[package]]
name = "atomic-waker"
version = "1.1.2"
@@ -1012,6 +1027,9 @@ name = "bitflags"
version = "2.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5c8214115b7bf84099f1309324e63141d4c5d7cc26862f97a0a857dbefe165bd"
+dependencies = [
+ "serde",
+]
[[package]]
name = "bitvec"
@@ -1406,6 +1424,12 @@ dependencies = [
"cc",
]
+[[package]]
+name = "cobs"
+version = "0.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "67ba02a97a2bd10f4b59b25c7973101c79642302776489e030cd13cdab09ed15"
+
[[package]]
name = "colorchoice"
version = "1.0.3"
@@ -1451,6 +1475,25 @@ dependencies = [
"crossbeam-utils",
]
+[[package]]
+name = "config"
+version = "0.15.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "595aae20e65c3be792d05818e8c63025294ac3cb7e200f11459063a352a6ef80"
+dependencies = [
+ "async-trait",
+ "convert_case 0.6.0",
+ "json5",
+ "pathdiff",
+ "ron",
+ "rust-ini",
+ "serde",
+ "serde_json",
+ "toml",
+ "winnow 0.7.10",
+ "yaml-rust2",
+]
+
[[package]]
name = "console-api"
version = "0.8.1"
@@ -1620,6 +1663,12 @@ dependencies = [
"cfg-if",
]
+[[package]]
+name = "critical-section"
+version = "1.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "790eea4361631c5e7d22598ecd5723ff611904e3344ce8720784c93e3d83d40b"
+
[[package]]
name = "crossbeam"
version = "0.8.4"
@@ -1978,6 +2027,29 @@ dependencies = [
"syn 2.0.101",
]
+[[package]]
+name = "dlopen2"
+version = "0.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9e1297103d2bbaea85724fcee6294c2d50b1081f9ad47d0f6f6f61eda65315a6"
+dependencies = [
+ "dlopen2_derive",
+ "libc",
+ "once_cell",
+ "winapi",
+]
+
+[[package]]
+name = "dlopen2_derive"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f2b99bf03862d7f545ebc28ddd33a665b50865f4dfd84031a393823879bd4c54"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.101",
+]
+
[[package]]
name = "dlv-list"
version = "0.5.2"
@@ -2038,6 +2110,18 @@ version = "1.15.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719"
+[[package]]
+name = "embedded-io"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ef1a6892d9eef45c8fa6b9e0086428a2cca8491aca8f787c534a3d6d0bcb3ced"
+
+[[package]]
+name = "embedded-io"
+version = "0.6.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "edd0f118536f44f5ccd48bcb8b111bdc3de888b58c74639dfb034a357d0f206d"
+
[[package]]
name = "encoding_rs"
version = "0.8.35"
@@ -3016,6 +3100,16 @@ dependencies = [
"tracing",
]
+[[package]]
+name = "halfbrown"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "aa2c385c6df70fd180bbb673d93039dbd2cd34e41d782600bdf6e1ca7bce39aa"
+dependencies = [
+ "hashbrown 0.15.3",
+ "serde",
+]
+
[[package]]
name = "handlebars"
version = "4.5.0"
@@ -3030,6 +3124,15 @@ dependencies = [
"thiserror 1.0.69",
]
+[[package]]
+name = "hash32"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b0c35f58762feb77d74ebe43bdbc3210f09be9fe6742234d573bacc26ed92b67"
+dependencies = [
+ "byteorder",
+]
+
[[package]]
name = "hashbrown"
version = "0.12.3"
@@ -3056,6 +3159,15 @@ dependencies = [
"foldhash",
]
+[[package]]
+name = "hashlink"
+version = "0.10.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7382cf6263419f2d8df38c55d7da83da5c18aef87fc7a7fc1fb1e344edfe14c1"
+dependencies = [
+ "hashbrown 0.15.3",
+]
+
[[package]]
name = "hdrhistogram"
version = "7.5.4"
@@ -3069,6 +3181,20 @@ dependencies = [
"num-traits",
]
+[[package]]
+name = "heapless"
+version = "0.7.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cdc6457c0eb62c71aac4bc17216026d8410337c4126773b9c5daba343f17964f"
+dependencies = [
+ "atomic-polyfill",
+ "hash32",
+ "rustc_version",
+ "serde",
+ "spin",
+ "stable_deref_trait",
+]
+
[[package]]
name = "heck"
version = "0.5.0"
@@ -3544,6 +3670,127 @@ dependencies = [
"tracing",
]
+[[package]]
+name = "iggy_connector_data_producer"
+version = "0.1.0"
+dependencies = [
+ "async-trait",
+ "chrono",
+ "dashmap",
+ "iggy",
+ "once_cell",
+ "postcard",
+ "rand 0.9.1",
+ "reqwest",
+ "serde",
+ "serde_json",
+ "serde_yml",
+ "simd-json",
+ "thiserror 2.0.12",
+ "tokio",
+ "toml",
+ "tracing",
+ "tracing-subscriber",
+]
+
+[[package]]
+name = "iggy_connector_quickwit_sink"
+version = "0.1.0"
+dependencies = [
+ "async-trait",
+ "dashmap",
+ "iggy_connector_sdk",
+ "once_cell",
+ "postcard",
+ "reqwest",
+ "serde",
+ "serde_yml",
+ "simd-json",
+ "toml",
+ "tracing",
+]
+
+[[package]]
+name = "iggy_connector_runtime"
+version = "0.1.0"
+dependencies = [
+ "config",
+ "dashmap",
+ "dlopen2",
+ "flume",
+ "futures",
+ "iggy",
+ "iggy_binary_protocol",
+ "iggy_connector_sdk",
+ "mimalloc",
+ "once_cell",
+ "postcard",
+ "rand 0.9.1",
+ "serde",
+ "serde_json",
+ "simd-json",
+ "thiserror 2.0.12",
+ "tokio",
+ "tracing",
+ "tracing-subscriber",
+]
+
+[[package]]
+name = "iggy_connector_sdk"
+version = "0.1.0"
+dependencies = [
+ "async-trait",
+ "chrono",
+ "dashmap",
+ "flume",
+ "iggy",
+ "iggy_binary_protocol",
+ "once_cell",
+ "postcard",
+ "serde",
+ "serde_json",
+ "simd-json",
+ "strum_macros",
+ "thiserror 2.0.12",
+ "tokio",
+ "tracing",
+ "tracing-subscriber",
+ "uuid",
+]
+
+[[package]]
+name = "iggy_connector_stdout_sink"
+version = "0.1.0"
+dependencies = [
+ "async-trait",
+ "dashmap",
+ "iggy_connector_sdk",
+ "once_cell",
+ "postcard",
+ "serde",
+ "toml",
+ "tracing",
+]
+
+[[package]]
+name = "iggy_connector_test_source"
+version = "0.1.0"
+dependencies = [
+ "async-trait",
+ "dashmap",
+ "flume",
+ "humantime",
+ "iggy_connector_sdk",
+ "once_cell",
+ "postcard",
+ "rand 0.9.1",
+ "serde",
+ "simd-json",
+ "tokio",
+ "toml",
+ "tracing",
+]
+
[[package]]
name = "iggy_examples"
version = "0.0.5"
@@ -3809,6 +4056,17 @@ dependencies = [
"wasm-bindgen",
]
+[[package]]
+name = "json5"
+version = "0.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "96b0db21af676c1ce64250b5f40f3ce2cf27e4e47cb91ed91eb6fe9350b430c1"
+dependencies = [
+ "pest",
+ "pest_derive",
+ "serde",
+]
+
[[package]]
name = "jsonwebtoken"
version = "9.3.1"
@@ -3983,6 +4241,16 @@ dependencies = [
"vcpkg",
]
+[[package]]
+name = "libyml"
+version = "0.0.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3302702afa434ffa30847a83305f0a69d6abd74293b6554c18ec85c7ef30c980"
+dependencies = [
+ "anyhow",
+ "version_check",
+]
+
[[package]]
name = "libz-sys"
version = "1.1.22"
@@ -4815,6 +5083,12 @@ version = "1.0.15"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "57c0d7b74b563b49d38dae00a0c37d4d6de9b432382b2892f0574ddcae73fd0a"
+[[package]]
+name = "pathdiff"
+version = "0.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "df94ce210e5bc13cb6651479fa48d14f601d9858cfe0467f43ae157023b938d3"
+
[[package]]
name = "pbkdf2"
version = "0.12.2"
@@ -4991,6 +5265,19 @@ dependencies = [
"portable-atomic",
]
+[[package]]
+name = "postcard"
+version = "1.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "170a2601f67cc9dba8edd8c4870b15f71a6a2dc196daec8c83f72b59dff628a8"
+dependencies = [
+ "cobs",
+ "embedded-io 0.4.0",
+ "embedded-io 0.6.1",
+ "heapless",
+ "serde",
+]
+
[[package]]
name = "potential_utf"
version = "0.1.2"
@@ -5445,6 +5732,26 @@ dependencies = [
"thiserror 2.0.12",
]
+[[package]]
+name = "ref-cast"
+version = "1.0.24"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4a0ae411dbe946a674d89546582cea4ba2bb8defac896622d6496f14c23ba5cf"
+dependencies = [
+ "ref-cast-impl",
+]
+
+[[package]]
+name = "ref-cast-impl"
+version = "1.0.24"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1165225c21bff1f3bbce98f5a1f889949bc902d3575308cc7b0de30b4f6d27c7"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.101",
+]
+
[[package]]
name = "regex"
version = "1.11.1"
@@ -5637,6 +5944,18 @@ dependencies = [
"syn 1.0.109",
]
+[[package]]
+name = "ron"
+version = "0.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b91f7eff05f748767f183df4320a63d6936e9c6107d97c9e6bdd9784f4289c94"
+dependencies = [
+ "base64 0.21.7",
+ "bitflags 2.9.0",
+ "serde",
+ "serde_derive",
+]
+
[[package]]
name = "route-recognizer"
version = "0.3.1"
@@ -6104,6 +6423,21 @@ dependencies = [
"syn 2.0.101",
]
+[[package]]
+name = "serde_yml"
+version = "0.0.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "59e2dd588bf1597a252c3b920e0143eb99b0f76e4e082f4c92ce34fbc9e71ddd"
+dependencies = [
+ "indexmap 2.9.0",
+ "itoa",
+ "libyml",
+ "memchr",
+ "ryu",
+ "serde",
+ "version_check",
+]
+
[[package]]
name = "serial_test"
version = "3.2.0"
@@ -6253,6 +6587,21 @@ version = "0.3.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d66dc143e6b11c1eddc06d5c423cfc97062865baf299914ab64caa38182078fe"
+[[package]]
+name = "simd-json"
+version = "0.15.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c962f626b54771990066e5435ec8331d1462576cd2d1e62f24076ae014f92112"
+dependencies = [
+ "getrandom 0.3.3",
+ "halfbrown",
+ "ref-cast",
+ "serde",
+ "serde_json",
+ "simdutf8",
+ "value-trait",
+]
+
[[package]]
name = "simdutf8"
version = "0.1.5"
@@ -7169,6 +7518,18 @@ version = "0.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65"
+[[package]]
+name = "value-trait"
+version = "0.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0508fce11ad19e0aab49ce20b6bec7f8f82902ded31df1c9fc61b90f0eb396b8"
+dependencies = [
+ "float-cmp",
+ "halfbrown",
+ "itoa",
+ "ryu",
+]
+
[[package]]
name = "vcpkg"
version = "0.2.15"
@@ -8036,6 +8397,17 @@ dependencies = [
"lzma-sys",
]
+[[package]]
+name = "yaml-rust2"
+version = "0.10.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "18b783b2c2789414f8bb84ca3318fc9c2d7e7be1c22907d37839a58dedb369d3"
+dependencies = [
+ "arraydeque",
+ "encoding_rs",
+ "hashlink",
+]
+
[[package]]
name = "yansi"
version = "1.0.1"
diff --git a/Cargo.toml b/Cargo.toml
index 97035f30..3a4984f1 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -30,6 +30,12 @@ members = [
"core/binary_protocol",
"core/cli",
"core/common",
+ "core/connectors/data_producer",
+ "core/connectors/runtime",
+ "core/connectors/sdk",
+ "core/connectors/sinks/quickwit_sink",
+ "core/connectors/sinks/stdout_sink",
+ "core/connectors/sources/test_source",
"core/examples",
"core/integration",
"core/sdk",
@@ -68,16 +74,18 @@ byte-unit = { version = "5.1.6", default-features = false, features = [
] }
bytes = "1.10.1"
charming = "0.4.0"
-chrono = "0.4.41"
+chrono = { version = "0.4.41", features = ["serde"] }
clap = { version = "4.5.37", features = ["derive"] }
+config = { version = "0.15.11" }
comfy-table = "7.1.4"
crc32fast = "1.4.2"
crossbeam = "0.8.4"
dashmap = "6.1.0"
+derive_builder = "0.20.2"
derive_more = { version = "2.0.1", features = ["full"] }
derive-new = "0.7.0"
dirs = "6.0.0"
-derive_builder = "0.20.2"
+dlopen2 = "0.7.0"
enum_dispatch = "0.3.13"
figlet-rs = "0.1.5"
flume = "0.11.1"
@@ -87,9 +95,11 @@ human-repr = "1.1.0"
humantime = "2.2.0"
keyring = { version = "3.6.2", features = ["sync-secret-service", "vendored"] }
nonzero_lit = "0.1.2"
+once_cell = "1.21.3"
openssl = { version = "0.10.72", features = ["vendored"] }
passterm = "=2.0.1"
quinn = "0.11.8"
+postcard = { version = "1.1.1", features = ["alloc"] }
rand = "0.9.1"
reqwest = { version = "0.12.15", default-features = false, features = [
"json",
@@ -101,7 +111,9 @@ rustls = { version = "0.23.27", features = ["ring"] }
serde = { version = "1.0.219", features = ["derive", "rc"] }
serde_json = "1.0.140"
serde_with = { version = "3.12.0", features = ["base64", "macros"] }
+serde_yml = "0.0.12"
serial_test = "3.2.0"
+simd-json = { version = "0.15.1", features = ["serde_impl"] }
sysinfo = "0.35.0"
tempfile = "3.19.1"
thiserror = "2.0.12"
@@ -116,6 +128,7 @@ tracing-subscriber = { version = "0.3.19", default-features = false, features =
"ansi",
] }
uuid = { version = "1.16.0", features = [
+ "v4",
"v7",
"fast-rng",
"serde",
@@ -123,6 +136,7 @@ uuid = { version = "1.16.0", features = [
] }
rust-s3 = { version = "0.35.1", features = ["default"] }
strum = { version = "0.27.1", features = ["derive"] }
+strum_macros = "0.27.1"
aes-gcm = "0.10.3"
base64 = "0.22.1"
twox-hash = { version = "2.1.0", features = ["xxhash32"] }
@@ -151,8 +165,9 @@ mimalloc = "0.1"
console-subscriber = "0.4.1"
# Path dependencies
-iggy_common = { path = "core/common", version = "0.7.0" }
iggy_binary_protocol = { path = "core/binary_protocol", version = "0.7.0" }
+iggy_common = { path = "core/common", version = "0.7.0" }
+iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.1.0" }
iggy = { path = "core/sdk", version = "0.7.0" }
server = { path = "core/server" }
integration = { path = "core/integration" }
diff --git a/core/connectors/README.md b/core/connectors/README.md
new file mode 100644
index 00000000..0b362a55
--- /dev/null
+++ b/core/connectors/README.md
@@ -0,0 +1,23 @@
+# Apache Iggy Connectors
+
+Readme is WiP, as of now you can do the following to test the connectors:
+
+1. Build the project in release mode, and make sure that the plugins specified in `core/connectors/config.toml` under `path` are available. BTW you can use either of `toml`, `json` or `yaml` formats.
+
+2. Run `docker compose up -d` from `/core/connectors` which will start the Quickwit server to be used by an example sink connector.
+
+3. Set environment variable `IGGY_CONNECTORS_RUNTIME_CONFIG_PATH=core/connectors/runtime/config` pointing to the runtime configuration file.
+
+4. Start the Iggy server and invoke the following commands to create the example streams and topics used by the connectors.
+
+```
+iggy --username iggy --password iggy stream create example
+iggy --username iggy --password iggy stream create qw
+iggy --username iggy --password iggy topic create qw records 1 none 1d
+```
+
+5. Execute `cargo r --bin iggy_connector_data_producer -r` which will start the example data producer application, sending the messages to previously created `qw` stream and `records` topic.
+
+6. Start the connector runtime `cargo r --bin iggy_connector_runtime -r` - you should be able to browse Quickwit UI at `http://localhost:7280` with records being constantly added to the `events` index (this is part of the Quickwit sink). At the same time, you should see the new messages being added to the `example` stream and `topic1` topic by the test source connector - you can use Iggy Web UI to browse the data.
+
+New connector can be built simply by implementing either `Sink` or `Source` trait. Please check the existing examples under `core/connectors/sinks` and `core/connectors/sources` directories.
diff --git a/core/connectors/data_producer/Cargo.toml b/core/connectors/data_producer/Cargo.toml
new file mode 100644
index 00000000..9789b88d
--- /dev/null
+++ b/core/connectors/data_producer/Cargo.toml
@@ -0,0 +1,48 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[package]
+name = "iggy_connector_data_producer"
+version = "0.1.0"
+description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
+edition = "2024"
+license = "Apache-2.0"
+keywords = ["iggy", "messaging", "streaming"]
+categories = ["command-line-utilities", "database", "network-programming"]
+homepage = "https://iggy.apache.org"
+documentation = "https://iggy.apache.org/docs"
+repository = "https://github.com/apache/iggy"
+readme = "../../README.md"
+
+[dependencies]
+iggy = { workspace = true }
+async-trait = { workspace = true }
+chrono = { workspace = true }
+dashmap = { workspace = true }
+once_cell = { workspace = true }
+serde = { workspace = true }
+postcard = { workspace = true }
+rand = { workspace = true }
+reqwest = { workspace = true }
+serde_yml = { workspace = true }
+serde_json = { workspace = true }
+simd-json = { workspace = true }
+thiserror = { workspace = true }
+toml = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }
+tracing-subscriber = { workspace = true }
diff --git a/core/connectors/data_producer/src/main.rs b/core/connectors/data_producer/src/main.rs
new file mode 100644
index 00000000..2607d955
--- /dev/null
+++ b/core/connectors/data_producer/src/main.rs
@@ -0,0 +1,146 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use std::{env, str::FromStr, time::Duration};
+
+use chrono::{DateTime, Days, Utc};
+use iggy::prelude::{
+ Client, IggyClient, IggyClientBuilder, IggyDuration, IggyError, IggyMessage, Partitioning,
+};
+use rand::{
+ Rng,
+ distr::{Alphanumeric, Uniform},
+};
+use serde::{Deserialize, Serialize};
+use thiserror::Error;
+use tokio::time::sleep;
+use tracing::info;
+use tracing_subscriber::{EnvFilter, Registry, layer::SubscriberExt, util::SubscriberInitExt};
+
+const SOURCES: [&str; 6] = ["browser", "mobile", "desktop", "email", "network", "other"];
+const STATES: [&str; 5] = ["active", "inactive", "blocked", "deleted", "unknown"];
+const DOMAINS: [&str; 5] = [
+ "gmail.com",
+ "yahoo.com",
+ "hotmail.com",
+ "outlook.com",
+ "aol.com",
+];
+
+#[tokio::main]
+async fn main() -> Result<(), DataProducerError> {
+ Registry::default()
+ .with(tracing_subscriber::fmt::layer())
+ .with(EnvFilter::try_from_default_env().unwrap_or(EnvFilter::new("INFO")))
+ .init();
+ info!("Starting data producer...");
+ let address = env::var("IGGY_ADDRESS").unwrap_or("localhost:8090".to_owned());
+ let username = env::var("IGGY_USERNAME").unwrap_or("iggy".to_owned());
+ let password = env::var("IGGY_PASSWORD").unwrap_or("iggy".to_owned());
+ let stream = env::var("IGGY_STREAM").unwrap_or("qw".to_owned());
+ let topic = env::var("IGGY_TOPIC").unwrap_or("records".to_owned());
+ let client = create_client(&address, &username, &password).await?;
+ let mut producer = client
+ .producer(&stream, &topic)?
+ .batch_size(1000)
+ .send_interval(IggyDuration::from_str("5ms").unwrap())
+ .partitioning(Partitioning::balanced())
+ .build();
+ producer.init().await?;
+
+ let mut rng = rand::rng();
+ let mut batches_count = 0;
+ while batches_count < 100000 {
+ let records_count = rng.sample(Uniform::new(500u32, 1000).unwrap());
+ let messages = (0..records_count)
+ .map(|_| random_record())
+ .flat_map(|record| serde_json::to_string(&record).ok())
+ .flat_map(|payload| IggyMessage::from_str(&payload).ok())
+ .collect::<Vec<_>>();
+ producer.send(messages).await?;
+ info!("Sent {records_count} messages");
+ sleep(Duration::from_millis(10)).await;
+ batches_count += 1;
+ }
+
+ info!("Reached maximum batches count");
+ Ok(())
+}
+
+async fn create_client(
+ address: &str,
+ username: &str,
+ password: &str,
+) -> Result<IggyClient, IggyError> {
+ let connection_string = format!("iggy://{username}:{password}@{address}");
+ let client = IggyClientBuilder::from_connection_string(&connection_string)?.build()?;
+ client.connect().await?;
+ Ok(client)
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct Record {
+ user_id: String,
+ user_type: u8,
+ email: String,
+ source: String,
+ state: String,
+ created_at: DateTime<Utc>,
+ message: String,
+}
+
+fn random_record() -> Record {
+ let mut rng = rand::rng();
+ let source =
+ SOURCES[rng.sample(Uniform::new(0u8, SOURCES.len() as u8).unwrap()) as usize].to_owned();
+ let state =
+ STATES[rng.sample(Uniform::new(0u8, STATES.len() as u8).unwrap()) as usize].to_owned();
+ let email = format!(
+ "{}@{}",
+ random_string(rng.sample(Uniform::new(3u32, 20).unwrap()) as usize),
+ DOMAINS[rng.sample(Uniform::new(0u8, DOMAINS.len() as u8).unwrap()) as usize]
+ );
+ let created_at = Utc::now()
+ .checked_sub_days(Days::new(rng.sample(Uniform::new(0u64, 1000).unwrap())))
+ .unwrap();
+ Record {
+ user_id: format!("user_{}", rng.sample(Uniform::new(1u32, 100).unwrap())),
+ user_type: rng.sample(Uniform::new(1u8, 5).unwrap()),
+ email,
+ source,
+ state,
+ message: random_string(rng.sample(Uniform::new(10u32, 100).unwrap()) as usize),
+ created_at,
+ }
+}
+
+fn random_string(size: usize) -> String {
+ let mut rng = rand::rng();
+ let text: String = (0..size)
+ .map(|_| rng.sample(Alphanumeric) as char)
+ .collect();
+ text
+}
+
+#[derive(Debug, Error)]
+enum DataProducerError {
+ #[error("Iggy client error")]
+ IggyClient(#[from] iggy::prelude::ClientError),
+ #[error("Iggy error")]
+ IggyError(#[from] iggy::prelude::IggyError),
+}
diff --git a/core/connectors/docker-compose.yml b/core/connectors/docker-compose.yml
new file mode 100644
index 00000000..5657e42e
--- /dev/null
+++ b/core/connectors/docker-compose.yml
@@ -0,0 +1,38 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+services:
+ quickwit:
+ image: quickwit/quickwit:edge
+ container_name: quickwit
+ restart: unless-stopped
+ volumes:
+ - quickwit:/quickwit/qwdata
+ ports:
+ - 7280:7280
+ - 7281:7281
+ command: run
+ environment:
+ - QW_ENABLE_OTLP_ENDPOINT=true
+ - QW_ENABLE_JAEGER_ENDPOINT=true
+ - QW_ENABLE_OPENTELEMETRY_OTLP_EXPORTER=true
+ - QW_ENABLE_INGEST_V2=true
+ - OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:7281
+ - RUST_LOG=info,quickwit_actors=error,indexing_split_store=error,tantivy=error,quickwit_serve=warn,quickwit_indexing=warn,quickwit=warn
+
+volumes:
+ quickwit:
diff --git a/core/connectors/runtime/Cargo.toml b/core/connectors/runtime/Cargo.toml
new file mode 100644
index 00000000..9e10a261
--- /dev/null
+++ b/core/connectors/runtime/Cargo.toml
@@ -0,0 +1,50 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[package]
+name = "iggy_connector_runtime"
+version = "0.1.0"
+description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
+edition = "2024"
+license = "Apache-2.0"
+keywords = ["iggy", "messaging", "streaming"]
+categories = ["command-line-utilities", "database", "network-programming"]
+homepage = "https://iggy.apache.org"
+documentation = "https://iggy.apache.org/docs"
+repository = "https://github.com/apache/iggy"
+readme = "../../README.md"
+
+[dependencies]
+config = { workspace = true }
+dashmap = { workspace = true }
+dlopen2 = { workspace = true }
+flume = { workspace = true }
+futures = { workspace = true }
+iggy = { workspace = true }
+iggy_binary_protocol = { workspace = true }
+iggy_connector_sdk = { workspace = true }
+mimalloc = { workspace = true }
+once_cell = { workspace = true }
+postcard = { workspace = true }
+rand = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+simd-json = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }
+tracing-subscriber = { workspace = true }
diff --git a/core/connectors/runtime/config.toml b/core/connectors/runtime/config.toml
new file mode 100644
index 00000000..618120dd
--- /dev/null
+++ b/core/connectors/runtime/config.toml
@@ -0,0 +1,131 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[iggy]
+address = "localhost:8090"
+username = "iggy"
+password = "iggy"
+token = "secret"
+
+[sources.test1]
+enabled = true
+name = "Test source"
+path = "target/release/libiggy_connector_test_source"
+
+[[sources.test1.streams]]
+stream = "example"
+topic = "topic1"
+schema = "json"
+batch_size = 1000
+send_interval = "5ms"
+
+[sources.test1.config]
+interval = "100ms"
+# max_count = 1000
+messages_range = [10, 50]
+payload_size = 200
+
+[sources.test1.transforms.add_fields]
+enabled = true
+
+[[sources.test1.transforms.add_fields.fields]]
+key = "test_field"
+value.static = "hello!"
+
+[sinks.quickwit]
+enabled = true
+name = "Quickwit sink 1"
+path = "target/release/libiggy_connector_quickwit_sink"
+
+[[sinks.quickwit.streams]]
+stream = "qw"
+topics = ["records"]
+schema = "json"
+batch_size = 1000
+poll_interval = "5ms"
+consumer_group = "qw_sink_connector"
+
+[sinks.quickwit.transforms.add_fields]
+enabled = true
+
+[[sinks.quickwit.transforms.add_fields.fields]]
+key = "service_name"
+value.static = "qw_connector"
+
+[[sinks.quickwit.transforms.add_fields.fields]]
+key = "timestamp"
+value.computed = "timestamp_millis"
+
+[[sinks.quickwit.ransforms.add_fields.fields]]
+key = "random_id"
+value.computed = "uuid_v7"
+
+[sinks.quickwit.transforms.delete_fields]
+enabled = true
+fields = ["email", "created_at"]
+
+[sinks.quickwit.config]
+url = "http://localhost:7280"
+index = """
+version: 0.9
+
+index_id: events
+
+doc_mapping:
+ mode: strict
+ field_mappings:
+ - name: timestamp
+ type: datetime
+ input_formats: [unix_timestamp]
+ output_format: unix_timestamp_nanos
+ indexed: false
+ fast: true
+ fast_precision: milliseconds
+ - name: service_name
+ type: text
+ tokenizer: raw
+ fast: true
+ - name: random_id
+ type: text
+ tokenizer: raw
+ fast: true
+ - name: user_id
+ type: text
+ tokenizer: raw
+ fast: true
+ - name: user_type
+ type: u64
+ fast: true
+ - name: source
+ type: text
+ tokenizer: default
+ - name: state
+ type: text
+ tokenizer: default
+ - name: message
+ type: text
+ tokenizer: default
+
+ timestamp_field: timestamp
+
+indexing_settings:
+ commit_timeout_secs: 10
+
+retention:
+ period: 7 days
+ schedule: daily
+"""
diff --git a/core/connectors/runtime/src/main.rs b/core/connectors/runtime/src/main.rs
new file mode 100644
index 00000000..99fe9eb6
--- /dev/null
+++ b/core/connectors/runtime/src/main.rs
@@ -0,0 +1,361 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use config::{Config, Environment, File};
+use dlopen2::wrapper::{Container, WrapperApi};
+use iggy::prelude::{Client, IggyClient, IggyClientBuilder, IggyConsumer, IggyError, IggyProducer};
+use iggy_connector_sdk::{
+ Schema, StreamDecoder, StreamEncoder,
+ sink::ConsumeCallback,
+ source::{HandleCallback, SendCallback},
+ transforms::{Transform, TransformType},
+};
+use mimalloc::MiMalloc;
+use serde::{Deserialize, Serialize};
+use std::{
+ collections::HashMap,
+ env,
+ sync::{Arc, atomic::AtomicU32},
+};
+use thiserror::Error;
+use tracing::{error, info};
+use tracing_subscriber::{EnvFilter, Registry, layer::SubscriberExt, util::SubscriberInitExt};
+
+mod sink;
+mod source;
+mod transform;
+
+#[global_allocator]
+static GLOBAL: MiMalloc = MiMalloc;
+
+#[derive(WrapperApi)]
+pub struct SinkApi {
+ open: extern "C" fn(id: u32, config_ptr: *const u8, config_len: usize) -> i32,
+ #[allow(clippy::too_many_arguments)]
+ consume: extern "C" fn(
+ id: u32,
+ topic_meta_ptr: *const u8,
+ topic_meta_len: usize,
+ messages_meta_ptr: *const u8,
+ messages_meta_len: usize,
+ messages_ptr: *const u8,
+ messages_len: usize,
+ ) -> i32,
+ close: extern "C" fn(id: u32) -> i32,
+}
+
+#[derive(WrapperApi)]
+struct SourceApi {
+ open: extern "C" fn(id: u32, config_ptr: *const u8, config_len: usize) -> i32,
+ handle: extern "C" fn(id: u32, callback: SendCallback) -> i32,
+ close: extern "C" fn(id: u32) -> i32,
+}
+
+static PLUGIN_ID: AtomicU32 = AtomicU32::new(1);
+
+struct SinkConnector {
+ container: Container<SinkApi>,
+ plugins: Vec<SinkConnectorPlugin>,
+}
+
+struct SinkConnectorPlugin {
+ id: u32,
+ consumers: Vec<SinkConnectorConsumer>,
+}
+
+struct SinkConnectorConsumer {
+ batch_size: u32,
+ consumer: IggyConsumer,
+ decoder: Arc<dyn StreamDecoder>,
+ transforms: Vec<Arc<dyn Transform>>,
+}
+
+struct SinkConnectorWrapper {
+ callback: ConsumeCallback,
+ plugins: Vec<SinkConnectorPlugin>,
+}
+
+struct SinkWithPlugins {
+ container: Container<SinkApi>,
+ plugin_ids: Vec<u32>,
+}
+
+struct SourceConnector {
+ container: Container<SourceApi>,
+ plugins: Vec<SourceConnectorPlugin>,
+}
+
+struct SourceConnectorPlugin {
+ id: u32,
+ transforms: Vec<Arc<dyn Transform>>,
+ producer: Option<SourceConnectorProducer>,
+}
+
+struct SourceConnectorProducer {
+ encoder: Arc<dyn StreamEncoder>,
+ producer: IggyProducer,
+}
+
+struct SourceWithPlugins {
+ container: Container<SourceApi>,
+ plugin_ids: Vec<u32>,
+}
+
+struct SourceConnectorWrapper {
+ callback: HandleCallback,
+ plugins: Vec<SourceConnectorPlugin>,
+}
+
+#[tokio::main]
+async fn main() -> Result<(), RuntimeError> {
+ 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}...");
+ let builder = Config::builder()
+ .add_source(File::with_name(&config_path))
+ .add_source(Environment::with_prefix("IGGY").separator("_"));
+
+ let config: RuntimeConfig = builder
+ .build()
+ .expect("Failed to build config")
+ .try_deserialize()
+ .expect("Failed to deserialize config");
+
+ let iggy_address = config.iggy.address;
+ let iggy_username = config.iggy.username.expect("Iggy username must be set");
+ let iggy_password = config.iggy.password.expect("Iggy password must be set");
+ let consumer_client = create_client(&iggy_address, &iggy_username, &iggy_password).await?;
+ consumer_client.connect().await?;
+ let producer_client = create_client(&iggy_address, &iggy_username, &iggy_password).await?;
+ producer_client.connect().await?;
+
+ let sources = source::init(config.sources, &producer_client).await?;
+ let sinks = sink::init(config.sinks, &consumer_client).await?;
+
+ let mut sink_wrappers = vec![];
+ let mut sink_with_plugins = HashMap::new();
+ for (key, sink) in sinks {
+ let plugin_ids = sink.plugins.iter().map(|plugin| plugin.id).collect();
+ sink_wrappers.push(SinkConnectorWrapper {
+ callback: sink.container.consume,
+ plugins: sink.plugins,
+ });
+ sink_with_plugins.insert(
+ key,
+ SinkWithPlugins {
+ container: sink.container,
+ plugin_ids,
+ },
+ );
+ }
+
+ let mut source_wrappers = vec![];
+ let mut source_with_plugins = HashMap::new();
+ for (key, source) in sources {
+ let plugin_ids = source.plugins.iter().map(|plugin| plugin.id).collect();
+ source_wrappers.push(SourceConnectorWrapper {
+ callback: source.container.handle,
+ plugins: source.plugins,
+ });
+ source_with_plugins.insert(
+ key,
+ SourceWithPlugins {
+ container: source.container,
+ plugin_ids,
+ },
+ );
+ }
+
+ source::handle(source_wrappers);
+ sink::consume(sink_wrappers);
+ info!("All sources and sinks spawned.");
+
+ #[cfg(unix)]
+ let (mut ctrl_c, mut sigterm) = {
+ use tokio::signal::unix::{SignalKind, signal};
+ (
+ signal(SignalKind::interrupt()).expect("Failed to create SIGINT signal"),
+ signal(SignalKind::terminate()).expect("Failed to create SIGTERM signal"),
+ )
+ };
+
+ #[cfg(unix)]
+ tokio::select! {
+ _ = ctrl_c.recv() => {
+ info!("Received SIGINT. Shutting down connectors...");
+ },
+ _ = sigterm.recv() => {
+ info!("Received SIGTERM. Shutting down connectors...");
+ }
+ }
+
+ for (key, source) in source_with_plugins {
+ for plugin_id in source.plugin_ids {
+ info!("Closing source connector with ID: {plugin_id} for plugin: {key}");
+ source.container.close(plugin_id);
+ info!("Closed source connector with ID: {plugin_id} for plugin: {key}");
+ }
+ }
+
+ for (key, sink) in sink_with_plugins {
+ for plugin_id in sink.plugin_ids {
+ info!("Closing sink connector with ID: {plugin_id} for plugin: {key}",);
+ sink.container.close(plugin_id);
+ info!("Closed sink connector with ID: {plugin_id} for plugin: {key}");
+ }
+ }
+
+ producer_client.shutdown().await?;
+ consumer_client.shutdown().await?;
+
+ info!("All connectors closed.");
+ Ok(())
+}
+
+async fn create_client(
+ address: &str,
+ username: &str,
+ password: &str,
+) -> Result<IggyClient, IggyError> {
+ let connection_string = format!("iggy://{username}:{password}@{address}");
+ let client = IggyClientBuilder::from_connection_string(&connection_string)?.build()?;
+ client.connect().await?;
+ Ok(client)
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct Record {
+ title: String,
+ name: String,
+ age: u32,
+ text: String,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+struct RuntimeConfig {
+ iggy: IggyConfig,
+ sinks: HashMap<String, SinkConfig>,
+ sources: HashMap<String, SourceConfig>,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct IggyConfig {
+ address: String,
+ username: Option<String>,
+ password: Option<String>,
+ token: Option<String>,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct SinkConfig {
+ enabled: bool,
+ name: String,
+ path: String,
+ transforms: Option<TransformsConfig>,
+ streams: Vec<StreamConsumerConfig>,
+ config: Option<serde_json::Value>,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct StreamConsumerConfig {
+ stream: String,
+ topics: Vec<String>,
+ schema: Schema,
+ batch_size: Option<u32>,
+ poll_interval: Option<String>,
+ consumer_group: Option<String>,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct StreamProducerConfig {
+ stream: String,
+ topic: String,
+ schema: Schema,
+ batch_size: Option<u32>,
+ send_interval: Option<String>,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct SourceConfig {
+ enabled: bool,
+ name: String,
+ path: String,
+ transforms: Option<TransformsConfig>,
+ streams: Vec<StreamProducerConfig>,
+ config: Option<serde_json::Value>,
+}
+
+#[derive(Debug, Default, Serialize, Deserialize)]
+struct TransformsConfig {
+ #[serde(flatten)]
+ transforms: HashMap<TransformType, serde_json::Value>,
+}
+
+#[derive(Debug, Default, Serialize, Deserialize)]
+struct InitialTransformConfig {
+ enabled: bool,
+}
+
+#[derive(Debug, Error)]
+pub enum RuntimeError {
+ #[error("Invalid config")]
+ InvalidConfig,
+ #[error("Invalid record")]
+ InvalidRecord,
+ #[error("Failed to serialize topic metadata")]
+ FailedToSerializeTopicMetadata,
+ #[error("Failed to serialize messages metadata")]
+ FailedToSerializeMessagesMetadata,
+ #[error("Failed to serialize raw messages")]
+ FailedToSerializeRawMessages,
+ #[error("Failed to serialize headers")]
+ FailedToSerializeHeaders,
+ #[error("Invalid transformer")]
+ InvalidTransformer,
+ #[error("Invalid transform")]
+ InvalidTransform,
+ #[error("Invalid sink")]
+ InvalidSink,
+ #[error("Connector SDK error")]
+ ConnectorSdkError(#[from] iggy_connector_sdk::Error),
+ #[error("Iggy client error")]
+ IggyClient(#[from] iggy::prelude::ClientError),
+ #[error("Iggy error")]
+ IggyError(#[from] iggy::prelude::IggyError),
+}
+
+const ALLOWED_PLUGIN_EXTENSIONS: [&str; 3] = ["so", "dylib", "dll"];
+
+pub fn resolve_plugin_path(path: &str) -> String {
+ let extension = path.split('.').next_back().unwrap_or_default();
+ if ALLOWED_PLUGIN_EXTENSIONS.contains(&extension) {
+ path.to_string()
+ } else {
+ let os_extension = match std::env::consts::OS {
+ "windows" => "dll",
+ "macos" => "dylib",
+ _ => "so",
+ };
+
+ format!("{path}.{os_extension}")
+ }
+}
diff --git a/core/connectors/runtime/src/sink.rs b/core/connectors/runtime/src/sink.rs
new file mode 100644
index 00000000..fd73fd4a
--- /dev/null
+++ b/core/connectors/runtime/src/sink.rs
@@ -0,0 +1,344 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use dlopen2::wrapper::Container;
+use futures::StreamExt;
+use iggy::prelude::{
+ AutoCommit, AutoCommitWhen, IggyClient, IggyConsumer, IggyDuration, IggyMessage,
+ PollingStrategy,
+};
+use iggy_connector_sdk::{
+ DecodedMessage, MessagesMetadata, RawMessage, RawMessages, ReceivedMessage, StreamDecoder,
+ TopicMetadata, sink::ConsumeCallback, transforms::Transform,
+};
+use std::{
+ collections::HashMap,
+ str::FromStr,
+ sync::{Arc, atomic::Ordering},
+ time::Instant,
+};
+use tracing::{error, info, warn};
+
+use crate::{
+ PLUGIN_ID, RuntimeError, SinkApi, SinkConfig, SinkConnector, SinkConnectorConsumer,
+ SinkConnectorPlugin, SinkConnectorWrapper, resolve_plugin_path, transform,
+};
+
+pub async fn init(
+ sink_configs: HashMap<String, SinkConfig>,
+ iggy_client: &IggyClient,
+) -> Result<HashMap<String, SinkConnector>, RuntimeError> {
+ let mut sink_connectors: HashMap<String, SinkConnector> = HashMap::new();
+ for (key, config) in sink_configs {
+ let name = config.name;
+ if !config.enabled {
+ warn!("Sink: {name} is disabled ({key})");
+ continue;
+ }
+
+ let plugin_id = PLUGIN_ID.load(Ordering::Relaxed);
+ let path = resolve_plugin_path(&config.path);
+ info!("Initializing sink container with name: {name} ({key}), plugin: {path}",);
+ if let Some(container) = sink_connectors.get_mut(&path) {
+ info!("Sink container for plugin: {path} is already loaded.",);
+ init_sink(
+ &container.container,
+ &config.config.unwrap_or_default(),
+ plugin_id,
+ );
+ container.plugins.push(SinkConnectorPlugin {
+ id: plugin_id,
+ consumers: vec![],
+ });
+ } else {
+ let container: Container<SinkApi> =
+ unsafe { Container::load(&path).expect("Failed to load sink container") };
+ info!("Sink container for plugin: {path} loaded successfully.",);
+ init_sink(&container, &config.config.unwrap_or_default(), plugin_id);
+ sink_connectors.insert(
+ path.to_owned(),
+ SinkConnector {
+ container,
+ plugins: vec![SinkConnectorPlugin {
+ id: plugin_id,
+ consumers: vec![],
+ }],
+ },
+ );
+ }
+
+ info!(
+ "Sink container with name: {name} ({key}), initialized successfully with ID: {plugin_id}."
+ );
+ PLUGIN_ID.fetch_add(1, Ordering::Relaxed);
+
+ let transforms = if let Some(transforms_config) = config.transforms {
+ let transforms = transform::load(transforms_config).expect("Failed to load transforms");
+ let types = transforms
+ .iter()
+ .map(|t| t.r#type().into())
+ .collect::<Vec<&'static str>>()
+ .join(", ");
+ info!("Enabled transforms for sink: {name} ({key}): {types}",);
+ transforms
+ } else {
+ vec![]
+ };
+
+ let connector = sink_connectors
+ .get_mut(&path)
+ .expect("Failed to get sink connector");
+ let plugin = connector
+ .plugins
+ .iter_mut()
+ .find(|p| p.id == plugin_id)
+ .expect("Failed to get sink plugin");
+
+ for stream in config.streams {
+ let poll_interval =
+ IggyDuration::from_str(&stream.poll_interval.unwrap_or("5ms".to_owned()))
+ .expect("Invalid poll interval");
+ let consumer_group = stream
+ .consumer_group
+ .unwrap_or(format!("iggy-connect-{key}"));
+ let batch_size = stream.batch_size.unwrap_or(1000);
+ for topic in stream.topics {
+ let mut consumer = iggy_client
+ .consumer_group(&consumer_group, &stream.stream, &topic)?
+ .auto_commit(AutoCommit::When(AutoCommitWhen::PollingMessages))
+ .create_consumer_group_if_not_exists()
+ .auto_join_consumer_group()
+ .polling_strategy(PollingStrategy::next())
+ .poll_interval(poll_interval)
+ .batch_size(batch_size)
+ .build();
+
+ consumer.init().await?;
+ plugin.consumers.push(SinkConnectorConsumer {
+ consumer,
+ decoder: stream.schema.decoder(),
+ batch_size,
+ transforms: transforms.clone(),
+ });
+ }
+ }
+ }
+
+ Ok(sink_connectors)
+}
+
+pub fn consume(sinks: Vec<SinkConnectorWrapper>) {
+ for sink in sinks {
+ for plugin in sink.plugins {
+ info!("Starting consume for sink with ID: {}...", plugin.id);
+ for consumer in plugin.consumers {
+ tokio::spawn(async move {
+ consume_messages(
+ plugin.id,
+ consumer.decoder,
+ consumer.batch_size,
+ sink.callback,
+ consumer.transforms,
+ consumer.consumer,
+ )
+ .await
+ .expect("Failed to consume messages")
+ });
+ }
+ info!(
+ "Consume for sink with ID: {} started successfully.",
+ plugin.id
+ );
+ }
+ }
+}
+
+async fn consume_messages(
+ plugin_id: u32,
+ decoder: Arc<dyn StreamDecoder>,
+ batch_size: u32,
+ consume: ConsumeCallback,
+ transforms: Vec<Arc<dyn Transform>>,
+ mut consumer: IggyConsumer,
+) -> Result<(), RuntimeError> {
+ info!("Started consuming messages for plugin with ID: {plugin_id}");
+ let batch_size = batch_size as usize;
+
+ let mut batch = Vec::with_capacity(batch_size);
+ let topic_metadata = TopicMetadata {
+ stream: consumer.stream().to_string(),
+ topic: consumer.topic().to_string(),
+ };
+
+ while let Some(message) = consumer.next().await {
+ let Ok(message) = message else {
+ error!("Failed to receive message.");
+ continue;
+ };
+
+ let partition_id = message.partition_id;
+ let current_offset = message.current_offset;
+ let message_offset = message.message.header.offset;
+ batch.push(message.message);
+ if current_offset != message_offset && batch.len() < batch_size {
+ continue;
+ }
+
+ let messages = std::mem::take(&mut batch);
+ let messages_count = messages.len();
+ let messages_metadata = MessagesMetadata {
+ partition_id,
+ current_offset,
+ schema: decoder.schema(),
+ };
+ info!("Processing {messages_count} messages");
+ let start = Instant::now();
+ if let Err(error) = process_messages(
+ plugin_id,
+ messages_metadata,
+ &topic_metadata,
+ messages,
+ &consume,
+ &transforms,
+ &decoder,
+ )
+ .await
+ {
+ error!("Failed to process {messages_count} messages. {error}");
+ continue;
+ }
+
+ let elapsed = start.elapsed();
+ info!("Consumed {messages_count} messages in {:#?}", elapsed);
+ }
+ info!("Stopped consuming messages for plugin with ID: {plugin_id}");
+ Ok(())
+}
+
+async fn process_messages(
+ plugin_id: u32,
+ messages_metadata: MessagesMetadata,
+ topic_metadata: &TopicMetadata,
+ messages: Vec<IggyMessage>,
+ consume: &ConsumeCallback,
+ transforms: &Vec<Arc<dyn Transform>>,
+ decoder: &Arc<dyn StreamDecoder>,
+) -> Result<(), RuntimeError> {
+ let messages = messages.into_iter().map(|message| ReceivedMessage {
+ id: message.header.id,
+ offset: message.header.offset,
+ headers: message.user_headers_map().unwrap_or_default(),
+ payload: message.payload.into(),
+ });
+
+ let count = messages.len();
+ let decoded_messages = messages.into_iter().flat_map(|message| {
+ let Ok(payload) = decoder.decode(message.payload) else {
+ return None;
+ };
+
+ Some(DecodedMessage {
+ id: Some(message.id),
+ offset: Some(message.offset),
+ headers: message.headers,
+ payload,
+ })
+ });
+ let mut raw_messages = Vec::with_capacity(count);
+ for message in decoded_messages {
+ let mut current_message = Some(message);
+ for transform in transforms.iter() {
+ let Some(message) = current_message else {
+ break;
+ };
+
+ current_message = transform.transform(topic_metadata, message)?;
+ }
+
+ // The transform may return no message based on some conditions
+ let Some(message) = current_message else {
+ continue;
+ };
+
+ let Some(offset) = message.offset else {
+ error!("Offset should be present.");
+ continue;
+ };
+
+ let Ok(payload) = message.payload.try_into_vec() else {
+ error!("Failed to get raw payload.");
+ continue;
+ };
+
+ let headers: Result<Vec<u8>, RuntimeError> = if let Some(headers) = message.headers {
+ Ok(postcard::to_allocvec(&headers).map_err(|error| {
+ error!("Failed to serialize headers. {error}");
+ RuntimeError::FailedToSerializeHeaders
+ })?)
+ } else {
+ Ok(vec![])
+ };
+
+ let Ok(headers) = headers else {
+ error!("Failed to serialize headers.");
+ continue;
+ };
+
+ raw_messages.push(RawMessage {
+ offset,
+ headers,
+ payload,
+ });
+ }
+
+ let topic_meta = postcard::to_allocvec(topic_metadata).map_err(|error| {
+ error!("Failed to serialize topic metadata. {error}");
+ RuntimeError::FailedToSerializeTopicMetadata
+ })?;
+
+ let messages_meta = postcard::to_allocvec(&messages_metadata).map_err(|error| {
+ error!("Failed to serialize messages metadata. {error}");
+ RuntimeError::FailedToSerializeMessagesMetadata
+ })?;
+
+ let messages = postcard::to_allocvec(&RawMessages {
+ schema: decoder.schema(),
+ messages: raw_messages,
+ })
+ .map_err(|error| {
+ error!("Failed to serialize raw messages. {error}");
+ RuntimeError::FailedToSerializeRawMessages
+ })?;
+
+ (consume)(
+ plugin_id,
+ topic_meta.as_ptr(),
+ topic_meta.len(),
+ messages_meta.as_ptr(),
+ messages_meta.len(),
+ messages.as_ptr(),
+ messages.len(),
+ );
+
+ Ok(())
+}
+
+fn init_sink(container: &Container<SinkApi>, config: &serde_json::Value, id: u32) {
+ let config = serde_json::to_string(config).expect("Invalid sink config.");
+ (container.open)(id, config.as_ptr(), config.len());
+}
diff --git a/core/connectors/runtime/src/source.rs b/core/connectors/runtime/src/source.rs
new file mode 100644
index 00000000..ac5be012
--- /dev/null
+++ b/core/connectors/runtime/src/source.rs
@@ -0,0 +1,299 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use dashmap::DashMap;
+use dlopen2::wrapper::Container;
+use flume::{Receiver, Sender};
+use iggy::prelude::{HeaderKey, HeaderValue, IggyClient, IggyDuration, IggyError, IggyMessage};
+use iggy_connector_sdk::{
+ DecodedMessage, Error, ProducedMessages, StreamEncoder, TopicMetadata, transforms::Transform,
+};
+use once_cell::sync::Lazy;
+use std::{
+ collections::HashMap,
+ str::FromStr,
+ sync::{Arc, atomic::Ordering},
+};
+use tracing::{debug, error, info, warn};
+
+use crate::{
+ PLUGIN_ID, RuntimeError, SourceApi, SourceConfig, SourceConnector, SourceConnectorPlugin,
+ SourceConnectorProducer, SourceConnectorWrapper, resolve_plugin_path, transform,
+};
+
+pub static SOURCE_SENDERS: Lazy<DashMap<u32, Sender<ProducedMessages>>> = Lazy::new(DashMap::new);
+
+pub async fn init(
+ source_configs: HashMap<String, SourceConfig>,
+ iggy_client: &IggyClient,
+) -> Result<HashMap<String, SourceConnector>, RuntimeError> {
+ let mut source_connectors: HashMap<String, SourceConnector> = HashMap::new();
+ for (key, config) in source_configs {
+ let name = config.name;
+ if !config.enabled {
+ warn!("Source: {name} is disabled ({key})");
+ continue;
+ }
+
+ let plugin_id = PLUGIN_ID.load(Ordering::Relaxed);
+ let path = resolve_plugin_path(&config.path);
+ info!("Initializing source container with name: {name} ({key}), plugin: {path}",);
+ if let Some(container) = source_connectors.get_mut(&path) {
+ info!("Source container for plugin: {path} is already loaded.",);
+ init_source(
+ &container.container,
+ &config.config.unwrap_or_default(),
+ plugin_id,
+ );
+ container.plugins.push(SourceConnectorPlugin {
+ id: plugin_id,
+ producer: None,
+ transforms: vec![],
+ });
+ } else {
+ let container: Container<SourceApi> =
+ unsafe { Container::load(&path).expect("Failed to load source container") };
+ info!("Source container for plugin: {path} loaded successfully.",);
+ init_source(&container, &config.config.unwrap_or_default(), plugin_id);
+ source_connectors.insert(
+ path.to_owned(),
+ SourceConnector {
+ container,
+ plugins: vec![SourceConnectorPlugin {
+ id: plugin_id,
+ producer: None,
+ transforms: vec![],
+ }],
+ },
+ );
+ }
+
+ info!(
+ "Source container with name: {name} ({key}), initialized successfully with ID: {plugin_id}."
+ );
+ PLUGIN_ID.fetch_add(1, Ordering::Relaxed);
+
+ let transforms = if let Some(transforms_config) = config.transforms {
+ let transforms = transform::load(transforms_config).expect("Failed to load transforms");
+ let types = transforms
+ .iter()
+ .map(|t| t.r#type().into())
+ .collect::<Vec<&'static str>>()
+ .join(", ");
+ info!("Enabled transforms for source: {name} ({key}): {types}",);
+ transforms
+ } else {
+ vec![]
+ };
+
+ let connector = source_connectors
+ .get_mut(&path)
+ .expect("Failed to get source connector");
+ let plugin = connector
+ .plugins
+ .iter_mut()
+ .find(|p| p.id == plugin_id)
+ .expect("Failed to get source plugin");
+
+ for stream in config.streams {
+ let send_interval =
+ IggyDuration::from_str(&stream.send_interval.unwrap_or("5ms".to_owned()))
+ .expect("Invalid send interval");
+ let batch_size = stream.batch_size.unwrap_or(1000);
+ let mut producer = iggy_client
+ .producer(&stream.stream, &stream.topic)?
+ .send_interval(send_interval)
+ .batch_size(batch_size)
+ .build();
+
+ producer.init().await?;
+ plugin.producer = Some(SourceConnectorProducer {
+ producer,
+ encoder: stream.schema.encoder(),
+ });
+ plugin.transforms = transforms.clone();
+ }
+ }
+
+ Ok(source_connectors)
+}
+
+pub fn handle(sources: Vec<SourceConnectorWrapper>) {
+ for source in sources {
+ for plugin in source.plugins {
+ let plugin_id = plugin.id;
+ info!("Starting handler for source with ID: {plugin_id}...");
+ let handle = source.callback;
+ tokio::task::spawn_blocking(move || {
+ handle(plugin_id, handle_produced_messages);
+ });
+ info!("Handler for source with ID: {plugin_id} started successfully.");
+
+ let (sender, receiver): (Sender<ProducedMessages>, Receiver<ProducedMessages>) =
+ flume::unbounded();
+ SOURCE_SENDERS.insert(plugin_id, sender);
+ tokio::spawn(async move {
+ info!("Source #{plugin_id} started");
+ let producer = &plugin.producer.expect("Producer not initialized");
+ let encoder = producer.encoder.clone();
+ let producer = &producer.producer;
+ let mut number = 1;
+
+ let topic_metadata = TopicMetadata {
+ stream: producer.stream().to_string(),
+ topic: producer.topic().to_string(),
+ };
+
+ while let Ok(received_messages) = receiver.recv_async().await {
+ let count = received_messages.messages.len();
+ info!("[Source #{plugin_id} received {count} messages",);
+ let schema = received_messages.schema;
+ let mut messages: Vec<DecodedMessage> = Vec::with_capacity(count);
+ for message in received_messages.messages {
+ let Ok(payload) = schema.try_into_payload(message.payload) else {
+ error!(
+ "Failed to decode message payload with schema: {}",
+ received_messages.schema
+ );
+ continue;
+ };
+
+ debug!(
+ "[Source #{plugin_id}] Message: {number} | Schema: {schema} | Payload: {payload}"
+ );
+ messages.push(DecodedMessage {
+ id: message.id,
+ offset: None,
+ headers: message.headers,
+ payload,
+ });
+ number += 1;
+ }
+
+ let Ok(iggy_messages) =
+ process_messages(&encoder, &topic_metadata, messages, &plugin.transforms)
+ else {
+ error!(
+ "Failed to process {count} messages before sending them to stream: {}, topic: {}.",
+ producer.stream(),
+ producer.topic()
+ );
+ continue;
+ };
+
+ if let Err(error) = producer.send(iggy_messages).await {
+ error!(
+ "Failed to send {count} messages to stream: {}, topic: {}. {error}",
+ producer.stream(),
+ producer.topic(),
+ );
+ continue;
+ }
+
+ info!(
+ "Sent {count} messages to stream: {}, topic: {}",
+ producer.stream(),
+ producer.topic()
+ );
+ }
+ });
+ }
+ }
+}
+
+fn process_messages(
+ encoder: &Arc<dyn StreamEncoder>,
+ topic_metadata: &TopicMetadata,
+ messages: Vec<DecodedMessage>,
+ transforms: &Vec<Arc<dyn Transform>>,
+) -> Result<Vec<IggyMessage>, Error> {
+ let mut iggy_messages = Vec::with_capacity(messages.len());
+ for message in messages {
+ let mut current_message = Some(message);
+ for transform in transforms.iter() {
+ let Some(message) = current_message else {
+ break;
+ };
+
+ current_message = transform.transform(topic_metadata, message)?;
+ }
+
+ // The transform may return no message based on some conditions
+ let Some(message) = current_message else {
+ continue;
+ };
+
+ let Ok(payload) = encoder.encode(message.payload) else {
+ error!("Failed to encode message payload");
+ continue;
+ };
+
+ let Ok(iggy_message) = build_iggy_message(payload, message.id, message.headers) else {
+ error!("Failed to build Iggy message");
+ continue;
+ };
+
+ iggy_messages.push(iggy_message);
+ }
+ Ok(iggy_messages)
+}
+
+fn init_source(container: &Container<SourceApi>, config: &serde_json::Value, id: u32) {
+ let config = serde_json::to_string(config).expect("Invalid source config.");
+ (container.open)(id, config.as_ptr(), config.len());
+}
+
+extern "C" fn handle_produced_messages(
+ plugin_id: u32,
+ messages_ptr: *const u8,
+ messages_len: usize,
+) {
+ unsafe {
+ if let Some(sender) = SOURCE_SENDERS.get(&plugin_id) {
+ let messages = std::slice::from_raw_parts(messages_ptr, messages_len);
+ let Ok(messages) = postcard::from_bytes::<ProducedMessages>(messages) else {
+ eprintln!("Failed to deserialize produced messages.");
+ return;
+ };
+ let _ = sender.send(messages);
+ }
+ }
+}
+
+fn build_iggy_message(
+ payload: Vec<u8>,
+ id: Option<u128>,
+ headers: Option<HashMap<HeaderKey, HeaderValue>>,
+) -> Result<IggyMessage, IggyError> {
+ match (id, headers) {
+ (Some(id), Some(h)) => IggyMessage::builder()
+ .payload(payload.into())
+ .id(id)
+ .user_headers(h)
+ .build(),
+ (Some(id), None) => IggyMessage::builder()
+ .payload(payload.into())
+ .id(id)
+ .build(),
+ (None, Some(h)) => IggyMessage::builder()
+ .payload(payload.into())
+ .user_headers(h)
+ .build(),
+ (None, None) => IggyMessage::builder().payload(payload.into()).build(),
+ }
+}
diff --git a/core/connectors/runtime/src/transform.rs b/core/connectors/runtime/src/transform.rs
new file mode 100644
index 00000000..2417fbfa
--- /dev/null
+++ b/core/connectors/runtime/src/transform.rs
@@ -0,0 +1,66 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use std::sync::Arc;
+
+use iggy_connector_sdk::transforms::{
+ Transform, TransformType, add_fields::AddFields, delete_fields::DeleteFields,
+};
+use serde::de::DeserializeOwned;
+use tracing::error;
+
+use crate::{InitialTransformConfig, RuntimeError, TransformsConfig};
+
+pub fn load(config: TransformsConfig) -> Result<Vec<Arc<dyn Transform>>, RuntimeError> {
+ let mut transforms: Vec<Arc<dyn Transform>> = vec![];
+ let mut types = vec![];
+ for (r#type, transform_config) in config.transforms {
+ let initial_config =
+ serde_json::from_value::<InitialTransformConfig>(transform_config.clone())
+ .unwrap_or_default();
+ if !initial_config.enabled {
+ continue;
+ }
+
+ let transform = load_transform(r#type, transform_config)?;
+ types.push(r#type);
+ transforms.push(transform);
+ }
+
+ Ok(transforms)
+}
+
+fn load_transform(
+ r#type: TransformType,
+ config: serde_json::Value,
+) -> Result<Arc<dyn Transform>, RuntimeError> {
+ Ok(match r#type {
+ TransformType::AddFields => Arc::new(AddFields::new(as_config(r#type, config)?)),
+ TransformType::DeleteFields => Arc::new(DeleteFields::new(as_config(r#type, config)?)),
+ })
+}
+
+fn as_config<T: DeserializeOwned>(
+ r#type: TransformType,
+ config: serde_json::Value,
+) -> Result<T, RuntimeError> {
+ serde_json::from_value::<T>(config).map_err(|error| {
+ error!("Failed to deserialize config for transform: {type}. {error}");
+ RuntimeError::InvalidConfig
+ })
+}
diff --git a/core/connectors/sdk/Cargo.toml b/core/connectors/sdk/Cargo.toml
new file mode 100644
index 00000000..0cace762
--- /dev/null
+++ b/core/connectors/sdk/Cargo.toml
@@ -0,0 +1,48 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[package]
+name = "iggy_connector_sdk"
+version = "0.1.0"
+description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
+edition = "2024"
+license = "Apache-2.0"
+keywords = ["iggy", "messaging", "streaming"]
+categories = ["command-line-utilities", "database", "network-programming"]
+homepage = "https://iggy.apache.org"
+documentation = "https://iggy.apache.org/docs"
+repository = "https://github.com/apache/iggy"
+readme = "../../README.md"
+
+[dependencies]
+async-trait = { workspace = true }
+chrono = { workspace = true }
+dashmap = { workspace = true }
+flume = { workspace = true }
+iggy = { workspace = true }
+iggy_binary_protocol = { workspace = true }
+once_cell = { workspace = true }
+postcard = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+simd-json = { workspace = true }
+strum_macros = { workspace = true }
+thiserror = { workspace = true }
+tracing = { workspace = true }
+tracing-subscriber = { workspace = true }
+tokio = { workspace = true }
+uuid = { workspace = true }
diff --git a/core/connectors/sdk/src/decoders/json.rs b/core/connectors/sdk/src/decoders/json.rs
new file mode 100644
index 00000000..f2bda418
--- /dev/null
+++ b/core/connectors/sdk/src/decoders/json.rs
@@ -0,0 +1,37 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use crate::{Error, Payload, Schema, StreamDecoder};
+use tracing::error;
+
+pub struct JsonStreamDecoder;
+
+impl StreamDecoder for JsonStreamDecoder {
+ fn schema(&self) -> Schema {
+ Schema::Json
+ }
+
+ fn decode(&self, mut payload: Vec<u8>) -> Result<Payload, Error> {
+ Ok(Payload::Json(
+ simd_json::to_owned_value(&mut payload).map_err(|error| {
+ error!("Failed to decode JSON payload: {error}");
+ Error::CannotDecode(self.schema())
+ })?,
+ ))
+ }
+}
diff --git a/core/connectors/sdk/src/decoders/mod.rs b/core/connectors/sdk/src/decoders/mod.rs
new file mode 100644
index 00000000..e1b7e1b7
--- /dev/null
+++ b/core/connectors/sdk/src/decoders/mod.rs
@@ -0,0 +1,21 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+pub mod json;
+pub mod raw;
+pub mod text;
diff --git a/core/connectors/sdk/src/decoders/raw.rs b/core/connectors/sdk/src/decoders/raw.rs
new file mode 100644
index 00000000..efbddc12
--- /dev/null
+++ b/core/connectors/sdk/src/decoders/raw.rs
@@ -0,0 +1,31 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use crate::{Error, Payload, Schema, StreamDecoder};
+
+pub struct RawStreamDecoder;
+
+impl StreamDecoder for RawStreamDecoder {
+ fn schema(&self) -> Schema {
+ Schema::Raw
+ }
+
+ fn decode(&self, payload: Vec<u8>) -> Result<Payload, Error> {
+ Ok(Payload::Raw(payload))
+ }
+}
diff --git a/core/connectors/sdk/src/decoders/text.rs b/core/connectors/sdk/src/decoders/text.rs
new file mode 100644
index 00000000..6d7722cb
--- /dev/null
+++ b/core/connectors/sdk/src/decoders/text.rs
@@ -0,0 +1,37 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use crate::{Error, Payload, Schema, StreamDecoder};
+use tracing::error;
+
+pub struct TextStreamDecoder;
+
+impl StreamDecoder for TextStreamDecoder {
+ fn schema(&self) -> Schema {
+ Schema::Text
+ }
+
+ fn decode(&self, payload: Vec<u8>) -> Result<Payload, Error> {
+ Ok(Payload::Text(String::from_utf8(payload).map_err(
+ |error| {
+ error!("Failed to decode text payload: {error}");
+ Error::CannotDecode(self.schema())
+ },
+ )?))
+ }
+}
diff --git a/core/connectors/sdk/src/encoders/json.rs b/core/connectors/sdk/src/encoders/json.rs
new file mode 100644
index 00000000..03f4e4f0
--- /dev/null
+++ b/core/connectors/sdk/src/encoders/json.rs
@@ -0,0 +1,39 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use crate::{Error, Payload, Schema, StreamEncoder};
+
+pub struct JsonStreamEncoder;
+
+impl StreamEncoder for JsonStreamEncoder {
+ fn schema(&self) -> Schema {
+ Schema::Json
+ }
+
+ fn encode(&self, payload: Payload) -> Result<Vec<u8>, Error> {
+ match payload {
+ Payload::Text(value) => {
+ Ok(simd_json::to_vec(&value).map_err(|_| Error::InvalidJsonPayload)?)
+ }
+ Payload::Json(value) => {
+ Ok(simd_json::to_vec(&value).map_err(|_| Error::InvalidJsonPayload)?)
+ }
+ _ => Err(Error::InvalidPayloadType),
+ }
+ }
+}
diff --git a/core/connectors/sdk/src/encoders/mod.rs b/core/connectors/sdk/src/encoders/mod.rs
new file mode 100644
index 00000000..e1b7e1b7
--- /dev/null
+++ b/core/connectors/sdk/src/encoders/mod.rs
@@ -0,0 +1,21 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+pub mod json;
+pub mod raw;
+pub mod text;
diff --git a/core/connectors/sdk/src/encoders/raw.rs b/core/connectors/sdk/src/encoders/raw.rs
new file mode 100644
index 00000000..68b3b9bb
--- /dev/null
+++ b/core/connectors/sdk/src/encoders/raw.rs
@@ -0,0 +1,34 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use crate::{Error, Payload, Schema, StreamEncoder};
+
+pub struct RawStreamEncoder;
+
+impl StreamEncoder for RawStreamEncoder {
+ fn schema(&self) -> Schema {
+ Schema::Raw
+ }
+
+ fn encode(&self, payload: Payload) -> Result<Vec<u8>, Error> {
+ match payload {
+ Payload::Raw(value) => Ok(value),
+ _ => Err(Error::InvalidPayloadType),
+ }
+ }
+}
diff --git a/core/connectors/sdk/src/encoders/text.rs b/core/connectors/sdk/src/encoders/text.rs
new file mode 100644
index 00000000..6fb2d4e9
--- /dev/null
+++ b/core/connectors/sdk/src/encoders/text.rs
@@ -0,0 +1,37 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use crate::{Error, Payload, Schema, StreamEncoder};
+
+pub struct TextStreamEncoder;
+
+impl StreamEncoder for TextStreamEncoder {
+ fn schema(&self) -> Schema {
+ Schema::Text
+ }
+
+ fn encode(&self, payload: Payload) -> Result<Vec<u8>, Error> {
+ match payload {
+ Payload::Text(value) => Ok(value.into_bytes()),
+ Payload::Json(value) => {
+ Ok(simd_json::to_vec(&value).map_err(|_| Error::InvalidJsonPayload)?)
+ }
+ _ => Err(Error::InvalidPayloadType),
+ }
+ }
+}
diff --git a/core/connectors/sdk/src/lib.rs b/core/connectors/sdk/src/lib.rs
new file mode 100644
index 00000000..749c666e
--- /dev/null
+++ b/core/connectors/sdk/src/lib.rs
@@ -0,0 +1,239 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use async_trait::async_trait;
+use decoders::{json::JsonStreamDecoder, raw::RawStreamDecoder, text::TextStreamDecoder};
+use encoders::{json::JsonStreamEncoder, raw::RawStreamEncoder, text::TextStreamEncoder};
+use iggy::prelude::{HeaderKey, HeaderValue};
+use once_cell::sync::OnceCell;
+use serde::{Deserialize, Serialize};
+use std::{collections::HashMap, sync::Arc};
+use strum_macros::{Display, IntoStaticStr};
+use thiserror::Error;
+use tokio::runtime::Runtime;
+
+pub mod decoders;
+pub mod encoders;
+pub mod sink;
+pub mod source;
+pub mod transforms;
+
+static RUNTIME: OnceCell<Runtime> = OnceCell::new();
+
+pub fn get_runtime() -> &'static Runtime {
+ RUNTIME.get_or_init(|| Runtime::new().expect("Failed to create Tokio runtime"))
+}
+
+#[async_trait]
+pub trait Source: Send + Sync {
+ async fn open(&mut self) -> Result<(), Error>;
+ async fn poll(&self) -> Result<ProducedMessages, Error>;
+ async fn close(&mut self);
+}
+
+#[async_trait]
+pub trait Sink: Send + Sync {
+ async fn open(&mut self) -> Result<(), Error>;
+ async fn consume(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: MessagesMetadata,
+ messages: Vec<ConsumedMessage>,
+ ) -> Result<(), Error>;
+ async fn close(&mut self);
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub enum Payload {
+ Json(simd_json::OwnedValue),
+ Raw(Vec<u8>),
+ Text(String),
+}
+
+impl Payload {
+ pub fn try_into_vec(self) -> Result<Vec<u8>, Error> {
+ match self {
+ Payload::Json(value) => {
+ Ok(simd_json::to_vec(&value).map_err(|_| Error::InvalidJsonPayload)?)
+ }
+ Payload::Raw(value) => Ok(value),
+ Payload::Text(text) => Ok(text.into_bytes()),
+ }
+ }
+}
+
+impl std::fmt::Display for Payload {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ match self {
+ Payload::Json(value) => write!(
+ f,
+ "Json({})",
+ simd_json::to_string_pretty(value).unwrap_or_default()
+ ),
+ Payload::Raw(value) => write!(f, "Raw({:#?})", value),
+ Payload::Text(text) => write!(f, "Text({})", text),
+ }
+ }
+}
+
+#[repr(C)]
+#[derive(
+ Debug, Copy, Clone, Eq, Hash, PartialEq, Serialize, Deserialize, Display, IntoStaticStr,
+)]
+#[serde(rename_all = "snake_case")]
+pub enum Schema {
+ #[strum(to_string = "json")]
+ Json,
+ #[strum(to_string = "raw")]
+ Raw,
+ #[strum(to_string = "text")]
+ Text,
+}
+
+impl Schema {
+ pub fn try_into_payload(self, mut value: Vec<u8>) -> Result<Payload, Error> {
+ match self {
+ Schema::Json => Ok(Payload::Json(
+ simd_json::to_owned_value(&mut value).map_err(|_| Error::InvalidJsonPayload)?,
+ )),
+ Schema::Raw => Ok(Payload::Raw(value)),
+ Schema::Text => Ok(Payload::Text(
+ String::from_utf8(value).map_err(|_| Error::InvalidTextPayload)?,
+ )),
+ }
+ }
+
+ pub fn decoder(self) -> Arc<dyn StreamDecoder> {
+ match self {
+ Schema::Json => Arc::new(JsonStreamDecoder),
+ Schema::Raw => Arc::new(RawStreamDecoder),
+ Schema::Text => Arc::new(TextStreamDecoder),
+ }
+ }
+
+ pub fn encoder(self) -> Arc<dyn StreamEncoder> {
+ match self {
+ Schema::Json => Arc::new(JsonStreamEncoder),
+ Schema::Raw => Arc::new(RawStreamEncoder),
+ Schema::Text => Arc::new(TextStreamEncoder),
+ }
+ }
+}
+
+#[repr(C)]
+#[derive(Debug, Serialize, Deserialize)]
+pub struct TopicMetadata {
+ pub stream: String,
+ pub topic: String,
+}
+
+#[repr(C)]
+#[derive(Debug, Serialize, Deserialize)]
+pub struct MessagesMetadata {
+ pub partition_id: u32,
+ pub current_offset: u64,
+ pub schema: Schema,
+}
+
+#[repr(C)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ReceivedMessage {
+ pub id: u128,
+ pub offset: u64,
+ pub headers: Option<HashMap<HeaderKey, HeaderValue>>,
+ pub payload: Vec<u8>,
+}
+
+#[repr(C)]
+#[derive(Debug, Serialize, Deserialize)]
+pub struct ProducedMessages {
+ pub schema: Schema,
+ pub messages: Vec<ProducedMessage>,
+}
+
+#[repr(C)]
+#[derive(Debug, Serialize, Deserialize)]
+pub struct ProducedMessage {
+ pub id: Option<u128>,
+ pub headers: Option<HashMap<HeaderKey, HeaderValue>>,
+ pub payload: Vec<u8>,
+}
+
+#[repr(C)]
+#[derive(Debug, Serialize, Deserialize)]
+pub struct DecodedMessage {
+ pub id: Option<u128>,
+ pub offset: Option<u64>,
+ pub headers: Option<HashMap<HeaderKey, HeaderValue>>,
+ pub payload: Payload,
+}
+
+#[repr(C)]
+#[derive(Debug, Serialize, Deserialize)]
+pub struct RawMessages {
+ pub schema: Schema,
+ pub messages: Vec<RawMessage>,
+}
+
+#[repr(C)]
+#[derive(Debug, Serialize, Deserialize)]
+pub struct RawMessage {
+ pub offset: u64,
+ pub headers: Vec<u8>,
+ pub payload: Vec<u8>,
+}
+
+#[repr(C)]
+#[derive(Debug, Serialize, Deserialize)]
+pub struct ConsumedMessage {
+ pub offset: u64,
+ pub headers: Option<HashMap<HeaderKey, HeaderValue>>,
+ pub payload: Payload,
+}
+
+pub trait StreamDecoder: Send + Sync {
+ fn schema(&self) -> Schema;
+ fn decode(&self, payload: Vec<u8>) -> Result<Payload, Error>;
+}
+
+pub trait StreamEncoder: Send + Sync {
+ fn schema(&self) -> Schema;
+ fn encode(&self, payload: Payload) -> Result<Vec<u8>, Error>;
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Error)]
+pub enum Error {
+ #[error("Invalid config")]
+ InvalidConfig,
+ #[error("Invalid record")]
+ InvalidRecord,
+ #[error("Invalid transformer")]
+ InvalidTransformer,
+ #[error("HTTP request failed: {0}")]
+ HttpRequestFailed(String),
+ #[error("Init error: {0}")]
+ InitError(String),
+ #[error("Invalid payload type")]
+ InvalidPayloadType,
+ #[error("Invalid JSON payload.")]
+ InvalidJsonPayload,
+ #[error("Invalid text payload.")]
+ InvalidTextPayload,
+ #[error("Cannot decode schema {0}")]
+ CannotDecode(Schema),
+}
diff --git a/core/connectors/sdk/src/sink.rs b/core/connectors/sdk/src/sink.rs
new file mode 100644
index 00000000..bd836ac6
--- /dev/null
+++ b/core/connectors/sdk/src/sink.rs
@@ -0,0 +1,240 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use serde::de::DeserializeOwned;
+use tokio::sync::watch;
+use tracing::{error, info};
+use tracing_subscriber::fmt;
+
+use crate::{ConsumedMessage, MessagesMetadata, RawMessages, Sink, TopicMetadata, get_runtime};
+
+pub type ConsumeCallback = extern "C" fn(
+ plugin_id: u32,
+ topic_meta_ptr: *const u8,
+ topic_meta_len: usize,
+ messages_meta_ptr: *const u8,
+ messages_meta_len: usize,
+ messages_ptr: *const u8,
+ messages_len: usize,
+) -> i32;
+
+#[derive(Debug)]
+pub struct SinkContainer<T: Sink + std::fmt::Debug> {
+ id: u32,
+ sink: Option<T>,
+ shutdown: Option<watch::Sender<()>>,
+}
+
+impl<T: Sink + std::fmt::Debug> SinkContainer<T> {
+ pub const fn new(id: u32) -> Self {
+ Self {
+ id,
+ sink: None,
+ shutdown: None,
+ }
+ }
+
+ /// # Safety
+ /// This is safe
+ pub unsafe fn open<F, C>(
+ &mut self,
+ id: u32,
+ config_ptr: *const u8,
+ config_len: usize,
+ factory: F,
+ ) -> i32
+ where
+ F: FnOnce(u32, C) -> T,
+ C: DeserializeOwned,
+ {
+ unsafe {
+ _ = fmt::try_init();
+ let slice = std::slice::from_raw_parts(config_ptr, config_len);
+ let Ok(config_str) = std::str::from_utf8(slice) else {
+ return -1;
+ };
+
+ let Ok(config) = serde_json::from_str(config_str) else {
+ return -1;
+ };
+
+ let mut sink = factory(id, config);
+ let runtime = get_runtime();
+ let result = runtime.block_on(sink.open());
+ self.id = id;
+ self.sink = Some(sink);
+ if result.is_ok() { 0 } else { 1 }
+ }
+ }
+
+ /// # Safety
+ /// This is safe
+ pub unsafe fn close(&mut self) -> i32 {
+ let Some(mut sink) = self.sink.take() else {
+ error!(
+ "Sink with ID: {} is not initialized - cannot close.",
+ self.id
+ );
+ return -1;
+ };
+
+ info!("Closing sink with ID: {}...", self.id);
+ if let Some(tx) = self.shutdown.take() {
+ let _ = tx.send(());
+ }
+
+ let runtime = get_runtime();
+ runtime.block_on(sink.close());
+ info!("Closed sink with ID: {}", self.id);
+ 0
+ }
+
+ /// # Safety
+ /// This is safe
+ pub unsafe fn consume(
+ &self,
+ topic_meta_ptr: *const u8,
+ topic_meta_len: usize,
+ messages_meta_ptr: *const u8,
+ messages_meta_len: usize,
+ messages_ptr: *const u8,
+ messages_len: usize,
+ ) -> i32 {
+ unsafe {
+ let Some(sink) = self.sink.as_ref() else {
+ error!(
+ "Sink with ID: {} is not initialized - cannot consume.",
+ self.id
+ );
+ return -1;
+ };
+
+ let topic_meta_slice =
+ std::slice::from_raw_parts(topic_meta_ptr, topic_meta_len).to_vec();
+ let messages_meta_slice =
+ std::slice::from_raw_parts(messages_meta_ptr, messages_meta_len).to_vec();
+ let messages_slice = std::slice::from_raw_parts(messages_ptr, messages_len).to_vec();
+
+ let Ok(topic_metadata) = postcard::from_bytes::<TopicMetadata>(&topic_meta_slice)
+ else {
+ error!("Failed to decode topic metadata");
+ return -1;
+ };
+
+ let Ok(messages_metadata) =
+ postcard::from_bytes::<MessagesMetadata>(&messages_meta_slice)
+ else {
+ error!("Failed to decode messages metadata");
+ return -1;
+ };
+
+ let Ok(raw_messages) = postcard::from_bytes::<RawMessages>(&messages_slice) else {
+ error!("Failed to decode raw messages");
+ return -1;
+ };
+
+ let mut messages = Vec::with_capacity(raw_messages.messages.len());
+ for message in raw_messages.messages {
+ let headers = if message.headers.is_empty() {
+ None
+ } else {
+ let Ok(headers) = postcard::from_bytes(&message.headers) else {
+ error!("Failed to decode message headers");
+ continue;
+ };
+ Some(headers)
+ };
+
+ let Ok(payload) = messages_metadata.schema.try_into_payload(message.payload) else {
+ error!("Failed to decode message payload");
+ continue;
+ };
+
+ messages.push(ConsumedMessage {
+ offset: message.offset,
+ headers,
+ payload,
+ })
+ }
+
+ let runtime = get_runtime();
+ let result =
+ runtime.block_on(sink.consume(&topic_metadata, messages_metadata, messages));
+ if result.is_ok() { 0 } else { 1 }
+ }
+ }
+}
+
+#[macro_export]
+macro_rules! sink_connector {
+ ($type:ty) => {
+ const _: fn() = || {
+ fn assert_trait<T: $crate::Sink>() {}
+ assert_trait::<$type>();
+ };
+
+ use dashmap::DashMap;
+ use once_cell::sync::Lazy;
+ use $crate::sink::SinkContainer;
+
+ static INSTANCES: Lazy<DashMap<u32, SinkContainer<$type>>> = Lazy::new(DashMap::new);
+
+ #[unsafe(no_mangle)]
+ unsafe extern "C" fn open(id: u32, config_ptr: *const u8, config_len: usize) -> i32 {
+ let mut container = SinkContainer::new(id);
+ let result = container.open(id, config_ptr, config_len, <$type>::new);
+ INSTANCES.insert(id, container);
+ result
+ }
+
+ #[unsafe(no_mangle)]
+ unsafe extern "C" fn consume(
+ id: u32,
+ topic_meta_ptr: *const u8,
+ topic_meta_len: usize,
+ messages_meta_ptr: *const u8,
+ messages_meta_len: usize,
+ messages_ptr: *const u8,
+ messages_len: usize,
+ ) -> i32 {
+ let Some(instance) = INSTANCES.get(&id) else {
+ eprintln!(
+ "Sink connector with ID: {id} was not found and consume cannot be invoked."
+ );
+ return -1;
+ };
+ instance.consume(
+ topic_meta_ptr,
+ topic_meta_len,
+ messages_meta_ptr,
+ messages_meta_len,
+ messages_ptr,
+ messages_len,
+ )
+ }
+
+ #[unsafe(no_mangle)]
+ unsafe extern "C" fn close(id: u32) -> i32 {
+ let Some(mut instance) = INSTANCES.remove(&id) else {
+ eprintln!("Sink connector with ID: {id} was not found and cannot be closed.");
+ return -1;
+ };
+ instance.1.close()
+ }
+ };
+}
diff --git a/core/connectors/sdk/src/source.rs b/core/connectors/sdk/src/source.rs
new file mode 100644
index 00000000..ef47b7ee
--- /dev/null
+++ b/core/connectors/sdk/src/source.rs
@@ -0,0 +1,218 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use crate::{Error, Source, get_runtime};
+use serde::de::DeserializeOwned;
+use std::sync::Arc;
+use tokio::{sync::watch, task::JoinHandle};
+use tracing::{error, info};
+use tracing_subscriber::fmt;
+
+#[repr(C)]
+pub struct RawMessage {
+ pub offset: u64,
+ pub headers_ptr: *const u8,
+ pub headers_len: usize,
+ pub payload_ptr: *const u8,
+ pub payload_len: usize,
+}
+
+pub type HandleCallback = extern "C" fn(plugin_id: u32, callback: SendCallback) -> i32;
+
+pub type SendCallback = extern "C" fn(plugin_id: u32, messages_ptr: *const u8, messages_len: usize);
+
+#[derive(Debug)]
+pub struct SourceContainer<T: Source + std::fmt::Debug> {
+ id: u32,
+ source: Option<Arc<T>>,
+ shutdown: Option<watch::Sender<()>>,
+ task: Option<JoinHandle<()>>,
+}
+
+impl<T: Source + std::fmt::Debug + 'static> SourceContainer<T> {
+ pub const fn new(id: u32) -> Self {
+ Self {
+ id,
+ source: None,
+ shutdown: None,
+ task: None,
+ }
+ }
+
+ /// # Safety
+ /// This is safe
+ pub unsafe fn open<F, C>(
+ &mut self,
+ id: u32,
+ config_ptr: *const u8,
+ config_len: usize,
+ factory: F,
+ ) -> i32
+ where
+ F: FnOnce(u32, C) -> T,
+ C: DeserializeOwned,
+ {
+ unsafe {
+ _ = fmt::try_init();
+ let slice = std::slice::from_raw_parts(config_ptr, config_len);
+ let Ok(config_str) = std::str::from_utf8(slice) else {
+ return -1;
+ };
+
+ let Ok(config) = serde_json::from_str(config_str) else {
+ return -1;
+ };
+
+ let mut source = factory(id, config);
+ let runtime = get_runtime();
+ let result = runtime.block_on(source.open());
+ self.id = id;
+ self.source = Some(Arc::new(source));
+ if result.is_ok() { 0 } else { 1 }
+ }
+ }
+
+ /// # Safety
+ /// This is safe
+ pub unsafe fn close(&mut self) -> i32 {
+ let Some(source) = self.source.take() else {
+ error!(
+ "Source with ID: {} is not initialized - cannot close.",
+ self.id
+ );
+ return -1;
+ };
+
+ info!("Closing source with ID: {}...", self.id);
+ if let Some(sender) = self.shutdown.take() {
+ let _ = sender.send(());
+ }
+
+ let runtime = get_runtime();
+ if let Some(handle) = self.task.take() {
+ let _ = runtime.block_on(handle);
+ }
+
+ let Ok(mut source) = Arc::try_unwrap(source) else {
+ error!("Source with ID: {} was already closed.", self.id);
+ return -1;
+ };
+
+ runtime.block_on(source.close());
+ info!("Closed source with ID: {}", self.id);
+ 0
+ }
+
+ /// # Safety
+ /// This is safe
+ pub unsafe fn handle(&mut self, callback: SendCallback) -> i32 {
+ let Some(source) = self.source.as_ref() else {
+ error!(
+ "Source with ID: {} is not initialized - cannot handle.",
+ self.id
+ );
+ return -1;
+ };
+
+ let runtime = get_runtime();
+ let (shutdown_tx, shutdown_rx) = watch::channel(());
+ let plugin_id = self.id;
+ let source = Arc::clone(source);
+ let handle = runtime.spawn(async move {
+ let _ = handle_messages(plugin_id, source, callback, shutdown_rx).await;
+ });
+
+ self.shutdown = Some(shutdown_tx);
+ self.task = Some(handle);
+ 0
+ }
+}
+
+async fn handle_messages<T: Source>(
+ plugin_id: u32,
+ source: Arc<T>,
+ callback: SendCallback,
+ mut shutdown: watch::Receiver<()>,
+) -> Result<(), Error> {
+ loop {
+ tokio::select! {
+ _ = shutdown.changed() => {
+ info!("Shutting down source container with ID: {plugin_id}");
+ break;
+ }
+ messages = source.poll() => {
+ let Ok(messages) = messages else {
+ error!("Failed to poll messages for source container with ID: {plugin_id}");
+ continue;
+ };
+
+ let Ok(messages) = postcard::to_allocvec(&messages) else {
+ error!("Failed to serialize messages for source container with ID: {plugin_id}");
+ continue;
+ };
+
+ callback(plugin_id, messages.as_ptr(), messages.len());
+ }
+ }
+ }
+
+ Ok(())
+}
+
+#[macro_export]
+macro_rules! source_connector {
+ ($type:ty) => {
+ const _: fn() = || {
+ fn assert_trait<T: $crate::Source>() {}
+ assert_trait::<$type>();
+ };
+
+ use dashmap::DashMap;
+ use once_cell::sync::Lazy;
+ use $crate::source::SendCallback;
+ use $crate::source::SourceContainer;
+
+ static INSTANCES: Lazy<DashMap<u32, SourceContainer<$type>>> = Lazy::new(DashMap::new);
+
+ #[unsafe(no_mangle)]
+ unsafe extern "C" fn open(id: u32, config_ptr: *const u8, config_len: usize) -> i32 {
+ let mut container = SourceContainer::new(id);
+ let result = container.open(id, config_ptr, config_len, <$type>::new);
+ INSTANCES.insert(id, container);
+ result
+ }
+
+ #[unsafe(no_mangle)]
+ unsafe extern "C" fn handle(id: u32, callback: SendCallback) -> i32 {
+ let Some(mut instance) = INSTANCES.get_mut(&id) else {
+ eprintln!("Source connector with ID: {id} was not found and cannot be handled.");
+ return -1;
+ };
+ instance.handle(callback)
+ }
+
+ #[unsafe(no_mangle)]
+ unsafe extern "C" fn close(id: u32) -> i32 {
+ let Some(mut instance) = INSTANCES.remove(&id) else {
+ eprintln!("Source connector with ID: {id} was not found and cannot be closed.");
+ return -1;
+ };
+ instance.1.close()
+ }
+ };
+}
diff --git a/core/connectors/sdk/src/transforms/add_fields.rs b/core/connectors/sdk/src/transforms/add_fields.rs
new file mode 100644
index 00000000..569a16b8
--- /dev/null
+++ b/core/connectors/sdk/src/transforms/add_fields.rs
@@ -0,0 +1,128 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use serde::{Deserialize, Serialize};
+use simd_json::OwnedValue;
+use strum_macros::{Display, IntoStaticStr};
+
+use crate::{DecodedMessage, Error, Payload, TopicMetadata};
+
+use super::{Transform, TransformType};
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct AddFieldsConfig {
+ fields: Vec<Field>,
+}
+
+pub struct AddFields {
+ fields: Vec<Field>,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct Field {
+ key: String,
+ value: FieldValue,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+enum FieldValue {
+ Static(simd_json::OwnedValue),
+ Computed(ComputedValue),
+}
+
+#[derive(Debug, Copy, Clone, Eq, PartialEq, Serialize, Deserialize, Display, IntoStaticStr)]
+#[serde(rename_all = "snake_case")]
+enum ComputedValue {
+ #[strum(to_string = "date_time")]
+ DateTime,
+ #[strum(to_string = "timestamp_nanos")]
+ TimestampNanos,
+ #[strum(to_string = "timestamp_micros")]
+ TimestampMicros,
+ #[strum(to_string = "timestamp_millis")]
+ TimestampMillis,
+ #[strum(to_string = "timestamp_seconds")]
+ TimestampSeconds,
+ #[strum(to_string = "uuid_v4")]
+ UuidV4,
+ #[strum(to_string = "uuid_v7")]
+ UuidV7,
+}
+
+impl AddFields {
+ pub fn new(config: AddFieldsConfig) -> Self {
+ Self {
+ fields: config.fields,
+ }
+ }
+}
+
+impl Transform for AddFields {
+ fn r#type(&self) -> TransformType {
+ TransformType::AddFields
+ }
+
+ fn transform(
+ &self,
+ _metadata: &TopicMetadata,
+ mut message: DecodedMessage,
+ ) -> Result<Option<DecodedMessage>, Error> {
+ if self.fields.is_empty() {
+ return Ok(Some(message));
+ }
+
+ let Payload::Json(OwnedValue::Object(ref mut map)) = message.payload else {
+ return Ok(Some(message));
+ };
+
+ for field in &self.fields {
+ match &field.value {
+ FieldValue::Static(value) => map.insert(field.key.clone(), value.clone()),
+ FieldValue::Computed(value) => match value {
+ ComputedValue::DateTime => {
+ map.insert(field.key.clone(), chrono::Utc::now().to_rfc3339().into())
+ }
+ ComputedValue::TimestampMillis => map.insert(
+ field.key.clone(),
+ chrono::Utc::now().timestamp_millis().into(),
+ ),
+ ComputedValue::TimestampMicros => map.insert(
+ field.key.clone(),
+ chrono::Utc::now().timestamp_micros().into(),
+ ),
+ ComputedValue::TimestampNanos => map.insert(
+ field.key.clone(),
+ chrono::Utc::now().timestamp_nanos_opt().into(),
+ ),
+ ComputedValue::TimestampSeconds => {
+ map.insert(field.key.clone(), chrono::Utc::now().timestamp().into())
+ }
+ ComputedValue::UuidV4 => {
+ map.insert(field.key.clone(), uuid::Uuid::new_v4().to_string().into())
+ }
+ ComputedValue::UuidV7 => {
+ map.insert(field.key.clone(), uuid::Uuid::now_v7().to_string().into())
+ }
+ },
+ };
+ }
+
+ Ok(Some(message))
+ }
+}
diff --git a/core/connectors/sdk/src/transforms/delete_fields.rs b/core/connectors/sdk/src/transforms/delete_fields.rs
new file mode 100644
index 00000000..d7dd755a
--- /dev/null
+++ b/core/connectors/sdk/src/transforms/delete_fields.rs
@@ -0,0 +1,61 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use std::collections::HashSet;
+
+use serde::{Deserialize, Serialize};
+use simd_json::OwnedValue;
+
+use crate::{DecodedMessage, Error, Payload, TopicMetadata};
+
+use super::{Transform, TransformType};
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct DeleteFieldsConfig {
+ fields: Vec<String>,
+}
+
+pub struct DeleteFields {
+ fields: HashSet<String>,
+}
+
+impl DeleteFields {
+ pub fn new(config: DeleteFieldsConfig) -> Self {
+ Self {
+ fields: config.fields.into_iter().collect(),
+ }
+ }
+}
+
+impl Transform for DeleteFields {
+ fn r#type(&self) -> TransformType {
+ TransformType::DeleteFields
+ }
+
+ fn transform(
+ &self,
+ _metadata: &TopicMetadata,
+ mut message: DecodedMessage,
+ ) -> Result<Option<DecodedMessage>, Error> {
+ if let Payload::Json(OwnedValue::Object(ref mut map)) = message.payload {
+ map.retain(|key, _| !self.fields.contains(key));
+ }
+
+ Ok(Some(message))
+ }
+}
diff --git a/core/connectors/sdk/src/transforms/mod.rs b/core/connectors/sdk/src/transforms/mod.rs
new file mode 100644
index 00000000..35cf3382
--- /dev/null
+++ b/core/connectors/sdk/src/transforms/mod.rs
@@ -0,0 +1,45 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use serde::{Deserialize, Serialize};
+use strum_macros::{Display, IntoStaticStr};
+
+use crate::{DecodedMessage, Error, TopicMetadata};
+
+pub mod add_fields;
+pub mod delete_fields;
+
+pub trait Transform: Send + Sync {
+ fn r#type(&self) -> TransformType;
+ fn transform(
+ &self,
+ metadata: &TopicMetadata,
+ message: DecodedMessage,
+ ) -> Result<Option<DecodedMessage>, Error>;
+}
+
+#[derive(
+ Debug, Copy, Clone, Eq, PartialEq, Hash, Serialize, Deserialize, Display, IntoStaticStr,
+)]
+#[serde(rename_all = "snake_case")]
+pub enum TransformType {
+ #[strum(to_string = "add_fields")]
+ AddFields,
+ #[strum(to_string = "delete_fields")]
+ DeleteFields,
+}
diff --git a/core/connectors/sinks/quickwit_sink/Cargo.toml b/core/connectors/sinks/quickwit_sink/Cargo.toml
new file mode 100644
index 00000000..5848d24f
--- /dev/null
+++ b/core/connectors/sinks/quickwit_sink/Cargo.toml
@@ -0,0 +1,45 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[package]
+name = "iggy_connector_quickwit_sink"
+version = "0.1.0"
+description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
+edition = "2024"
+license = "Apache-2.0"
+keywords = ["iggy", "messaging", "streaming"]
+categories = ["command-line-utilities", "database", "network-programming"]
+homepage = "https://iggy.apache.org"
+documentation = "https://iggy.apache.org/docs"
+repository = "https://github.com/apache/iggy"
+readme = "../../README.md"
+
+[lib]
+crate-type = ["cdylib"]
+
+[dependencies]
+iggy_connector_sdk = { workspace = true }
+async-trait = { workspace = true }
+dashmap = { workspace = true }
+once_cell = { workspace = true }
+serde = { workspace = true }
+postcard = { workspace = true }
+reqwest = { workspace = true }
+serde_yml = { workspace = true }
+simd-json = { workspace = true }
+toml = { workspace = true }
+tracing = { workspace = true }
diff --git a/core/connectors/sinks/quickwit_sink/src/lib.rs b/core/connectors/sinks/quickwit_sink/src/lib.rs
new file mode 100644
index 00000000..cf0962cd
--- /dev/null
+++ b/core/connectors/sinks/quickwit_sink/src/lib.rs
@@ -0,0 +1,209 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use async_trait::async_trait;
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata, sink_connector,
+};
+use serde::{Deserialize, Serialize};
+use tracing::{error, info, warn};
+
+sink_connector!(QuickwitSink);
+
+#[derive(Debug)]
+pub struct QuickwitSink {
+ id: u32,
+ config: QuickwitSinkConfig,
+ client: reqwest::Client,
+ index_id: String,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct QuickwitSinkConfig {
+ url: String,
+ index: String,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct IndexConfig {
+ index_id: String,
+}
+
+impl QuickwitSink {
+ pub fn new(id: u32, config: QuickwitSinkConfig) -> Self {
+ let index_config =
+ serde_yml::from_str::<IndexConfig>(&config.index).expect("Invalid index config.");
+ QuickwitSink {
+ id,
+ config,
+ index_id: index_config.index_id,
+ client: reqwest::Client::new(),
+ }
+ }
+
+ async fn has_index(&self) -> Result<bool, Error> {
+ let url = format!("{}/api/v1/indexes/{}", self.config.url, self.index_id);
+ let response = self.client.get(&url).send().await.map_err(|error| {
+ error!(
+ "Failed to send HTTP request to check if index with ID: {} exists. {error}",
+ self.index_id
+ );
+ Error::HttpRequestFailed(error.to_string())
+ })?;
+ let status = response.status();
+ if status.is_success() {
+ Ok(true)
+ } else if status == reqwest::StatusCode::NOT_FOUND {
+ Ok(false)
+ } else {
+ Err(Error::HttpRequestFailed(format!(
+ "Unexpected status code: {status}",
+ )))
+ }
+ }
+
+ async fn create_index(&self) -> Result<(), Error> {
+ info!("Creating index: {}", self.index_id);
+ let url = format!("{}/api/v1/indexes", self.config.url);
+ let response = self
+ .client
+ .post(&url)
+ .header("content-type", "application/yaml")
+ .body(self.config.index.to_owned())
+ .send()
+ .await
+ .map_err(|error| {
+ error!(
+ "Failed to send HTTP request to create index: {}. {error}",
+ self.index_id
+ );
+ Error::HttpRequestFailed(error.to_string())
+ })?;
+
+ if !response.status().is_success() {
+ let status = response.status();
+ let reason = response.text().await.unwrap_or_default();
+ error!(
+ "Received an invalid HTTP response when creating index: {}. Status code: {status}, reason: {reason}",
+ self.index_id
+ );
+ return Err(Error::InitError(format!(
+ "Failed to create index: {}. {reason}",
+ self.index_id
+ )));
+ }
+
+ info!("Created index: {}", self.index_id);
+ Ok(())
+ }
+
+ pub async fn ingest(&self, messages: Vec<simd_json::OwnedValue>) -> Result<(), Error> {
+ let url = format!(
+ "{}/api/v1/{}/ingest?commit=auto",
+ self.config.url, self.index_id
+ );
+ info!("Ingesting messages for index: {}...", self.index_id);
+ let messages_count = messages.len();
+ let messages = messages
+ .into_iter()
+ .filter_map(|record| simd_json::to_string(&record).ok())
+ .collect::<Vec<_>>()
+ .join("\n");
+
+ let response = self
+ .client
+ .post(&url)
+ .body(messages)
+ .send()
+ .await
+ .map_err(|error| {
+ error!(
+ "Failed to send HTTP request to ingest messages for index: {}. {error}",
+ self.index_id
+ );
+ Error::HttpRequestFailed(error.to_string())
+ })?;
+
+ if !response.status().is_success() {
+ let status = response.status();
+ let text = response.text().await.unwrap_or_default();
+ error!(
+ "Received an invalid HTTP response when ingesting messages for index: {}. Status code: {status}, reason: {text}",
+ self.index_id
+ );
+ return Err(Error::HttpRequestFailed(format!(
+ "Status code: {status}, reason: {text}"
+ )));
+ }
+
+ info!(
+ "Ingested {messages_count} messages for index: {}",
+ self.index_id
+ );
+ Ok(())
+ }
+}
+
+#[async_trait]
+impl Sink for QuickwitSink {
+ async fn open(&mut self) -> Result<(), Error> {
+ info!(
+ "Initialized QuickwitSink with ID: {} for URL: {}",
+ self.id, self.config.url
+ );
+ if !self.has_index().await? {
+ self.create_index().await?;
+ }
+ Ok(())
+ }
+
+ async fn consume(
+ &self,
+ _topic_metadata: &TopicMetadata,
+ messages_metadata: MessagesMetadata,
+ messages: Vec<ConsumedMessage>,
+ ) -> Result<(), Error> {
+ info!(
+ "Quickwit sink with ID: {} received: {} messages, format: {}",
+ self.id,
+ messages.len(),
+ messages_metadata.schema
+ );
+
+ let mut json_payloads = Vec::with_capacity(messages.len());
+ for message in messages {
+ match message.payload {
+ Payload::Json(value) => json_payloads.push(value),
+ _ => {
+ warn!("Unsupported payload format: {}", messages_metadata.schema);
+ }
+ }
+ }
+
+ if json_payloads.is_empty() {
+ return Ok(());
+ }
+
+ self.ingest(json_payloads).await?;
+ Ok(())
+ }
+
+ async fn close(&mut self) {
+ info!("Quickwit sink with ID: {} is shutting down", self.id);
+ }
+}
diff --git a/core/connectors/sinks/stdout_sink/Cargo.toml b/core/connectors/sinks/stdout_sink/Cargo.toml
new file mode 100644
index 00000000..8e834833
--- /dev/null
+++ b/core/connectors/sinks/stdout_sink/Cargo.toml
@@ -0,0 +1,42 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[package]
+name = "iggy_connector_stdout_sink"
+version = "0.1.0"
+description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
+edition = "2024"
+license = "Apache-2.0"
+keywords = ["iggy", "messaging", "streaming"]
+categories = ["command-line-utilities", "database", "network-programming"]
+homepage = "https://iggy.apache.org"
+documentation = "https://iggy.apache.org/docs"
+repository = "https://github.com/apache/iggy"
+readme = "../../README.md"
+
+[lib]
+crate-type = ["cdylib"]
+
+[dependencies]
+iggy_connector_sdk = { workspace = true }
+async-trait = { workspace = true }
+dashmap = { workspace = true }
+once_cell = { workspace = true }
+serde = { workspace = true }
+postcard = { workspace = true }
+toml = { workspace = true }
+tracing = { workspace = true }
diff --git a/core/connectors/sinks/stdout_sink/src/lib.rs b/core/connectors/sinks/stdout_sink/src/lib.rs
new file mode 100644
index 00000000..afa9df59
--- /dev/null
+++ b/core/connectors/sinks/stdout_sink/src/lib.rs
@@ -0,0 +1,88 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use async_trait::async_trait;
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector,
+};
+use serde::{Deserialize, Serialize};
+use tracing::info;
+
+sink_connector!(StdoutSink);
+
+#[derive(Debug)]
+pub struct StdoutSink {
+ id: u32,
+ print_payload: bool,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct StdoutSinkConfig {
+ print_payload: Option<bool>,
+}
+
+impl StdoutSink {
+ pub fn new(id: u32, config: StdoutSinkConfig) -> Self {
+ StdoutSink {
+ id,
+ print_payload: config.print_payload.unwrap_or(false),
+ }
+ }
+}
+
+#[async_trait]
+impl Sink for StdoutSink {
+ async fn open(&mut self) -> Result<(), Error> {
+ info!(
+ "Initialized stdout sink with ID: {}. Print payload: {}",
+ self.id, self.print_payload
+ );
+ Ok(())
+ }
+
+ async fn consume(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: MessagesMetadata,
+ messages: Vec<ConsumedMessage>,
+ ) -> Result<(), Error> {
+ info!(
+ "Stdout sink with ID: {} received: {} messages, schema: {}, stream: {}, topic: {}, partition: {}, offset: {}",
+ self.id,
+ messages.len(),
+ messages_metadata.schema,
+ topic_metadata.stream,
+ topic_metadata.stream,
+ messages_metadata.partition_id,
+ messages_metadata.current_offset
+ );
+ if self.print_payload {
+ for message in messages {
+ info!(
+ "Message offset: {}, payload: {:#?}",
+ message.offset, message.payload
+ );
+ }
+ }
+ Ok(())
+ }
+
+ async fn close(&mut self) {
+ info!("Stdout sink with ID: {} is shutting down", self.id);
+ }
+}
diff --git a/core/connectors/sources/test_source/Cargo.toml b/core/connectors/sources/test_source/Cargo.toml
new file mode 100644
index 00000000..f0eb76c8
--- /dev/null
+++ b/core/connectors/sources/test_source/Cargo.toml
@@ -0,0 +1,47 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[package]
+name = "iggy_connector_test_source"
+version = "0.1.0"
+description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
+edition = "2024"
+license = "Apache-2.0"
+keywords = ["iggy", "messaging", "streaming"]
+categories = ["command-line-utilities", "database", "network-programming"]
+homepage = "https://iggy.apache.org"
+documentation = "https://iggy.apache.org/docs"
+repository = "https://github.com/apache/iggy"
+readme = "../../README.md"
+
+[lib]
+crate-type = ["cdylib"]
+
+[dependencies]
+iggy_connector_sdk = { workspace = true }
+async-trait = { workspace = true }
+dashmap = { workspace = true }
+once_cell = { workspace = true }
+flume = { workspace = true }
+humantime = { workspace = true }
+serde = { workspace = true }
+rand = { workspace = true }
+postcard = { workspace = true }
+simd-json = { workspace = true }
+toml = { workspace = true }
+tracing = { workspace = true }
+tokio = { workspace = true }
diff --git a/core/connectors/sources/test_source/src/lib.rs b/core/connectors/sources/test_source/src/lib.rs
new file mode 100644
index 00000000..9d8c4cf1
--- /dev/null
+++ b/core/connectors/sources/test_source/src/lib.rs
@@ -0,0 +1,157 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use std::{str::FromStr, time::Duration};
+
+use async_trait::async_trait;
+use iggy_connector_sdk::{ProducedMessage, ProducedMessages, Schema, Source, source_connector};
+use rand::{
+ Rng,
+ distr::{Alphanumeric, Uniform},
+};
+use serde::{Deserialize, Serialize};
+use tokio::{sync::Mutex, time::sleep};
+use tracing::info;
+
+source_connector!(TestSource);
+
+#[derive(Debug)]
+pub struct TestSource {
+ id: u32,
+ max_count: Option<usize>,
+ interval: Duration,
+ messages_range: (u32, u32),
+ payload_size: u32,
+ state: Mutex<State>,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct TestSourceConfig {
+ interval: Option<String>,
+ max_count: Option<usize>,
+ messages_range: Option<(u32, u32)>,
+ payload_size: Option<u32>,
+}
+
+#[derive(Debug)]
+struct State {
+ current_number: usize,
+}
+
+impl TestSource {
+ pub fn new(id: u32, config: TestSourceConfig) -> Self {
+ let interval = config.interval.unwrap_or("1s".to_string());
+ let interval = humantime::Duration::from_str(&interval)
+ .unwrap_or(humantime::Duration::from_str("1s").unwrap());
+ TestSource {
+ id,
+ max_count: config.max_count,
+ interval: *interval,
+ messages_range: config.messages_range.unwrap_or((10, 50)),
+ payload_size: config.payload_size.unwrap_or(100),
+ state: Mutex::new(State { current_number: 0 }),
+ }
+ }
+
+ fn generate_messages(&self) -> Vec<ProducedMessage> {
+ let mut messages = Vec::new();
+ let mut rng = rand::rng();
+ let messages_count =
+ rng.sample(Uniform::new(self.messages_range.0, self.messages_range.1).unwrap());
+ for _ in 0..messages_count {
+ let record = Record {
+ title: "Hello".to_string(),
+ name: "World".to_string(),
+ age: 30,
+ text: self.generate_random_text(),
+ };
+ let message = ProducedMessage {
+ id: None,
+ headers: None,
+ payload: simd_json::to_vec(&record).unwrap(),
+ };
+ messages.push(message);
+ }
+ messages
+ }
+
+ fn generate_random_text(&self) -> String {
+ let mut rng = rand::rng();
+ let text: String = (0..self.payload_size)
+ .map(|_| rng.sample(Alphanumeric) as char)
+ .collect();
+ text
+ }
+}
+
+#[async_trait]
+impl Source for TestSource {
+ async fn open(&mut self) -> Result<(), iggy_connector_sdk::Error> {
+ info!(
+ "Initialized test source with ID {}. Interval: {:#?}, max offset: {:#?}, messages range: {} - {}, payload size: {}",
+ self.id,
+ self.interval,
+ self.max_count,
+ self.messages_range.0,
+ self.messages_range.1,
+ self.payload_size
+ );
+ Ok(())
+ }
+
+ async fn poll(&self) -> Result<ProducedMessages, iggy_connector_sdk::Error> {
+ sleep(self.interval).await;
+ let mut state = self.state.lock().await;
+ if let Some(max_count) = self.max_count {
+ if state.current_number >= max_count {
+ info!(
+ "Reached max number of {max_count} messages for test source with ID {}",
+ self.id
+ );
+ return Ok(ProducedMessages {
+ schema: Schema::Json,
+ messages: vec![],
+ });
+ }
+ }
+
+ let messages = self.generate_messages();
+ state.current_number += messages.len();
+ info!(
+ "Generated {} messages by test source with ID {}",
+ messages.len(),
+ self.id
+ );
+ Ok(ProducedMessages {
+ schema: Schema::Json,
+ messages,
+ })
+ }
+
+ async fn close(&mut self) {
+ info!("Test source with ID {} is shutting down", self.id);
+ }
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct Record {
+ title: String,
+ name: String,
+ age: u32,
+ text: String,
+}