# Apache Iggy connector

> **CocoIndex v1, Rust SDK (experimental).** This page documents the CocoIndex **v1** Rust SDK (the `cocoindex` crate). The Rust SDK is experimental: any release may change or remove its API, and it is not yet published on crates.io. The Python SDK pages are at https://cocoindex.io/docs/.
>
> Source: https://cocoindex.io/docs/rust/connectors/iggy/ · Rust docs index: https://cocoindex.io/docs/rust/llms.txt · Rust SDK tour: https://github.com/cocoindex-io/cocoindex/blob/main/rust/sdk/SHOWCASE.md · Rust examples: https://github.com/cocoindex-io/cocoindex/tree/main/examples/rust

The `iggy` connector supports [Apache Iggy](https://iggy.apache.org/docs/) as
both a **source** and a **target**. Iggy organizes messages as
**streams → topics → partitions**; CocoIndex follows that model and treats an
Iggy topic partition either as a raw live stream or as an application-keyed
live map.

```rust
use cocoindex::connectors::iggy::{self, IggyConsumer, IggyProducer, IggyReceivedMessage};
```

`cocoindex::connectors::iggy::prelude` re-exports the upstream `iggy` crate's prelude (`StreamClient`, `TopicClient`, ...), for creating streams and topics without a direct dependency on that crate.

Enable the `iggy` Cargo feature of the `cocoindex` crate.

The connector has two connection handles: an `IggyProducer` for the target,
provided through your app context, and an `IggyConsumer` that you pass to the
source functions. Streams and topics are user-managed: CocoIndex does not
create or drop them.

```rust
cocoindex::context_key!(static PRODUCER: IggyProducer);

let producer = IggyProducer::connect("iggy://iggy:iggy@localhost:8090").await?;
let consumer = IggyConsumer::connect("iggy://iggy:iggy@localhost:8090").await?;
```

Both handles connect from a connection string; credentials in the string log in on connect and are stripped from the handle's stable identity. `producer.client()` gives the underlying `IggyClient`. For a client you built and connected yourself, `IggyProducer::from_client(client, state_id)` and `IggyConsumer::from_client(client, state_id)` wrap it; `state_id` is the server's stable identity (its address), never a credential.

## As Source

### As a live stream

Use `topic_as_stream()` when every Iggy message is an event and downstream code
does not need map-style deletion semantics.

```rust
pub fn topic_as_stream(
    consumer: &IggyConsumer,
    consumer_group: impl Into<String>,
    stream: impl Into<String>,
    topic: impl Into<String>,
) -> IggyTopicStream

pub fn topic_as_stream_with_options(
    consumer: &IggyConsumer,
    consumer_group: impl Into<String>,
    stream: impl Into<String>,
    topic: impl Into<String>,
    options: IggyPollOptions,
) -> IggyTopicStream

pub struct IggyPollOptions {
    /// Messages per poll (default `100`).
    pub batch_length: u32,
    /// How long `watch` waits after an empty poll before polling again (default 300 ms).
    pub poll_interval: Duration,
    /// How long to wait after a failed poll before retrying it, or `None`
    /// (the default) to fail the watch on the first poll error.
    pub polling_retry_interval: Option<Duration>,
    /// How many times to look again for a stream or topic that does not exist
    /// yet when the source starts (default `0`).
    pub init_retries: u32,
    /// The wait between `init_retries` (default 1 s).
    pub init_retry_interval: Duration,
    /// Whether the consumer group's stored offset may move backwards.
    pub allow_replay: bool,
}
```

`IggyTopicStream` is a `LiveStream<IggyReceivedMessage>`. It is also a `LiveMapFeed<String, IggyReceivedMessage>` for `ctx.mount_each_live`: each message becomes a distinct child keyed by its `"{partition}:{offset}"` position. An `IggyReceivedMessage` carries the message's `partition`, `offset`, `id`, `timestamp`, `checksum` and `payload`, with `payload_str()` for the text form.

The stream reads every partition of the topic, so it has no `partition_id`. It has no `initial_high_watermark` either: readiness is reached when every partition is read up to the offset it had when the watch started.

`IggyTopicStream::payloads()` adapts the stream to a `LiveStream<Vec<u8>>` when the
processing logic only needs message payload bytes.

```rust
let events = iggy::topic_as_stream(&consumer, "cocoindex-worker", "orders", "events").payloads();
```

To process every message in its own component instead, mount the stream itself:

```rust
let feed = iggy::topic_as_stream(&consumer, "cocoindex-worker", "orders", "events");
ctx.mount_each_live(&"stream", feed, move |_ctx, message: IggyReceivedMessage| {
    async move {
        println!("{}", message.payload_str());
        Ok(())
    }
})
.await?;
```

### As a live keyed map

Use `topic_as_map()` when message payloads encode an application-level key and
you want CocoIndex to treat the topic as a live map.

```rust
pub fn topic_as_map(
    consumer: &IggyConsumer,
    consumer_group: impl Into<String>,
    stream: impl Into<String>,
    topic: impl Into<String>,
    key_fn: IggyKeyFn,
) -> IggyTopicMap

pub fn topic_as_map_with_options(
    consumer: &IggyConsumer,
    consumer_group: impl Into<String>,
    stream: impl Into<String>,
    topic: impl Into<String>,
    key_fn: IggyKeyFn,
    options: IggySourceOptions,
) -> IggyTopicMap

pub type IggyKeyFn = Arc<dyn Fn(&IggyReceivedMessage) -> Option<String> + Send + Sync>;
pub type IggyIsDeletionFn = Arc<dyn Fn(&IggyReceivedMessage) -> bool + Send + Sync>;

pub struct IggySourceOptions {
    /// Partition to consume (default `0`).
    pub partition: u32,
    pub poll: IggyPollOptions,
    /// Custom deletion predicate over the message (default: never delete).
    pub is_deletion: Option<IggyIsDeletionFn>,
}
```

`IggyTopicMap` is a `LiveMapFeed<String, IggyReceivedMessage>`: the key is what `key_fn` returns and the value is the latest message for that key. It reads one partition.

Iggy messages do not carry Kafka-style keys or tombstones, so `key_fn` is
required. Return `None` from `key_fn` to skip a message. Set
`options.is_deletion` if your payload format has application-level delete
events.

```rust
/// `key_fn` recovering the key from a `"<key>=<value>"` payload.
fn key_fn() -> iggy::IggyKeyFn {
    Arc::new(|message: &IggyReceivedMessage| {
        message
            .payload_str()
            .split_once('=')
            .map(|(k, _)| k.to_string())
    })
}

let feed = iggy::topic_as_map(&consumer, "cocoindex-worker", "orders", "events", key_fn());
ctx.mount_each_live(&"messages", feed, move |_ctx, message: IggyReceivedMessage| {
    async move {
        println!("{}", message.payload_str());
        Ok(())
    }
})
.await?;
```

With options, for another partition or a shorter poll interval:

```rust
let feed = iggy::topic_as_map_with_options(
    &consumer,
    "cocoindex-worker",
    "orders",
    "events",
    key_fn(),
    iggy::IggySourceOptions {
        partition: 0,
        poll: iggy::IggyPollOptions {
            batch_length: 1,
            poll_interval: Duration::from_millis(50),
            ..Default::default()
        },
        is_deletion: None,
    },
);
```

### Readiness and offsets

The source connector disables Iggy auto-commit and stores offsets after the
downstream readiness handle completes. This mirrors Kafka-style back-pressure:
an offset is stored only after CocoIndex has finished processing the message.

The consumer group's offset is stored on the server once a polled batch's
children have synced, and a later live run resumes from it instead of
scanning. The catch-up scan of `topic_as_map` reads the partition log up to its
current offset, compacted to the latest message per key. Unless
`IggyPollOptions::allow_replay` is set, an offset lower than the one stored is
not stored.

## As Target

The target connector sends byte payloads to a user-managed Iggy
stream/topic/partition.

```rust
cocoindex::context_key!(static PRODUCER: IggyProducer);

// At startup: `Environment::builder().provide_key(&PRODUCER, producer)`.

let options = iggy::IggyTopicOptions {
    partition: 0,
    ..Default::default()
};
let target =
    iggy::mount_iggy_topic_target(&ctx, &PRODUCER, "orders", "derived-events", options).await?;

target.declare_message(&ctx, "order-123", br#"{"status":"ready"}"#)?;
```

`declare_message(ctx, key, value)` takes a non-empty `&str` key and any byte slice as the value; `declare_message_bytes` takes a binary key. A `&str` key and its UTF-8 bytes are distinct target states.

While the topic is still declared, a message target state that is no longer
declared gets a deletion message. Deletes need an application-level delete
payload because Iggy does not have Kafka-style tombstones:

```rust
let options = iggy::IggyTopicOptions {
    partition: 0,
    // Iggy has no tombstone; emit a delete marker for removed rows.
    deletion_value_fn: Some(Arc::new(|key: &[u8]| {
        format!("{{\"_deleted\":\"{}\"}}", String::from_utf8_lossy(key)).into_bytes()
    })),
};
let target =
    iggy::mount_iggy_topic_target(&ctx, &PRODUCER, "orders", "derived-events", options).await?;
```

Without a `deletion_value_fn`, deleting a declared message is an error. The callback receives the key as stored: a `&str` key is its UTF-8 bytes. The full program is `examples/rust/csv_to_iggy`.

### Abandoning a topic

Removing the topic declaration itself — or calling `App::drop_state` — **abandons**
the topic: CocoIndex stops producing to it and sends no deletion messages for
the messages it produced earlier. Changing the stream, topic, or partition in
code abandons the old one the same way. Avoid reusing an abandoned topic in a
later declaration: its stale keys remain and never get a deletion message, and
keys you declare again are republished as new even when their values are
unchanged.

To send deletion messages for everything before abandoning a topic, drain it
first: keep the topic declared but declare no messages in it, run an update,
then remove the declaration.

### Target APIs

`iggy_topic_target()` takes the same arguments as the two functions below and
builds the topic's target state alone, for `mount_target` or `declare_target_state_with_child`.

```rust
pub fn declare_iggy_topic_target(
    ctx: &Ctx,
    producer: &ContextKey<IggyProducer>,
    stream: impl Into<String>,
    topic: impl Into<String>,
    options: IggyTopicOptions,
) -> Result<IggyTopicTarget>

pub async fn mount_iggy_topic_target(
    ctx: &Ctx,
    producer: &ContextKey<IggyProducer>,
    stream: impl Into<String>,
    topic: impl Into<String>,
    options: IggyTopicOptions,
) -> Result<IggyTopicTarget>

pub fn iggy_topic_target(
    ctx: &Ctx,
    producer: &ContextKey<IggyProducer>,
    stream: impl Into<String>,
    topic: impl Into<String>,
    options: IggyTopicOptions,
) -> Result<TargetState<TopicSpec>>

pub struct IggyTopicOptions {
    /// Partition to send messages to (default `0`).
    pub partition: u32,
    /// Value to send when a previously-declared message is removed.
    pub deletion_value_fn: Option<DeletionValueFn>,
}
/// Takes the key as stored: a `&str` key is its UTF-8 bytes.
pub type DeletionValueFn = Arc<dyn Fn(&[u8]) -> Vec<u8> + Send + Sync>;

pub fn IggyTopicTarget::declare_message(
    &self,
    ctx: &Ctx,
    key: &str,
    value: impl AsRef<[u8]>,
) -> Result<()>

pub fn IggyTopicTarget::declare_message_bytes(
    &self,
    ctx: &Ctx,
    key: &[u8],
    value: impl AsRef<[u8]>,
) -> Result<()>
```