Package org.springframework.kafka.listener.adapter

Class ConvertingMessageListener<V>

java.lang.Object

org.springframework.kafka.listener.adapter.ConvertingMessageListener<V>

Type Parameters:

V - the desired value type after conversion.

All Implemented Interfaces:

AcknowledgingConsumerAwareMessageListener<Object,Object>, DelegatingMessageListener<MessageListener>, GenericMessageListener<org.apache.kafka.clients.consumer.ConsumerRecord<Object,Object>>, MessageListener<Object,Object>

public class ConvertingMessageListener<V>
extends Object
implements DelegatingMessageListener<MessageListener>, AcknowledgingConsumerAwareMessageListener<Object,Object>

A AcknowledgingConsumerAwareMessageListener adapter that implements converting received ConsumerRecord using specified MessageConverter and then passes result to specified MessageListener. If directly set, also headers can be mapped with implementation of KafkaHeaderMapper and then passed to converter as a part of message being actually processed. Otherwise, if header mapper is not specified, headers will not be accessible from converter's perspective.

Since:

3.0

Author:

Adrian Chlebosz

See Also:

• DelegatingMessageListener
• AcknowledgingConsumerAwareMessageListener

Constructor Summary

Constructors

Constructor	Description
ConvertingMessageListener(MessageListener<?,V> delegateMessageListener, Class<V> desiredValueType)	Construct an instance with the provided MessageListener and Class as a desired type of ConsumerRecord's value after conversion.

Method Summary

Modifier and Type	Method	Description
MessageListener	getDelegate()	Return the delegate.
void	onMessage(org.apache.kafka.clients.consumer.ConsumerRecord receivedRecord, Acknowledgment acknowledgment, org.apache.kafka.clients.consumer.Consumer consumer)	Invoked with data from kafka and provides access to the Consumer.
void	setKafkaHeaderMapper(KafkaHeaderMapper headerMapper)	Set a KafkaHeaderMapper.
void	setMessageConverter(MessageConverter messageConverter)	Set a MessageConverter.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface org.springframework.kafka.listener.AcknowledgingConsumerAwareMessageListener

onMessage

Methods inherited from interface org.springframework.kafka.listener.GenericMessageListener

onMessage, onMessage

Constructor Details

ConvertingMessageListener

public ConvertingMessageListener(MessageListener<?,V> delegateMessageListener,
 Class<V> desiredValueType)

Construct an instance with the provided MessageListener and Class as a desired type of ConsumerRecord's value after conversion. Default value of MessageConverter is used, which is GenericMessageConverter.

Parameters:

delegateMessageListener - the MessageListener to use when passing converted ConsumerRecord further.

desiredValueType - the Class setting desired type of ConsumerRecord's value.

Method Details

setMessageConverter

public void setMessageConverter(MessageConverter messageConverter)

Set a MessageConverter.

Parameters:

messageConverter - the message converter to use for conversion of incoming ConsumerRecord.

Since:

3.0

setKafkaHeaderMapper

public void setKafkaHeaderMapper(KafkaHeaderMapper headerMapper)

Set a KafkaHeaderMapper.

Parameters:

headerMapper - the header mapper to use for mapping headers of incoming ConsumerRecord.

Since:

3.0

getDelegate

public MessageListener getDelegate()

Description copied from interface: DelegatingMessageListener

Return the delegate.

Specified by:

getDelegate in interface DelegatingMessageListener<V>

Returns:

the delegate.

onMessage

public void onMessage(org.apache.kafka.clients.consumer.ConsumerRecord receivedRecord,
 Acknowledgment acknowledgment,
 org.apache.kafka.clients.consumer.Consumer consumer)

Description copied from interface: GenericMessageListener

Invoked with data from kafka and provides access to the Consumer. The default implementation throws UnsupportedOperationException.

Specified by:

onMessage in interface AcknowledgingConsumerAwareMessageListener<Object,Object>

Specified by:

onMessage in interface GenericMessageListener<V>

Parameters:

receivedRecord - the data to be processed.

acknowledgment - the acknowledgment.

consumer - the consumer.