Module streams_app 

StreamsApp is one component that owns the schema-registry lifecycle and the managed KafkaStreams runtime. An application therefore does not hand-wire the cache, set_default_registry, the pre-warm, and KafkaStreams::builder().

Build it first, because that installs the process default registry. Then build a topology against it and run the topology. The registry must be installed before the code builds the topology, because the default Avro, Protobuf, and JSON serdes read it when the DSL or Topology builds them.

use apache_avro::AvroSchema;
use krabka_client_streams::{DefaultSerde, SchemaSerde, StreamsApp};
use krabka_schema_serde::format::avro::AvroSerde;
use serde::{Deserialize, Serialize};

#[derive(Clone, Serialize, Deserialize, AvroSchema)]
struct Order {
    id: String,
    total: f64,
}
impl DefaultSerde for Order {
    type Serde = SchemaSerde<Order, AvroSerde<Order>>;
}

let app = StreamsApp::builder()
    .bootstrap("127.0.0.1:9092")
    .application_id("orders")
    .schema_registry("http://127.0.0.1:8081")
    .build();

let topology = app.streams_builder();
topology
    .stream::<String, Order>(["orders"])
    .map_values(|o: &Order| Order {
        id: o.id.clone(),
        total: o.total * 2.0,
    })
    .to("orders-doubled");

let streams = app.run(topology).await?;
streams.close().await?;

Structs

StreamsApp

Owns the schema-registry lifecycle and the KafkaStreams wiring for a streams application.

StreamsAppBuilder

Use builder syntax to set the inputs and finish with build().