You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by pi...@apache.org on 2025/06/10 18:36:32 UTC
(iggy) branch connectors_api updated (b678ecbc -> e15e23a0)
This is an automated email from the ASF dual-hosted git repository.
piotr pushed a change to branch connectors_api
in repository https://gitbox.apache.org/repos/asf/iggy.git
discard b678ecbc add initial version of connectors http api
new e15e23a0 add initial version of connectors http api
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (b678ecbc)
\
N -- N -- N refs/heads/connectors_api (e15e23a0)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
core/connectors/runtime/src/context.rs | 18 ++++++++++++++++++
core/connectors/runtime/src/manager/mod.rs | 18 ++++++++++++++++++
core/connectors/runtime/src/manager/sink.rs | 19 ++++++++++++++++++-
core/connectors/runtime/src/manager/source.rs | 19 ++++++++++++++++++-
4 files changed, 72 insertions(+), 2 deletions(-)
(iggy) 01/01: add initial version of connectors http api
Posted by pi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
piotr pushed a commit to branch connectors_api
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit e15e23a014a897a9a74de28660bab8c89eb21a98
Author: spetz <pi...@gmail.com>
AuthorDate: Tue Jun 10 20:31:04 2025 +0200
add initial version of connectors http api
---
Cargo.lock | 1 +
core/connectors/runtime/Cargo.toml | 1 +
core/connectors/runtime/config.toml | 4 +
core/connectors/runtime/src/api.rs | 228 ++++++++++++++++++++++++++
core/connectors/runtime/src/configs.rs | 19 ++-
core/connectors/runtime/src/context.rs | 88 ++++++++++
core/connectors/runtime/src/error.rs | 18 ++
core/connectors/runtime/src/main.rs | 26 ++-
core/connectors/runtime/src/manager/mod.rs | 20 +++
core/connectors/runtime/src/manager/sink.rs | 79 +++++++++
core/connectors/runtime/src/manager/source.rs | 79 +++++++++
core/connectors/runtime/src/sink.rs | 6 +
core/connectors/runtime/src/source.rs | 6 +
13 files changed, 561 insertions(+), 14 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 6be9aa34..7e021dd6 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3857,6 +3857,7 @@ dependencies = [
name = "iggy_connector_runtime"
version = "0.1.0"
dependencies = [
+ "axum 0.8.4",
"config",
"dashmap",
"dlopen2",
diff --git a/core/connectors/runtime/Cargo.toml b/core/connectors/runtime/Cargo.toml
index a975d0ae..75121533 100644
--- a/core/connectors/runtime/Cargo.toml
+++ b/core/connectors/runtime/Cargo.toml
@@ -29,6 +29,7 @@ repository = "https://github.com/apache/iggy"
readme = "../../README.md"
[dependencies]
+axum = { workspace = true }
config = { workspace = true }
dashmap = { workspace = true }
dlopen2 = { workspace = true }
diff --git a/core/connectors/runtime/config.toml b/core/connectors/runtime/config.toml
index d7361472..f9d5e14d 100644
--- a/core/connectors/runtime/config.toml
+++ b/core/connectors/runtime/config.toml
@@ -15,6 +15,10 @@
# specific language governing permissions and limitations
# under the License.
+[http_api]
+enabled = true
+address = "127.0.0.1:8081"
+
[iggy]
address = "localhost:8090"
username = "iggy"
diff --git a/core/connectors/runtime/src/api.rs b/core/connectors/runtime/src/api.rs
new file mode 100644
index 00000000..5c72b8f9
--- /dev/null
+++ b/core/connectors/runtime/src/api.rs
@@ -0,0 +1,228 @@
+/* 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 axum::{
+ Json, Router,
+ extract::{Path, State},
+ http::StatusCode,
+ response::IntoResponse,
+ routing::get,
+};
+use serde::{Deserialize, Serialize};
+use thiserror::Error;
+use tokio::{net::TcpListener, spawn};
+use tracing::{error, info};
+
+use crate::{
+ context::RuntimeContext,
+ error::RuntimeError,
+ manager::{
+ sink::{SinkDetailsResponse, SinkInfoResponse},
+ source::{SourceDetailsResponse, SourceInfoResponse},
+ },
+};
+
+const NAME: &str = env!("CARGO_PKG_NAME");
+
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct HttpApiConfig {
+ pub enabled: bool,
+ pub address: String,
+}
+
+pub async fn init(config: &HttpApiConfig, context: Arc<RuntimeContext>) {
+ if !config.enabled {
+ info!("HTTP API is disabled");
+ return;
+ }
+
+ let address = config.address.to_owned();
+ let app = Router::new()
+ .route("/", get(|| async { "Connector Runtime API" }))
+ .route(
+ "/health",
+ get(|| async { Json(serde_json::json!({ "status": "healthy" })) }),
+ )
+ .route("/sinks", get(get_sinks))
+ .route("/sinks/{key}", get(get_sink))
+ .route("/sinks/{key}/config", get(get_sink_config))
+ .route("/sources", get(get_sources))
+ .route("/sources/{key}", get(get_source))
+ .route("/sources/{key}/config", get(get_source_config))
+ .with_state(context);
+
+ let Ok(listener) = TcpListener::bind(&address).await else {
+ panic!("Failed to bind to HTTP API address: {address}");
+ };
+
+ spawn(async move {
+ info!("Started {NAME} HTTP API on: {address}");
+ if let Err(error) = axum::serve(listener, app.into_make_service()).await {
+ panic!("Failed to start {NAME} HTTP API on: {address}. {error}");
+ }
+ });
+}
+
+async fn get_sinks(
+ State(context): State<Arc<RuntimeContext>>,
+) -> Result<Json<Vec<SinkInfoResponse>>, ApiError> {
+ let sinks = context
+ .sinks
+ .get_all()
+ .into_iter()
+ .map(|(key, sink)| SinkInfoResponse {
+ id: sink.id,
+ key: key.to_owned(),
+ name: sink.name.to_owned(),
+ path: sink.path.to_owned(),
+ enabled: sink.enabled,
+ running: sink.running,
+ })
+ .collect::<Vec<_>>();
+ Ok(Json(sinks))
+}
+
+async fn get_sink(
+ State(context): State<Arc<RuntimeContext>>,
+ Path(key): Path<String>,
+) -> Result<Json<SinkDetailsResponse>, ApiError> {
+ let Some(sink) = context.sinks.get(&key) else {
+ return Err(ApiError::Error(RuntimeError::SinkNotFound(key)));
+ };
+ let sink = SinkDetailsResponse {
+ info: SinkInfoResponse {
+ id: sink.info.id,
+ key: sink.key.to_owned(),
+ name: sink.info.name.to_owned(),
+ path: sink.info.path.to_owned(),
+ enabled: sink.info.enabled,
+ running: sink.info.running,
+ },
+ };
+ Ok(Json(sink))
+}
+
+async fn get_sink_config(
+ State(context): State<Arc<RuntimeContext>>,
+ Path(key): Path<String>,
+) -> Result<Json<Option<serde_json::Value>>, ApiError> {
+ let Some(sink) = context.sinks.get(&key) else {
+ return Err(ApiError::Error(RuntimeError::SinkNotFound(key)));
+ };
+ Ok(Json(sink.config.clone()))
+}
+
+async fn get_sources(
+ State(context): State<Arc<RuntimeContext>>,
+) -> Result<Json<Vec<SourceInfoResponse>>, ApiError> {
+ let sources = context
+ .sources
+ .get_all()
+ .into_iter()
+ .map(|(key, source)| SourceInfoResponse {
+ id: source.id,
+ key: key.to_owned(),
+ name: source.name.to_owned(),
+ path: source.path.to_owned(),
+ enabled: source.enabled,
+ running: source.running,
+ })
+ .collect::<Vec<_>>();
+ Ok(Json(sources))
+}
+
+async fn get_source(
+ State(context): State<Arc<RuntimeContext>>,
+ Path(key): Path<String>,
+) -> Result<Json<SourceDetailsResponse>, ApiError> {
+ let Some(source) = context.sources.get(&key) else {
+ return Err(ApiError::Error(RuntimeError::SourceNotFound(key)));
+ };
+ let source = SourceDetailsResponse {
+ info: SourceInfoResponse {
+ id: source.info.id,
+ key: source.key.to_owned(),
+ name: source.info.name.to_owned(),
+ path: source.info.path.to_owned(),
+ enabled: source.info.enabled,
+ running: source.info.running,
+ },
+ };
+ Ok(Json(source))
+}
+
+async fn get_source_config(
+ State(context): State<Arc<RuntimeContext>>,
+ Path(key): Path<String>,
+) -> Result<Json<Option<serde_json::Value>>, ApiError> {
+ let Some(sink) = context.sources.get(&key) else {
+ return Err(ApiError::Error(RuntimeError::SourceNotFound(key)));
+ };
+ Ok(Json(sink.config.clone()))
+}
+
+#[derive(Debug, Error)]
+enum ApiError {
+ #[error(transparent)]
+ Error(#[from] RuntimeError),
+ #[error(transparent)]
+ JsonError(#[from] serde_json::Error),
+}
+
+#[derive(Debug, Serialize)]
+pub struct ErrorResponse {
+ pub code: String,
+ pub reason: String,
+}
+
+impl IntoResponse for ApiError {
+ fn into_response(self) -> axum::response::Response {
+ match self {
+ ApiError::Error(error) => {
+ error!("There was an error: {error}");
+ let status_code = match error {
+ RuntimeError::MissingIggyCredentials => StatusCode::BAD_REQUEST,
+ RuntimeError::InvalidConfiguration(_) => StatusCode::BAD_REQUEST,
+ RuntimeError::SinkNotFound(_) => StatusCode::NOT_FOUND,
+ RuntimeError::SourceNotFound(_) => StatusCode::NOT_FOUND,
+ _ => StatusCode::INTERNAL_SERVER_ERROR,
+ };
+ (
+ status_code,
+ Json(ErrorResponse {
+ code: error.as_code().to_owned(),
+ reason: error.to_string(),
+ }),
+ )
+ }
+ ApiError::JsonError(error) => {
+ error!("There was a JSON error: {error}");
+ (
+ StatusCode::INTERNAL_SERVER_ERROR,
+ Json(ErrorResponse {
+ code: "json_error".to_owned(),
+ reason: error.to_string(),
+ }),
+ )
+ }
+ }
+ .into_response()
+ }
+}
diff --git a/core/connectors/runtime/src/configs.rs b/core/connectors/runtime/src/configs.rs
index aa6d8fa8..b1ecb504 100644
--- a/core/connectors/runtime/src/configs.rs
+++ b/core/connectors/runtime/src/configs.rs
@@ -20,14 +20,17 @@ use iggy_connector_sdk::{Schema, transforms::TransformType};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
-#[derive(Debug, Deserialize, Serialize)]
+use crate::api::HttpApiConfig;
+
+#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct RuntimeConfig {
+ pub http_api: HttpApiConfig,
pub iggy: IggyConfig,
pub sinks: HashMap<String, SinkConfig>,
pub sources: HashMap<String, SourceConfig>,
}
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct IggyConfig {
pub address: String,
pub username: Option<String>,
@@ -35,7 +38,7 @@ pub struct IggyConfig {
pub token: Option<String>,
}
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SinkConfig {
pub enabled: bool,
pub name: String,
@@ -45,7 +48,7 @@ pub struct SinkConfig {
pub config: Option<serde_json::Value>,
}
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StreamConsumerConfig {
pub stream: String,
pub topics: Vec<String>,
@@ -55,7 +58,7 @@ pub struct StreamConsumerConfig {
pub consumer_group: Option<String>,
}
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StreamProducerConfig {
pub stream: String,
pub topic: String,
@@ -64,7 +67,7 @@ pub struct StreamProducerConfig {
pub send_interval: Option<String>,
}
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SourceConfig {
pub enabled: bool,
pub name: String,
@@ -74,13 +77,13 @@ pub struct SourceConfig {
pub config: Option<serde_json::Value>,
}
-#[derive(Debug, Default, Serialize, Deserialize)]
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct TransformsConfig {
#[serde(flatten)]
pub transforms: HashMap<TransformType, serde_json::Value>,
}
-#[derive(Debug, Default, Serialize, Deserialize)]
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct SharedTransformConfig {
pub enabled: bool,
}
diff --git a/core/connectors/runtime/src/context.rs b/core/connectors/runtime/src/context.rs
new file mode 100644
index 00000000..c2a3f194
--- /dev/null
+++ b/core/connectors/runtime/src/context.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 crate::{
+ SinkConnectorWrapper, SourceConnectorWrapper,
+ configs::RuntimeConfig,
+ manager::{
+ sink::{SinkDetails, SinkInfo, SinkManager},
+ source::{SourceDetails, SourceInfo, SourceManager},
+ },
+};
+
+use tracing::error;
+
+pub struct RuntimeContext {
+ pub sinks: SinkManager,
+ pub sources: SourceManager,
+}
+
+pub fn init(
+ config: &RuntimeConfig,
+ sink_wrappers: &[SinkConnectorWrapper],
+ source_wrappers: &[SourceConnectorWrapper],
+) -> RuntimeContext {
+ let mut sinks = vec![];
+ for sink_wrapper in sink_wrappers.iter() {
+ for sink_plugin in sink_wrapper.plugins.iter() {
+ let Some(sink_config) = config.sinks.get(&sink_plugin.key) else {
+ error!("Missing sink config for: {}", sink_plugin.key);
+ continue;
+ };
+
+ sinks.push(SinkDetails {
+ key: sink_plugin.key.to_owned(),
+ info: SinkInfo {
+ id: sink_plugin.id,
+ name: sink_plugin.name.to_owned(),
+ path: sink_plugin.path.to_owned(),
+ enabled: sink_config.enabled,
+ running: true,
+ },
+ config: sink_config.config.clone(),
+ });
+ }
+ }
+
+ let mut sources = vec![];
+ for source_wrapper in source_wrappers.iter() {
+ for source_plugin in source_wrapper.plugins.iter() {
+ let Some(source_config) = config.sources.get(&source_plugin.key) else {
+ error!("Missing source config for: {}", source_plugin.key);
+ continue;
+ };
+
+ sources.push(SourceDetails {
+ key: source_plugin.key.to_owned(),
+ info: SourceInfo {
+ id: source_plugin.id,
+ name: source_plugin.name.to_owned(),
+ path: source_plugin.path.to_owned(),
+ enabled: source_config.enabled,
+ running: true,
+ },
+ config: source_config.config.clone(),
+ });
+ }
+ }
+
+ RuntimeContext {
+ sinks: SinkManager::new(sinks),
+ sources: SourceManager::new(sources),
+ }
+}
diff --git a/core/connectors/runtime/src/error.rs b/core/connectors/runtime/src/error.rs
index 702353e8..8d26ef0f 100644
--- a/core/connectors/runtime/src/error.rs
+++ b/core/connectors/runtime/src/error.rs
@@ -38,4 +38,22 @@ pub enum RuntimeError {
IggyError(#[from] iggy::prelude::IggyError),
#[error("Missing Iggy credentials")]
MissingIggyCredentials,
+ #[error("JSON error")]
+ JsonError(#[from] serde_json::Error),
+ #[error("Sink not found with key: {0}")]
+ SinkNotFound(String),
+ #[error("Source not found with key: {0}")]
+ SourceNotFound(String),
+}
+
+impl RuntimeError {
+ pub fn as_code(&self) -> &'static str {
+ match self {
+ RuntimeError::SinkNotFound(_) => "sink_not_found",
+ RuntimeError::SourceNotFound(_) => "source_not_found",
+ RuntimeError::MissingIggyCredentials => "invalid_configuration",
+ RuntimeError::InvalidConfiguration(_) => "invalid_configuration",
+ _ => "error",
+ }
+ }
}
diff --git a/core/connectors/runtime/src/main.rs b/core/connectors/runtime/src/main.rs
index f87057bf..f3737c93 100644
--- a/core/connectors/runtime/src/main.rs
+++ b/core/connectors/runtime/src/main.rs
@@ -36,8 +36,11 @@ use std::{
use tracing::{debug, error, info};
use tracing_subscriber::{EnvFilter, Registry, layer::SubscriberExt, util::SubscriberInitExt};
+mod api;
pub(crate) mod configs;
+pub(crate) mod context;
pub(crate) mod error;
+mod manager;
mod sink;
mod source;
mod transform;
@@ -91,10 +94,11 @@ async fn main() -> Result<(), RuntimeError> {
.try_deserialize()
.expect("Failed to deserialize runtime config");
- let iggy_address = config.iggy.address;
- let iggy_username = config.iggy.username;
- let iggy_password = config.iggy.password;
- let iggy_token = config.iggy.token;
+ let iggy_config = config.iggy.clone();
+ let iggy_address = iggy_config.address;
+ let iggy_username = iggy_config.username;
+ let iggy_password = iggy_config.password;
+ let iggy_token = iggy_config.token;
let consumer_client = create_iggy_client(
&iggy_address,
iggy_username.as_deref(),
@@ -112,8 +116,8 @@ async fn main() -> Result<(), RuntimeError> {
.await?;
producer_client.connect().await?;
- let sources = source::init(config.sources, &producer_client).await?;
- let sinks = sink::init(config.sinks, &consumer_client).await?;
+ let sources = source::init(config.sources.clone(), &producer_client).await?;
+ let sinks = sink::init(config.sinks.clone(), &consumer_client).await?;
let mut sink_wrappers = vec![];
let mut sink_with_plugins = HashMap::new();
@@ -149,6 +153,10 @@ 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;
+
source::handle(source_wrappers);
sink::consume(sink_wrappers);
info!("All sources and sinks spawned.");
@@ -261,6 +269,9 @@ struct SinkConnector {
struct SinkConnectorPlugin {
id: u32,
+ key: String,
+ name: String,
+ path: String,
consumers: Vec<SinkConnectorConsumer>,
}
@@ -288,6 +299,9 @@ struct SourceConnector {
struct SourceConnectorPlugin {
id: u32,
+ key: String,
+ name: String,
+ path: String,
transforms: Vec<Arc<dyn Transform>>,
producer: Option<SourceConnectorProducer>,
}
diff --git a/core/connectors/runtime/src/manager/mod.rs b/core/connectors/runtime/src/manager/mod.rs
new file mode 100644
index 00000000..8efc27b1
--- /dev/null
+++ b/core/connectors/runtime/src/manager/mod.rs
@@ -0,0 +1,20 @@
+/* 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 sink;
+pub mod source;
diff --git a/core/connectors/runtime/src/manager/sink.rs b/core/connectors/runtime/src/manager/sink.rs
new file mode 100644
index 00000000..db950e45
--- /dev/null
+++ b/core/connectors/runtime/src/manager/sink.rs
@@ -0,0 +1,79 @@
+/* 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 std::collections::HashMap;
+
+#[derive(Debug)]
+pub struct SinkManager {
+ sinks: HashMap<String, SinkDetails>,
+}
+
+impl SinkManager {
+ pub fn new(sinks: Vec<SinkDetails>) -> Self {
+ Self {
+ sinks: sinks
+ .into_iter()
+ .map(|sink| (sink.key.to_owned(), sink))
+ .collect(),
+ }
+ }
+
+ pub fn get(&self, key: &str) -> Option<&SinkDetails> {
+ self.sinks.get(key)
+ }
+
+ pub fn get_all(&self) -> HashMap<&String, &SinkInfo> {
+ self.sinks
+ .iter()
+ .map(|(key, sink)| (key, &sink.info))
+ .collect()
+ }
+}
+
+#[derive(Debug)]
+pub struct SinkDetails {
+ pub key: String,
+ pub info: SinkInfo,
+ pub config: Option<serde_json::Value>,
+}
+
+#[derive(Debug)]
+pub struct SinkInfo {
+ pub id: u32,
+ pub name: String,
+ pub path: String,
+ pub enabled: bool,
+ pub running: bool,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct SinkInfoResponse {
+ pub id: u32,
+ pub key: String,
+ pub name: String,
+ pub path: String,
+ pub enabled: bool,
+ pub running: bool,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct SinkDetailsResponse {
+ #[serde(flatten)]
+ pub info: SinkInfoResponse,
+}
diff --git a/core/connectors/runtime/src/manager/source.rs b/core/connectors/runtime/src/manager/source.rs
new file mode 100644
index 00000000..6ef774d8
--- /dev/null
+++ b/core/connectors/runtime/src/manager/source.rs
@@ -0,0 +1,79 @@
+/* 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 std::collections::HashMap;
+
+#[derive(Debug)]
+pub struct SourceManager {
+ sources: HashMap<String, SourceDetails>,
+}
+
+impl SourceManager {
+ pub fn new(sources: Vec<SourceDetails>) -> Self {
+ Self {
+ sources: sources
+ .into_iter()
+ .map(|source| (source.key.to_owned(), source))
+ .collect(),
+ }
+ }
+
+ pub fn get(&self, key: &str) -> Option<&SourceDetails> {
+ self.sources.get(key)
+ }
+
+ pub fn get_all(&self) -> HashMap<&String, &SourceInfo> {
+ self.sources
+ .iter()
+ .map(|(key, source)| (key, &source.info))
+ .collect()
+ }
+}
+
+#[derive(Debug)]
+pub struct SourceDetails {
+ pub key: String,
+ pub info: SourceInfo,
+ pub config: Option<serde_json::Value>,
+}
+
+#[derive(Debug)]
+pub struct SourceInfo {
+ pub id: u32,
+ pub name: String,
+ pub path: String,
+ pub enabled: bool,
+ pub running: bool,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct SourceInfoResponse {
+ pub id: u32,
+ pub key: String,
+ pub name: String,
+ pub path: String,
+ pub enabled: bool,
+ pub running: bool,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct SourceDetailsResponse {
+ #[serde(flatten)]
+ pub info: SourceInfoResponse,
+}
diff --git a/core/connectors/runtime/src/sink.rs b/core/connectors/runtime/src/sink.rs
index 17b014eb..6f6946a7 100644
--- a/core/connectors/runtime/src/sink.rs
+++ b/core/connectors/runtime/src/sink.rs
@@ -63,6 +63,9 @@ pub async fn init(
);
container.plugins.push(SinkConnectorPlugin {
id: plugin_id,
+ key: key.to_owned(),
+ name: name.to_owned(),
+ path: path.to_owned(),
consumers: vec![],
});
} else {
@@ -76,6 +79,9 @@ pub async fn init(
container,
plugins: vec![SinkConnectorPlugin {
id: plugin_id,
+ key: key.to_owned(),
+ name: name.to_owned(),
+ path: path.to_owned(),
consumers: vec![],
}],
},
diff --git a/core/connectors/runtime/src/source.rs b/core/connectors/runtime/src/source.rs
index 9cde82ca..bdc00123 100644
--- a/core/connectors/runtime/src/source.rs
+++ b/core/connectors/runtime/src/source.rs
@@ -63,6 +63,9 @@ pub async fn init(
);
container.plugins.push(SourceConnectorPlugin {
id: plugin_id,
+ key: key.to_owned(),
+ name: name.to_owned(),
+ path: path.to_owned(),
producer: None,
transforms: vec![],
});
@@ -77,6 +80,9 @@ pub async fn init(
container,
plugins: vec![SourceConnectorPlugin {
id: plugin_id,
+ key: key.to_owned(),
+ name: name.to_owned(),
+ path: path.to_owned(),
producer: None,
transforms: vec![],
}],