hubcio commented on code in PR #1826: URL: https://github.com/apache/iggy/pull/1826#discussion_r2117866758
########## core/connectors/README.md: ########## @@ -0,0 +1,93 @@ +# Apache Iggy Connectors + +The highly performant and modular runtime for statically typed, yet dynamically loaded connectors. Ingest the data from the external sources and push it further to the Iggy streams, or fetch the data from the Iggy streams and push it further to the external sources. Create your own Rust plugins by simply implementing either the `Source` or `Sink` trait and build custom pipelines for the data processing. + +**This is still WiP, and the runtime can be started only after compilation from the source code (no installable package yet).** + +## Features +- **High Performance**: Utilizes Rust's performance characteristics to ensure fast data ingestion and egress. +- **Low memory footprint**: Designed with memory efficiency in mind, minimizing the memory footprint of the connectors. +- **Modular Design**: Designed with modularity in mind, allowing for easy extension and customization. +- **Dynamic Loading**: Supports dynamic loading of plugins, enabling seamless integration with various data sources and sinks. +- **Statically Typed**: Ensures type safety and compile-time checks, reducing runtime errors. +- **Easy Customization**: Provides a simple interface for implementing custom connectors, making it easy to create new plugins. +- **Data transformation**: Supports data transformation with the help of existing functions. +- **Powerful configuration**: Define your sinks, sources, and transformations in the configuration file. + +## Quick Start + +1. Uncomment the `crate-type = ["cdylib"]` in `Cargo.toml` for the connectors under `core/connectors/sources` and `core/connectors/sinks` to compile them as dynamic libraries (`.so`, `.dylib`, `.dll`). Review Comment: what's the point of DLL? ########## 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> { Review Comment: this whole file should be placed in other directory, it's not production code. consider existing "examples". ########## core/connectors/sdk/src/source.rs: ########## @@ -0,0 +1,232 @@ +/* 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, Review Comment: users of this API need to know that config_ptr is NOT freed after. it is caller responsibility. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
