SDK

SDK provides the commonly used structs and traits such as Sink and Source, along with the sink_connector and source_connector macros to be used when developing connectors.

Moreover, it contains both, the decoders and encoders modules, implementing either StreamDecoder or StreamEncoder traits, which are used when consuming or producing data from/to Iggy streams. All the supported schemas - JSON, Raw bytes, Text, Protocol Buffers, FlatBuffers, and Avro - have both a decoder and an encoder.

Traits

A source produces messages to the configured stream and topic. A sink consumes messages from the configured stream(s) and topic(s):

use async_trait::async_trait;
use iggy_connector_sdk::{
    ConsumedMessage, Error, MessagesMetadata, ProducedMessages, TopicMetadata,
    source::SourceBatchResult,
};

#[async_trait]
pub trait Source: Send + Sync {
    async fn open(&mut self) -> Result<(), Error>;
    async fn poll(&self) -> Result<ProducedMessages, Error>;
    async fn on_batch_result(&self, _result: SourceBatchResult) -> Result<(), Error> {
        Ok(())
    }
    async fn close(&mut self) -> Result<(), Error>;
}

#[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) -> Result<(), Error>;
}

Expose the implementation with sink_connector!(MySink) or source_connector!(MySource). The sink constructor has the signature new(id: u32, config: MySinkConfig) -> Self; the source constructor also takes state: Option<ConnectorState>. The sink guide and source guide provide complete plugin implementations and build setup.

The macros generate the C FFI symbols loaded by the runtime: iggy_sink_open, iggy_sink_consume, iggy_sink_close, and iggy_sink_version for sinks, and iggy_source_open, iggy_source_handle_v2, iggy_source_batch_result, iggy_source_close, and iggy_source_version for sources. Configurations cross FFI as JSON, message batches and metadata use postcard, and source checkpoint bytes are opaque. The SDK creates one Tokio runtime per loaded plugin library, shared by that library's connector instances.

State

Source connectors can persist their position between restarts. Each batch of ProducedMessages may carry an optional ConnectorState. After processing and sending the batch to Iggy, the runtime saves that state through the configured file or HTTP backend before acknowledging the batch. The last saved state is passed to new() on the next startup. ConnectorState wraps raw bytes and provides serialize/deserialize helpers backed by MessagePack.

Sources with cursors or destructive operations must stage changes during poll() and apply them on SourceBatchResult::Ack; Nack discards staged progress so the batch can be polled again. The SDK allows one batch in flight and stops polling if on_batch_result() returns an error. Its default no-op is suitable only for sources without staged work.

Sinks can keep in-memory state, but the runtime does not persist it. Their consumer-group offsets are tracked by Iggy. See the Observability page for checkpoint storage and the sink guide for offset handling.

Retry utilities

The retry module ships helpers for connectors talking to flaky external systems: a CircuitBreaker, an HttpRetryMiddleware for reqwest-based clients, and a build_retry_client factory that wraps a reqwest client with the retry middleware (the circuit breaker is wired separately).

Transforms

Last but not least, the different transforms are available, to transform (add, update, delete etc.) the particular fields of the data being processed via external configuration. It's as simple as adding a new transform to the transforms section of the particular connector configuration file:

[transforms.add_fields]
enabled = true

[[transforms.add_fields.fields]]
key = "message"
value.static = "hello"

All the available transforms and their configuration options are documented on the Transforms page.

Protocol Buffers Support

The SDK includes support for Protocol Buffers (protobuf) format with both encoding and decoding capabilities. Protocol Buffers provide efficient serialization and are particularly useful for high-performance data streaming scenarios.

Configuration Example

This example uses the Random source and Stdout sink from the matching checkout. Start a server using the getting-started guide, then build the plugins, runtime, and CLI from the repository root:

cargo build --release -p iggy_connector_random_source -p iggy_connector_stdout_sink -p iggy-connectors -p iggy-cli
mkdir -p connectors

The source's schema = "proto" selects the default protobuf encoder, which wraps each JSON record in a google.protobuf.StringValue inside google.protobuf.Any. The sink reads raw bytes and applies proto_convert to expose the Any envelope as JSON. This does not unpack a custom protobuf message schema.

Main runtime config (connectors.toml):

[iggy]
address = "localhost:8090"
username = "iggy"
password = "iggy"

[connectors]
config_type = "local"
config_dir = "connectors"

Source connector config (connectors/protobuf_source.toml):

type = "source"
key = "protobuf"
enabled = true
version = 0
name = "Protobuf Source"
path = "target/release/libiggy_connector_random_source"

[[streams]]
stream = "protobuf_stream"
topic = "protobuf_topic"
schema = "proto"
batch_length = 1000
linger_time = "5ms"

[plugin_config]
interval = "100ms"
messages_range = [1, 10]
payload_size = 32
max_count = 100

Sink connector config (connectors/protobuf_sink.toml):

type = "sink"
key = "protobuf"
enabled = true
version = 0
name = "Protobuf Sink"
path = "target/release/libiggy_connector_stdout_sink"

[[streams]]
stream = "protobuf_stream"
topics = ["protobuf_topic"]
schema = "raw"

[plugin_config]
print_payload = true

[transforms.proto_convert]
enabled = true
source_format = "proto"
target_format = "json"
include_paths = ["."]
preserve_unknown_fields = false

[transforms.proto_convert.conversion_options]
validate_messages = true
pretty_json = false
include_metadata = false
type_url_prefix = "type.googleapis.com"
strict_mode = false

Create the stream and topic, then start the runtime from the repository root:

./target/release/iggy --username iggy --password iggy stream create protobuf_stream
./target/release/iggy --username iggy --password iggy topic create protobuf_stream protobuf_topic 1 none 1d
IGGY_CONNECTORS_CONFIG_PATH=connectors.toml ./target/release/iggy-connectors

The source sends 100 records, then continues polling without new messages. Stdout logs message offsets and the serialized JSON envelope bytes, containing type_url and base64 value. The sink's raw schema also determines how the plugin receives those transformed bytes.

The format-conversion transforms define no per-key defaults. Every non-optional key shown above must be present, or the configuration fails to deserialize (schema_path, message_type, field_mappings, and descriptor_set are optional).

The two [[streams]] shapes differ: a source produces to a single topic and can tune batching via batch_length and linger_time, while a sink consumes from a list of topics and can additionally set batch_length, poll_interval, and consumer_group.

Key Configuration Options

Programmatic Encoder and Decoder Configuration

These are SDK configuration fields, not Random or Stdout plugin_config keys. The runtime's schema = "proto" uses the default encoder or decoder.

• schema_path: Path to the .proto file containing message definitions
• message_type: Fully qualified name of the protobuf message type to use
• use_any_wrapper: Selects the Any fallback when no message descriptor is loaded; a loaded descriptor takes precedence

Transform Options

• proto_convert: Transform for converting between protobuf and other formats
• source_format / target_format: Formats to convert between - any schema value (json, raw, text, proto, flat_buffer, avro)
• preserve_unknown_fields: Accepted by proto_convert, but currently has no effect
• include_paths: Additional directories searched for imported .proto files
• field_mappings: Renames fields in a JSON input object before conversion (e.g., "old_field" = "new_field")
• conversion_options: pretty_json controls JSON text output and include_metadata enriches supported protobuf-to-JSON paths. validate_messages, type_url_prefix, and strict_mode are accepted but currently have no effect

The schema_registry_url field is reserved and currently not implemented. The SDK never contacts a schema registry, and schemas are loaded only from schema_path or descriptor_set.

Supported Features

• Encoding: A loaded message descriptor encodes matching JSON fields. The encoder supports booleans, strings, all protobuf integer types, and base64 strings for bytes or already-encoded nested messages. Float, double, and enum fields are unsupported by the encoder; nested JSON objects, repeated fields, maps, and proto2 groups are not a general-purpose schema conversion path.
• Decoding: A loaded descriptor extracts present fields. Integer, boolean, and string fields become JSON values; bytes become base64 and nested messages become metadata with base64 content. Missing fields are not filled with protobuf defaults, and float/double fields produce unsupported_wire_type placeholders. Without a descriptor, the default decoder returns an Any envelope's type_url and base64 value.
• Transforms: proto_convert supports JSON-to-protobuf schema encoding for scalar fields, including floating-point numbers and numeric enum values; bytes and already-encoded nested messages use base64 strings. It logs and omits fields it cannot encode. Its protobuf-to-JSON path exposes Any metadata or raw-data metadata rather than decoding a custom message descriptor. Converting protobuf to flat_buffer or avro rewraps bytes without transcoding them.
• Field Mapping: Encoder/decoder mappings use protobuf field names as keys and JSON field names as values. The encoder applies that mapping in reverse. Transform mappings rename JSON input keys directly.
• Any Wrapper: The default encoder puts JSON/text in a google.protobuf.StringValue, or binary data in a google.protobuf.BytesValue, inside google.protobuf.Any. The default decoder exposes the envelope without unpacking its inner message.

Programmatic Usage

From the matching repository root, create an example crate and schema directory:

mkdir -p connector-sdk-example/src schemas

Save this as connector-sdk-example/Cargo.toml. The path dependency uses the SDK from the same checkout as the runtime:

[package]
name = "connector-sdk-example"
version = "0.1.0"
edition = "2024"

[dependencies]
iggy_connector_sdk = { path = "../core/connectors/sdk" }
simd-json = { version = "0.18.1", features = ["serde_impl"] }

[workspace]

Save this as schemas/user.proto:

syntax = "proto3";
package com.example;

message User {
  uint64 id = 1;
  string name = 2;
}

Each Rust example below is a complete connector-sdk-example/src/main.rs. Run it from the repository root so the relative schema path resolves:

cargo run --manifest-path connector-sdk-example/Cargo.toml

Dynamic Schema Loading

You can load or reload schemas programmatically:

use iggy_connector_sdk::decoders::proto::{ProtoConfig, ProtoStreamDecoder};
use iggy_connector_sdk::encoders::proto::{ProtoEncoderConfig, ProtoStreamEncoder};
use iggy_connector_sdk::{Error, Payload, StreamDecoder, StreamEncoder};
use std::path::PathBuf;

fn main() -> Result<(), Error> {
    let mut decoder = ProtoStreamDecoder::new_default();
    decoder.update_config(
        ProtoConfig {
            schema_path: Some(PathBuf::from("schemas/user.proto")),
            message_type: Some("com.example.User".to_string()),
            ..ProtoConfig::default()
        },
        true,
    )?;
    let encoder = ProtoStreamEncoder::new_with_config(ProtoEncoderConfig {
        schema_path: Some(PathBuf::from("schemas/user.proto")),
        message_type: Some("com.example.User".to_string()),
        ..ProtoEncoderConfig::default()
    });
    let encoded = encoder.encode(Payload::Json(simd_json::json!({
        "id": 1,
        "name": "Alice"
    })))?;
    println!("{}", decoder.decode(encoded)?);
    Ok(())
}

The encoder follows the same pattern:

use iggy_connector_sdk::encoders::proto::{ProtoEncoderConfig, ProtoStreamEncoder};
use iggy_connector_sdk::{Error, Payload, StreamEncoder};
use std::path::PathBuf;

fn main() -> Result<(), Error> {
    let mut encoder = ProtoStreamEncoder::new_with_config(ProtoEncoderConfig {
        schema_path: Some(PathBuf::from("schemas/user.proto")),
        message_type: Some("com.example.User".to_string()),
        use_any_wrapper: false,
        ..ProtoEncoderConfig::default()
    });
    encoder.load_schema()?;
    let encoded = encoder.encode(Payload::Json(simd_json::json!({
        "id": 1,
        "name": "Alice"
    })))?;
    println!("{encoded:?}");
    Ok(())
}

Creating Converters with Schema

The loaded schema is used for JSON-to-protobuf conversion. This example maps user_id and full_name to the schema's field names:

use iggy_connector_sdk::transforms::{ProtoConvert, ProtoConvertConfig, Transform};
use iggy_connector_sdk::{DecodedMessage, Error, Payload, Schema, TopicMetadata};
use std::collections::HashMap;
use std::path::PathBuf;

fn main() -> Result<(), Error> {
    let converter = ProtoConvert::new(ProtoConvertConfig {
        source_format: Schema::Json,
        target_format: Schema::Proto,
        schema_path: Some(PathBuf::from("schemas/user.proto")),
        message_type: Some("com.example.User".to_string()),
        field_mappings: Some(HashMap::from([
            ("user_id".to_string(), "id".to_string()),
            ("full_name".to_string(), "name".to_string()),
        ])),
        ..ProtoConvertConfig::default()
    });
    let metadata = TopicMetadata {
        stream: "users".to_string(),
        topic: "users".to_string(),
    };
    let message = DecodedMessage {
        id: None,
        offset: None,
        checksum: None,
        timestamp: None,
        origin_timestamp: None,
        headers: None,
        payload: Payload::Json(simd_json::json!({
            "user_id": 1,
            "full_name": "Alice"
        })),
    };
    if let Some(converted) = converter.transform(&metadata, message)? {
        println!("{:?}", converted.payload);
    }
    Ok(())
}

Usage Notes

• Automatic Loading: Constructors attempt to load schema_path or descriptor_set; schema_path takes precedence when both are set. Constructors log loading errors and return an instance without a loaded schema.
• Manual Loading: load_schema() reloads the configured source. A returned error preserves an already-loaded schema. update_config(config, true) also restores the previous configuration on error; false changes the configuration while retaining the cached schema.
• Fallbacks: Missing files, compilation failures, absent schema sources, or an unmatched message_type can return Ok(()) without an active message descriptor. Successful reloads into fallback mode clear the previous descriptor. Invalid protobuf syntax and malformed descriptor bytes return errors. Check the actual encoded/decoded result when validating a schema setup.
• Encoding Errors: Errors encoding a loaded message descriptor are returned to the caller. The encoder does not retry that message as Any. The decoder attempts Any after a schema decoding error.
• Transform Configuration: Create a new converter to change its configuration. load_schema() can reload its existing source. Without a descriptor, JSON-to-protobuf conversion produces JSON text in Payload::Proto, not a schema-encoded binary message.
• Format Options: Encoder preserve_unknown_fields, compact_encoding, validate_message, and deterministic_encoding are accepted but have no effect. Decoder preserve_unknown_fields retains unknown varints as numbers and length-delimited data as base64; fixed-width unknown fields become placeholders. It does not retain the original wire encoding. See the Transforms page for conversion-specific limits.
• Protocol Buffers provide efficient binary serialization compared to JSON