org.flowable.eventregistry.spring.kafka

Class KafkaChannelDefinitionProcessor

java.lang.Object

org.flowable.eventregistry.spring.kafka.KafkaChannelDefinitionProcessor

All Implemented Interfaces:

ChannelModelProcessor, org.springframework.beans.factory.Aware, org.springframework.beans.factory.BeanFactoryAware

public class KafkaChannelDefinitionProcessor
extends Object
implements org.springframework.beans.factory.BeanFactoryAware, ChannelModelProcessor

Author:

Filip Hrisafov

Field Summary

Fields

Modifier and Type	Field and Description
protected org.springframework.beans.factory.BeanFactory	beanFactory
static String	CHANNEL_ID_PREFIX
protected org.springframework.kafka.config.KafkaListenerContainerFactory<?>	containerFactory
protected String	containerFactoryBeanName
protected org.springframework.util.StringValueResolver	embeddedValueResolver
protected org.springframework.kafka.config.KafkaListenerEndpointRegistry	endpointRegistry
protected org.springframework.beans.factory.config.BeanExpressionContext	expressionContext
protected org.springframework.kafka.core.KafkaOperations<Object,Object>	kafkaOperations
protected org.slf4j.Logger	logger
protected org.springframework.beans.factory.config.BeanExpressionResolver	resolver

Constructor Summary

Constructors

Constructor and Description
KafkaChannelDefinitionProcessor()

Method Summary

Modifier and Type	Method and Description
boolean	canProcess(ChannelModel channelModel)
protected org.springframework.kafka.config.KafkaListenerEndpoint	createKafkaListenerEndpoint(KafkaInboundChannelModel channelModel, String tenantId, EventRegistry eventRegistry)
protected org.springframework.kafka.listener.GenericMessageListener<org.apache.kafka.clients.consumer.ConsumerRecord<Object,Object>>	createMessageListener(EventRegistry eventRegistry, InboundChannelModel inboundChannelModel)
org.springframework.kafka.config.KafkaListenerContainerFactory<?>	getContainerFactory()
String	getContainerFactoryBeanName()
protected String	getEndpointGroupId(KafkaInboundChannelModel channelDefinition, String id)
protected String	getEndpointId(ChannelModel channelModel, String tenantId)
org.springframework.kafka.config.KafkaListenerEndpointRegistry	getEndpointRegistry()
org.springframework.kafka.core.KafkaOperations<Object,Object>	getKafkaOperations()
protected void	processOutboundDefinition(KafkaOutboundChannelModel channelModel)
void	registerChannelModel(ChannelModel channelModel, String tenantId, EventRegistry eventRegistry, EventRepositoryService eventRepositoryService, boolean fallbackToDefaultTenant)
protected void	registerEndpoint(org.springframework.kafka.config.KafkaListenerEndpoint endpoint, org.springframework.kafka.config.KafkaListenerContainerFactory<?> factory) Register a new KafkaListenerEndpoint alongside the KafkaListenerContainerFactory to use to create the underlying container.
protected String	resolve(String value)
protected org.springframework.kafka.config.KafkaListenerContainerFactory<?>	resolveContainerFactory(org.springframework.kafka.config.KafkaListenerEndpoint endpoint, org.springframework.kafka.config.KafkaListenerContainerFactory<?> containerFactory)
protected Object	resolveExpression(String value)
protected Integer	resolveExpressionAsInteger(String value, String attribute)
protected String	resolveExpressionAsString(String value, String attribute)
protected Pattern	resolvePattern(KafkaInboundChannelModel channelModel)
protected Properties	resolveProperties(List<KafkaInboundChannelModel.CustomProperty> consumerProperties)
protected Collection<String>	resolveTopics(KafkaInboundChannelModel channelDefinition)
protected void	resolveTopics(Object resolvedValue, List<String> result, KafkaInboundChannelModel channelDefinition)
void	setBeanFactory(org.springframework.beans.factory.BeanFactory beanFactory)
void	setContainerFactory(org.springframework.kafka.config.KafkaListenerContainerFactory<?> containerFactory)
void	setContainerFactoryBeanName(String containerFactoryBeanName)
void	setEndpointRegistry(org.springframework.kafka.config.KafkaListenerEndpointRegistry endpointRegistry)
void	setKafkaOperations(org.springframework.kafka.core.KafkaOperations<Object,Object> kafkaOperations)
void	unregisterChannelModel(ChannelModel channelModel, String tenantId, EventRepositoryService eventRepositoryService)

Methods inherited from class java.lang.Object

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

Field Detail

CHANNEL_ID_PREFIX

public static final String CHANNEL_ID_PREFIX

See Also:

Constant Field Values

logger

protected final org.slf4j.Logger logger

kafkaOperations

protected org.springframework.kafka.core.KafkaOperations<Object,Object> kafkaOperations

endpointRegistry

protected org.springframework.kafka.config.KafkaListenerEndpointRegistry endpointRegistry

containerFactoryBeanName

protected String containerFactoryBeanName

containerFactory

protected org.springframework.kafka.config.KafkaListenerContainerFactory<?> containerFactory

beanFactory

protected org.springframework.beans.factory.BeanFactory beanFactory

resolver

protected org.springframework.beans.factory.config.BeanExpressionResolver resolver

embeddedValueResolver

protected org.springframework.util.StringValueResolver embeddedValueResolver

expressionContext

protected org.springframework.beans.factory.config.BeanExpressionContext expressionContext

Constructor Detail

KafkaChannelDefinitionProcessor

public KafkaChannelDefinitionProcessor()

Method Detail

canProcess

public boolean canProcess(ChannelModel channelModel)

Specified by:

canProcess in interface ChannelModelProcessor

registerChannelModel

public void registerChannelModel(ChannelModel channelModel,
                                 String tenantId,
                                 EventRegistry eventRegistry,
                                 EventRepositoryService eventRepositoryService,
                                 boolean fallbackToDefaultTenant)

Specified by:

registerChannelModel in interface ChannelModelProcessor

createKafkaListenerEndpoint

protected org.springframework.kafka.config.KafkaListenerEndpoint createKafkaListenerEndpoint(KafkaInboundChannelModel channelModel,
                                                                                             String tenantId,
                                                                                             EventRegistry eventRegistry)

processOutboundDefinition

protected void processOutboundDefinition(KafkaOutboundChannelModel channelModel)

resolveExpressionAsInteger

protected Integer resolveExpressionAsInteger(String value,
                                             String attribute)

resolveExpressionAsString

protected String resolveExpressionAsString(String value,
                                           String attribute)

resolveTopics

protected Collection<String> resolveTopics(KafkaInboundChannelModel channelDefinition)

resolveTopics

protected void resolveTopics(Object resolvedValue,
                             List<String> result,
                             KafkaInboundChannelModel channelDefinition)

resolvePattern

protected Pattern resolvePattern(KafkaInboundChannelModel channelModel)

resolveExpression

protected Object resolveExpression(String value)

createMessageListener

protected org.springframework.kafka.listener.GenericMessageListener<org.apache.kafka.clients.consumer.ConsumerRecord<Object,Object>> createMessageListener(EventRegistry eventRegistry,
                                                                                                                                                           InboundChannelModel inboundChannelModel)

unregisterChannelModel

public void unregisterChannelModel(ChannelModel channelModel,
                                   String tenantId,
                                   EventRepositoryService eventRepositoryService)

Specified by:

unregisterChannelModel in interface ChannelModelProcessor

registerEndpoint

protected void registerEndpoint(org.springframework.kafka.config.KafkaListenerEndpoint endpoint,
                                org.springframework.kafka.config.KafkaListenerContainerFactory<?> factory)

Register a new KafkaListenerEndpoint alongside the KafkaListenerContainerFactory to use to create the underlying container.

The factory may be null if the default factory has to be used for that endpoint.

resolveContainerFactory

protected org.springframework.kafka.config.KafkaListenerContainerFactory<?> resolveContainerFactory(org.springframework.kafka.config.KafkaListenerEndpoint endpoint,
                                                                                                    org.springframework.kafka.config.KafkaListenerContainerFactory<?> containerFactory)

getEndpointId

protected String getEndpointId(ChannelModel channelModel,
                               String tenantId)

getEndpointGroupId

protected String getEndpointGroupId(KafkaInboundChannelModel channelDefinition,
                                    String id)

resolve

protected String resolve(String value)

resolveProperties

protected Properties resolveProperties(List<KafkaInboundChannelModel.CustomProperty> consumerProperties)

setBeanFactory

public void setBeanFactory(org.springframework.beans.factory.BeanFactory beanFactory)
                    throws org.springframework.beans.BeansException

Specified by:

setBeanFactory in interface org.springframework.beans.factory.BeanFactoryAware

Throws:

org.springframework.beans.BeansException

getKafkaOperations

public org.springframework.kafka.core.KafkaOperations<Object,Object> getKafkaOperations()

setKafkaOperations

public void setKafkaOperations(org.springframework.kafka.core.KafkaOperations<Object,Object> kafkaOperations)

getEndpointRegistry

public org.springframework.kafka.config.KafkaListenerEndpointRegistry getEndpointRegistry()

setEndpointRegistry

public void setEndpointRegistry(org.springframework.kafka.config.KafkaListenerEndpointRegistry endpointRegistry)

getContainerFactoryBeanName

public String getContainerFactoryBeanName()

setContainerFactoryBeanName

public void setContainerFactoryBeanName(String containerFactoryBeanName)

getContainerFactory

public org.springframework.kafka.config.KafkaListenerContainerFactory<?> getContainerFactory()

setContainerFactory

public void setContainerFactory(org.springframework.kafka.config.KafkaListenerContainerFactory<?> containerFactory)