partitionline 0.1.0

Pure-Rust Apache Kafka client and protocol implementation
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

//! Produce and fetch interceptors (Java `ProducerInterceptor` / `ConsumerInterceptor`).

use std::fmt;
use std::sync::Arc;

use crate::consumer::{FetchedRecord, OffsetAndMetadata, TopicPartition};
use crate::error::Error;
use crate::producer::{ProduceRecord, RecordMetadata};

/// Mutate a record before it is partitioned and queued, and observe acks.
pub trait ProducerInterceptor: Send + Sync + 'static {
    /// Return the record to send. The default is pass-through.
    fn on_send(&self, rec: ProduceRecord) -> ProduceRecord {
        rec
    }

    /// Broker ack (including `acks=0` local complete).
    fn on_ack(&self, _md: &RecordMetadata) {}

    /// The record failed (broker error, timeout, closed).
    fn on_error(&self, _err: &Error) {}

    /// The producer is closing. The default is a no-op. Safe to call more than once.
    fn close(&self) {}
}

/// Mutate fetch results before they are returned to the caller.
pub trait ConsumerInterceptor: Send + Sync + 'static {
    /// Return the records the caller should see. The default is pass-through.
    ///
    /// Filtering records here does not rewind fetch positions (Java
    /// `ConsumerInterceptor.onConsume`).
    fn on_consume(&self, recs: Vec<FetchedRecord>) -> Vec<FetchedRecord> {
        recs
    }

    /// Offsets were committed (`OffsetCommit` succeeded).
    fn on_commit(&self, _offsets: &[(TopicPartition, OffsetAndMetadata)]) {}

    /// The consumer is closing. The default is a no-op. Safe to call more than once.
    fn close(&self) {}
}

/// Chain of [`ProducerInterceptor`]s. Empty is a no-op.
#[derive(Clone, Default)]
pub struct ProducerInterceptors {
    inner: Vec<Arc<dyn ProducerInterceptor>>,
}

impl ProducerInterceptors {
    /// Append one interceptor. They run in insertion order.
    pub fn push(&mut self, i: impl ProducerInterceptor) {
        self.inner.push(Arc::new(i));
    }

    pub(crate) fn on_send(&self, rec: ProduceRecord) -> ProduceRecord {
        let mut rec = rec;
        for i in &self.inner {
            rec = i.on_send(rec);
        }
        rec
    }

    pub(crate) fn on_ack(&self, md: &RecordMetadata) {
        for i in &self.inner {
            i.on_ack(md);
        }
    }

    pub(crate) fn on_error(&self, err: &Error) {
        for i in &self.inner {
            i.on_error(err);
        }
    }

    pub(crate) fn close(&self) {
        for i in &self.inner {
            i.close();
        }
    }
}

impl fmt::Debug for ProducerInterceptors {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_tuple("ProducerInterceptors")
            .field(&self.inner.len())
            .finish()
    }
}

/// Chain of [`ConsumerInterceptor`]s. Empty is a no-op.
#[derive(Clone, Default)]
pub struct ConsumerInterceptors {
    inner: Vec<Arc<dyn ConsumerInterceptor>>,
}

impl ConsumerInterceptors {
    /// Append one interceptor. They run in insertion order.
    pub fn push(&mut self, i: impl ConsumerInterceptor) {
        self.inner.push(Arc::new(i));
    }

    pub(crate) fn on_consume(&self, recs: Vec<FetchedRecord>) -> Vec<FetchedRecord> {
        let mut recs = recs;
        for i in &self.inner {
            recs = i.on_consume(recs);
        }
        recs
    }

    pub(crate) fn on_commit(&self, offsets: &[(TopicPartition, OffsetAndMetadata)]) {
        for i in &self.inner {
            i.on_commit(offsets);
        }
    }

    pub(crate) fn close(&self) {
        for i in &self.inner {
            i.close();
        }
    }
}

impl fmt::Debug for ConsumerInterceptors {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_tuple("ConsumerInterceptors")
            .field(&self.inner.len())
            .finish()
    }
}