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
|
package sarama
// ProducerInterceptor allows you to intercept (and possibly mutate) the records
// received by the producer before they are published to the Kafka cluster.
// https://cwiki.apache.org/confluence/display/KAFKA/KIP-42%3A+Add+Producer+and+Consumer+Interceptors#KIP42:AddProducerandConsumerInterceptors-Motivation
type ProducerInterceptor interface {
// OnSend is called when the producer message is intercepted. Please avoid
// modifying the message until it's safe to do so, as this is _not_ a copy
// of the message.
OnSend(*ProducerMessage)
}
// ConsumerInterceptor allows you to intercept (and possibly mutate) the records
// received by the consumer before they are sent to the messages channel.
// https://cwiki.apache.org/confluence/display/KAFKA/KIP-42%3A+Add+Producer+and+Consumer+Interceptors#KIP42:AddProducerandConsumerInterceptors-Motivation
type ConsumerInterceptor interface {
// OnConsume is called when the consumed message is intercepted. Please
// avoid modifying the message until it's safe to do so, as this is _not_ a
// copy of the message.
OnConsume(*ConsumerMessage)
}
func (msg *ProducerMessage) safelyApplyInterceptor(interceptor ProducerInterceptor) {
defer func() {
if r := recover(); r != nil {
Logger.Printf("Error when calling producer interceptor: %v, %v", interceptor, r)
}
}()
interceptor.OnSend(msg)
}
func (msg *ConsumerMessage) safelyApplyInterceptor(interceptor ConsumerInterceptor) {
defer func() {
if r := recover(); r != nil {
Logger.Printf("Error when calling consumer interceptor: %v, %v", interceptor, r)
}
}()
interceptor.OnConsume(msg)
}
|
