Supported Kafka version 2.2.1

A collection of crates for working with the Kafka protocol. The Kafka protocol is a request/response orientated binary protocol. It consists of several independantly versioned RPC calls. The Kafka protocol supports fixed and variably sized big-endian encoded "leaf" types.

kafka-protocol

Traits implemented by all Kafka RPC types.

pub trait KafkaRpc: KafkaRpcType {
    fn version_added() -> i16;
    fn version_removed() -> Option<i16>;
}

pub trait KafkaRpcType: Sized {
    fn read<R: io::Read>(ctx: &mut DeserializeCtx<R>) -> Result<Self, CodecError>;
    fn size(&self, version: i16) -> usize;
    fn write<W: io::Write>(&self, ctx: &mut SerializeCtx<W>) -> Result<(), CodecError>;
}

Also includes KafkaRpcType implementations for several of the "leaf" Kafka RPC datatypes. These include bool, i8, i16, i32, i64, String, Vec<u8>. In addition, each "leaf" type can also be wrapped in an Option or a Vec.

kafka-protocol-derive

proc-macro for deriving the KafkaRpc and KafkaRpcType trait.

RPC versioning is expressed with the added and removed field attributes. An optional default can be supplied for each field as well. When serializing a derived KafkaRpc, the provided version: i16 will determine the wire format. Fields which are not present for the provided version will be ignored.

When deserializing, fields which are not expected will be set to either the default specified by the default field attribute, or the result of calling Default::default().

use kafka_protocol_derive::KafkaRpc;
#[derive(KafkaRpc)]
#[kafka(added = 0)]
struct MetadataRequestTopic {
  #[kafka(added = 0)]
  name: String
}
#[derive(KafkaRpc)]
struct MetadataRequest {
  topics: Option<Vec<MetadataRequestTopic>>,
  #[kafka(added = 4, default = "true")]
  allow_auto_creation: bool,
  #[kafka(added = 8)]
  include_cluster_authorize_operations: bool,
  #[kafka(added = 8)]
  include_topic_authorized_operations: bool,
}

kafka-api and codegen

Contains Kafka RPC types and errors. Includes a KafkaRequest and KafkaResponse type which represent a combined header and body. kafka-api provides methods for serializing and deserializing requests/responses given a type implementing std::io::Read or std::io::Write.

These types are generated by the codegen crate, currently including all RPC types for Kafka version 2.2.1.

kafka-transport

ClientTransport and ServerTransport implementations supporting asynchronous RPC communication. Requires #![feature(async_await)].

This is still a work-in-progress.