Apache Iggy connector
Consume Apache Iggy topics as live streams/maps and produce messages as target states.
The iggy connector supports Apache Iggy 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.
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.
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.
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.
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:
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.
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.
/// `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:
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.
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:
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.
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<()>