Class KafkaPublisher

java.lang.Object

org.chenile.pubsub.kafka.impl.KafkaPublisher

All Implemented Interfaces:

ChenilePub

public class KafkaPublisher
extends Object
implements ChenilePub

Kafka-based implementation of the ChenilePub interface. Provides methods to publish messages to Kafka topics synchronously or asynchronously.

Field Summary

Fields

Modifier and Type	Field	Description
private final org.springframework.kafka.core.KafkaTemplate<String,String>	kafkaTemplate
private final PubSubInfoProvider	pubSubInfoProvider

Constructor Summary

Constructors

Constructor	Description
KafkaPublisher(org.springframework.kafka.core.KafkaTemplate<String,String> kafkaTemplate, PubSubInfoProvider pubSubInfoProvider)

Method Summary

Modifier and Type	Method	Description
void	asyncPublish(String topic, String payload, Map<String,Object> properties)	Publishes a message asynchronously to the given Kafka topic.
private static List<org.apache.kafka.common.header.Header>	buildHeaders(String topic, Map<String,Object> properties)	Builds Kafka headers from message properties and adds the Chenile topic key.
private static Integer	getPartition(Map<String,Object> properties)	Extracts the partition number from the properties map.
void	publish(String topic, String payload, Map<String,Object> properties)	Publishes a message to the global Kafka topic.
void	publishToOperation(String service, String operationName, String payload, Map<String,Object> properties)	Publishes a message to the given service's operation topic.
private static String	substituteProperties(String text, Map<String,Object> properties)	Replaces placeholders in a string with values from the given properties map.

Methods inherited from class Object

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

Field Details

kafkaTemplate

private final org.springframework.kafka.core.KafkaTemplate<String,String> kafkaTemplate

pubSubInfoProvider

private final PubSubInfoProvider pubSubInfoProvider

Constructor Details

KafkaPublisher

@Autowired
public KafkaPublisher(org.springframework.kafka.core.KafkaTemplate<String,String> kafkaTemplate,
 PubSubInfoProvider pubSubInfoProvider)

Method Details

publishToOperation

public void publishToOperation(String service,
 String operationName,
 String payload,
 Map<String,Object> properties)

Publishes a message to the given service's operation topic.

Specified by:

publishToOperation in interface ChenilePub

Parameters:

service - the service name

operationName - the operation name

payload - the message payload

properties - additional message properties

asyncPublish

public void asyncPublish(String topic,
 String payload,
 Map<String,Object> properties)

Publishes a message asynchronously to the given Kafka topic.

Specified by:

asyncPublish in interface ChenilePub

Parameters:

topic - the Kafka topic

payload - the message payload

properties - additional message properties

publish

public void publish(String topic,
 String payload,
 Map<String,Object> properties)

Publishes a message to the global Kafka topic.

Specified by:

publish in interface ChenilePub

Parameters:

topic - logical topic name (added as header, not used as Kafka topic)

payload - the message payload

properties - additional message properties

buildHeaders

private static List<org.apache.kafka.common.header.Header> buildHeaders(String topic,
 Map<String,Object> properties)

Builds Kafka headers from message properties and adds the Chenile topic key.

Parameters:

topic - the logical topic

properties - message properties

Returns:

list of Kafka headers

substituteProperties

private static String substituteProperties(String text,
 Map<String,Object> properties)

Replaces placeholders in a string with values from the given properties map.

Parameters:

text - the string containing placeholders like {key}

properties - the map of properties

Returns:

the resolved string

getPartition

private static Integer getPartition(Map<String,Object> properties)

Extracts the partition number from the properties map.

Parameters:

properties - the message properties

Returns:

the partition number (default = 0)