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/09/19 19:22:43 UTC
(iggy) 02/02: Fix connectors config provider
This is an automated email from the ASF dual-hosted git repository.
piotr pushed a commit to branch connectors_test
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 17cf5224576414190b9bb84c1be25d81148843f4
Author: spetz <pi...@gmail.com>
AuthorDate: Fri Sep 19 21:22:35 2025 +0200
Fix connectors config provider
---
core/connectors/runtime/config.toml | 6 +++---
core/connectors/runtime/src/api/mod.rs | 6 +++---
core/connectors/runtime/src/configs.rs | 12 ++++++++++--
core/connectors/runtime/src/context.rs | 2 +-
core/connectors/runtime/src/main.rs | 2 +-
core/integration/tests/connectors/mod.rs | 3 +--
6 files changed, 19 insertions(+), 12 deletions(-)
diff --git a/core/connectors/runtime/config.toml b/core/connectors/runtime/config.toml
index ea868f8e..3e00608c 100644
--- a/core/connectors/runtime/config.toml
+++ b/core/connectors/runtime/config.toml
@@ -15,12 +15,12 @@
# specific language governing permissions and limitations
# under the License.
-[http_api] # Optional HTTP API configuration
+[http] # Optional HTTP API configuration
enabled = true
address = "127.0.0.1:8081"
# api_key = "secret" # Optional API key for authentication to be passed as `api-key` header
-[http_api.cors] # Optional CORS configuration for HTTP API
+[http.cors] # Optional CORS configuration for HTTP API
enabled = false
allowed_methods = ["GET", "POST", "PUT", "DELETE"]
allowed_origins = ["*"]
@@ -29,7 +29,7 @@ exposed_headers = [""]
allow_credentials = false
allow_private_network = false
-[http_api.tls] # Optional TLS configuration for HTTP API
+[http.tls] # Optional TLS configuration for HTTP API
enabled = false
cert_file = "core/certs/iggy_cert.pem"
key_file = "core/certs/iggy_key.pem"
diff --git a/core/connectors/runtime/src/api/mod.rs b/core/connectors/runtime/src/api/mod.rs
index c6a2ebc7..6cb09dbd 100644
--- a/core/connectors/runtime/src/api/mod.rs
+++ b/core/connectors/runtime/src/api/mod.rs
@@ -72,7 +72,7 @@ pub async fn init(config: &HttpApiConfig, context: Arc<RuntimeContext>) {
let address = listener
.local_addr()
.expect("Failed to get local address for HTTP server");
- info!("Started {NAME} on: {address}");
+ info!("Started {NAME} HTTP API on: {address}");
spawn(async move {
if let Err(error) = axum::serve(
listener,
@@ -80,7 +80,7 @@ pub async fn init(config: &HttpApiConfig, context: Arc<RuntimeContext>) {
)
.await
{
- error!("Failed to start {NAME} server, error: {error}");
+ error!("Failed to start {NAME} HTTP API, error: {error}");
}
});
return;
@@ -104,7 +104,7 @@ pub async fn init(config: &HttpApiConfig, context: Arc<RuntimeContext>) {
.serve(app.into_make_service_with_connect_info::<SocketAddr>())
.await
{
- error!("Failed to start {NAME} server, error: {error}");
+ error!("Failed to start {NAME} HTTP API, error: {error}");
}
});
}
diff --git a/core/connectors/runtime/src/configs.rs b/core/connectors/runtime/src/configs.rs
index 44834745..19a2b4dc 100644
--- a/core/connectors/runtime/src/configs.rs
+++ b/core/connectors/runtime/src/configs.rs
@@ -42,7 +42,7 @@ pub enum ConfigFormat {
#[derive(Debug, Default, Clone, Deserialize, Serialize)]
#[serde(default)]
pub struct RuntimeConfig {
- pub http_api: HttpApiConfig,
+ pub http: HttpApiConfig,
pub iggy: IggyConfig,
pub sinks: HashMap<String, SinkConfig>,
pub sources: HashMap<String, SourceConfig>,
@@ -109,11 +109,19 @@ pub struct SharedTransformConfig {
pub enabled: bool,
}
-#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StateConfig {
pub path: String,
}
+impl Default for StateConfig {
+ fn default() -> Self {
+ Self {
+ path: "local_state".to_owned(),
+ }
+ }
+}
+
impl Default for IggyConfig {
fn default() -> Self {
Self {
diff --git a/core/connectors/runtime/src/context.rs b/core/connectors/runtime/src/context.rs
index 6dd005b6..d7dea9f8 100644
--- a/core/connectors/runtime/src/context.rs
+++ b/core/connectors/runtime/src/context.rs
@@ -40,7 +40,7 @@ pub fn init(
RuntimeContext {
sinks: SinkManager::new(map_sinks(config, sink_wrappers)),
sources: SourceManager::new(map_sources(config, source_wrappers)),
- api_key: config.http_api.api_key.clone(),
+ api_key: config.http.api_key.clone(),
}
}
diff --git a/core/connectors/runtime/src/main.rs b/core/connectors/runtime/src/main.rs
index 9968e807..7b5d88a5 100644
--- a/core/connectors/runtime/src/main.rs
+++ b/core/connectors/runtime/src/main.rs
@@ -169,7 +169,7 @@ async fn main() -> Result<(), RuntimeError> {
let context = context::init(&config, &sink_wrappers, &source_wrappers);
let context = Arc::new(context);
- api::init(&config.http_api, context).await;
+ api::init(&config.http, context).await;
source::handle(source_wrappers);
sink::consume(sink_wrappers);
diff --git a/core/integration/tests/connectors/mod.rs b/core/integration/tests/connectors/mod.rs
index 4df65ac7..2689cdae 100644
--- a/core/integration/tests/connectors/mod.rs
+++ b/core/integration/tests/connectors/mod.rs
@@ -22,12 +22,11 @@ use integration::{
};
use serial_test::parallel;
use std::collections::HashMap;
-use tokio::time::timeout;
#[tokio::test]
#[parallel]
async fn connectors_runtime_should_start() {
- let _ = timeout(std::time::Duration::from_secs(3), setup()).await;
+ setup().await;
}
async fn setup() -> ConnectorsInfra {