feat(connectors): add the HTTP source webhook gateway connector

Cargo.toml

@@ -48,6 +48,7 @@ members = [
     "core/connectors/sinks/stdout_sink",
     "core/connectors/sinks/surrealdb_sink",
     "core/connectors/sources/elasticsearch_source",
+    "core/connectors/sources/http_source",
     "core/connectors/sources/influxdb_source",
     "core/connectors/sources/postgres_source",
     "core/connectors/sources/random_source",
@@ -86,6 +87,7 @@ aligned-vec = "0.6.4"
 anyhow = "1.0.104"
 apache-avro = "0.21.0"
 apple-native-keyring-store = { version = "1.0.1", features = ["keychain"] }
+arc-swap = "1.7.1"
 # "std" is load-bearing: it cascades to rand_core/getrandom, which the
 # crypto module's `OsRng` import needs even when no sibling crate in the
 # build graph happens to enable it via feature unification.
@@ -190,6 +192,7 @@ gloo = "0.12"
 governor = "0.10.4"
 harness_derive = { path = "core/harness_derive" }
 hash32 = "1.0.0"
+hex = "0.4.3"
 hostname = "0.4.2"
 http = "1.4.2"
 human-repr = "1.1.0"

core/connectors/runtime/example_config/connectors/http_source_github.toml

@@ -0,0 +1,60 @@
+# 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.
+
+# Instance 1 of a shared webhook listener. Pair it with
+# http_source_partner.toml to see two instances serving one port.
+#
+# The public listener binds loopback here so running the examples cannot
+# expose a port; production wants 0.0.0.0 behind a load balancer.
+
+type = "source"
+key = "http_github"
+enabled = true
+version = 0
+name = "HTTP source (GitHub)"
+path = "<BASE_DIR>/target/release/libiggy_connector_http_source"
+plugin_config_format = "toml"
+verbose = false
+
+[[streams]]
+stream = "webhooks"
+topic = "github_events"
+schema = "raw"
+batch_length = 100
+linger_time = "5ms"
+
+[plugin_config]
+listen_addr = "127.0.0.1:9090"
+admin_listen_addr = "127.0.0.1:9091"
+instance_name = "http_github"
+topic_path = "github_events"
+auth_bearer_token = "replace_with_secret_token"
+management_token = "replace_with_secret_token"
+max_body_size_bytes = 1048576
+buffer_capacity = 10000
+max_batch_size = 500
+include_http_metadata = true
+forward_headers = ["X-GitHub-Delivery", "X-Request-ID"]
+
+[[plugin_config.endpoints]]
+# Replace this: it is published in this repository, so anyone can reach it.
+# Generate your own with `openssl rand -hex 16`.
+endpoint_id = "a3f8c2e1b9d04f7a8e6c1d2b3a4f5e6d"
+auth_type = "hmac-sha256"
+auth_secret = "replace_with_webhook_signing_secret"
+hmac_header = "X-Hub-Signature-256"
+hmac_prefix = "sha256="

core/connectors/runtime/example_config/connectors/http_source_partner.toml

@@ -0,0 +1,55 @@
+# 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.
+
+# Instance 2, joining the listener bound by http_source_github.toml.
+# listen_addr, admin_listen_addr, max_body_size_bytes, and management_token
+# must match that file exactly, or this instance's open fails.
+#
+# No topic_path: this instance serves secret-path endpoints only.
+
+type = "source"
+key = "http_partner"
+enabled = true
+version = 0
+name = "HTTP source (partner)"
+path = "<BASE_DIR>/target/release/libiggy_connector_http_source"
+plugin_config_format = "toml"
+verbose = false
+
+[[streams]]
+stream = "webhooks"
+topic = "partner_events"
+schema = "raw"
+batch_length = 100
+linger_time = "5ms"
+
+[plugin_config]
+listen_addr = "127.0.0.1:9090"
+admin_listen_addr = "127.0.0.1:9091"
+instance_name = "http_partner"
+management_token = "replace_with_secret_token"
+max_body_size_bytes = 1048576
+buffer_capacity = 10000
+max_batch_size = 500
+include_http_metadata = true
+
+[[plugin_config.endpoints]]
+# Replace this: it is published in this repository, so anyone can reach it.
+# Generate your own with `openssl rand -hex 16`.
+endpoint_id = "0b7d9e2f4a6c8e1d3b5f7a9c2e4d6f81"
+auth_type = "bearer"
+auth_secret = "replace_with_partner_token"

core/connectors/sources/README.md

@@ -9,6 +9,7 @@ Source connectors are responsible for ingesting data from external sources into
 | Source | Description |
 | ------ | ----------- |
 | **elasticsearch_source** | Polls documents from Elasticsearch indices with timestamp-based tracking |
+| **http_source** | Webhook gateway: an embedded HTTP server shared by every instance, with per-endpoint bearer/HMAC auth and a management API for endpoints registered at runtime |
 | **influxdb_source** | Polls InfluxDB with cursor-based timestamp tracking; supports V2 (Flux, annotated CSV) and V3 (SQL, JSONL) |
 | **postgres_source** | Reads rows from PostgreSQL tables with multiple strategies: delete after read, mark as processed, or timestamp tracking |
 | **random_source** | Generates random test messages (useful for testing and development) |
@@ -69,7 +70,7 @@ enabled = true # Toggle source on/off
 version = 0
 name = "Random source" # Name of the source
 path = "libiggy_connector_random_source" # Path to the source connector
-config_format = "toml"
+plugin_config_format = "toml"
 verbose = false # Log message processing at info level instead of debug
 benchmark = false # Emit per-batch timing events on `iggy_connectors::benchmark` target
 

core/connectors/sources/http_source/Cargo.toml

@@ -0,0 +1,59 @@
+# 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_http_source"
+version = "0.4.1-edge.1"
+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"
+publish = false
+
+[package.metadata.cargo-machete]
+ignored = ["dashmap"]
+
+[lib]
+crate-type = ["cdylib", "lib"]
+
+[dependencies]
+arc-swap = { workspace = true }
+async-trait = { workspace = true }
+axum = { workspace = true }
+crossfire = { workspace = true }
+dashmap = { workspace = true }
+hex = { workspace = true }
+iggy_common = { workspace = true }
+iggy_connector_sdk = { workspace = true }
+prometheus-client = { workspace = true }
+rand = { workspace = true }
+ring = { workspace = true }
+secrecy = { workspace = true }
+serde = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }
+
+[dev-dependencies]
+reqwest = { workspace = true }
+rmp-serde = { workspace = true }
+serde_json = { workspace = true }
+toml = { workspace = true }