Class PostgresSubscribableChannel

java.lang.Object

org.springframework.integration.context.IntegrationObjectSupport

org.springframework.integration.channel.AbstractMessageChannel

org.springframework.integration.channel.AbstractSubscribableChannel

org.springframework.integration.jdbc.channel.PostgresSubscribableChannel

All Implemented Interfaces:

Aware, BeanFactoryAware, BeanNameAware, DisposableBean, InitializingBean, ApplicationContextAware, ComponentSourceAware, ExpressionCapable, IntegrationPattern, PostgresChannelMessageTableSubscriber.Subscription, NamedComponent, IntegrationManagement, SubscribableChannelManagement, TrackableComponent, MessageChannel, SubscribableChannel, InterceptableChannel

public class PostgresSubscribableChannel
extends AbstractSubscribableChannel
implements PostgresChannelMessageTableSubscriber.Subscription

An AbstractSubscribableChannel for receiving push notifications for messages send to a group id of a JdbcChannelMessageStore. Receiving such push notifications is only possible if using a Postgres database.

In order to function, the Postgres database that is used must define a trigger for sending notifications upon newly arrived messages. This trigger is defined in the schema-postgresql.sql file within this artifact but commented out.

Since:

6.0

Author:

Rafael Winterhalter, Artem Bilan, Igor Lovich, Norbert Schneider

Nested Class Summary

Nested classes/interfaces inherited from class org.springframework.integration.channel.AbstractMessageChannel

AbstractMessageChannel.ChannelInterceptorList

Nested classes/interfaces inherited from interface org.springframework.integration.support.management.IntegrationManagement

IntegrationManagement.ManagementOverrides

Field Summary

Fields inherited from class org.springframework.integration.channel.AbstractMessageChannel

interceptors, meters

Fields inherited from class org.springframework.integration.context.IntegrationObjectSupport

EXPRESSION_PARSER, logger

Fields inherited from interface org.springframework.integration.support.management.IntegrationManagement

METER_PREFIX, RECEIVE_COUNTER_NAME, SEND_TIMER_NAME

Fields inherited from interface org.springframework.messaging.MessageChannel

INDEFINITE_TIMEOUT

Constructor Summary

Constructors

Constructor	Description
PostgresSubscribableChannel(JdbcChannelMessageStore jdbcChannelMessageStore, Object groupId, PostgresChannelMessageTableSubscriber messageTableSubscriber)	Create a subscribable channel for a Postgres database.

Method Summary

Modifier and Type	Method	Description
protected boolean	doSend(Message<?> message, long timeout)	Subclasses must implement this method.
protected MessageDispatcher	getDispatcher()
Object	getGroupId()	Return the group id for which this subscription receives notifications.
String	getRegion()	Return the region for which this subscription receives notifications.
void	notifyUpdate()	Indicate that a message was added to the represented region and group id.
protected void	onInit()	Subclasses may implement this for initialization logic.
void	setDispatcherExecutor(Executor executor)	Set the executor to use for dispatching newly received messages.
void	setErrorHandler(ErrorHandler errorHandler)	Set a ErrorHandler for messages which cannot be dispatched by this channel.
void	setRetryTemplate(org.springframework.retry.support.RetryTemplate retryTemplate)	Set the retry template to use for retries in case of exception in downstream processing
void	setTransactionManager(PlatformTransactionManager transactionManager)	Set the transaction manager to use for message processing.
boolean	subscribe(MessageHandler handler)
boolean	unsubscribe(MessageHandler handle)

Methods inherited from class org.springframework.integration.channel.AbstractSubscribableChannel

getSubscriberCount

Methods inherited from class org.springframework.integration.channel.AbstractMessageChannel

addInterceptor, addInterceptor, destroy, getComponentType, getFullChannelName, getIChannelInterceptorList, getIntegrationPatternType, getInterceptors, getMetricsCaptor, getOverrides, isLoggingEnabled, isObserved, registerMetricsCaptor, registerObservationRegistry, removeInterceptor, removeInterceptor, send, send, setDatatypes, setInterceptors, setLoggingEnabled, setMessageConverter, setObservationConvention, setShouldTrack

Methods inherited from class org.springframework.integration.context.IntegrationObjectSupport

afterPropertiesSet, extractTypeIfPossible, generateId, getApplicationContext, getApplicationContextId, getBeanDescription, getBeanFactory, getBeanName, getChannelResolver, getComponentDescription, getComponentName, getComponentSource, getConversionService, getExpression, getIntegrationProperties, getMessageBuilderFactory, getTaskScheduler, isInitialized, setApplicationContext, setBeanFactory, setBeanName, setChannelResolver, setComponentDescription, setComponentName, setComponentSource, setConversionService, setMessageBuilderFactory, setPrimaryExpression, setTaskScheduler, toString

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.springframework.integration.support.management.IntegrationManagement

getManagedName, getManagedType, getThisAs, setManagedName, setManagedType

Methods inherited from interface org.springframework.messaging.MessageChannel

send, send

Methods inherited from interface org.springframework.integration.support.context.NamedComponent

getBeanName, getComponentName

Constructor Details

PostgresSubscribableChannel

public PostgresSubscribableChannel(JdbcChannelMessageStore jdbcChannelMessageStore,
 Object groupId,
 PostgresChannelMessageTableSubscriber messageTableSubscriber)

Create a subscribable channel for a Postgres database.

Parameters:

jdbcChannelMessageStore - The message store to use for the relevant region.

groupId - The group id that is targeted by the subscription.

messageTableSubscriber - The subscriber to use for receiving notifications.

Method Details

setDispatcherExecutor

public void setDispatcherExecutor(Executor executor)

Set the executor to use for dispatching newly received messages.

Parameters:

executor - The executor to use.

setTransactionManager

public void setTransactionManager(PlatformTransactionManager transactionManager)

Set the transaction manager to use for message processing. Each message will be processed in a separate transaction

Parameters:

transactionManager - The transaction manager to use

Since:

6.0.5

See Also:

• PlatformTransactionManager

setRetryTemplate

public void setRetryTemplate(org.springframework.retry.support.RetryTemplate retryTemplate)

Set the retry template to use for retries in case of exception in downstream processing

Parameters:

retryTemplate - The retry template to use

Since:

6.0.5

See Also:

• RetryTemplate

setErrorHandler

public void setErrorHandler(ErrorHandler errorHandler)

Set a ErrorHandler for messages which cannot be dispatched by this channel. Used as a recovery callback after RetryTemplate execution throws an exception.

Parameters:

errorHandler - the ErrorHandler to use.

Since:

6.0.9

onInit

protected void onInit()

Description copied from class: IntegrationObjectSupport

Subclasses may implement this for initialization logic.

Overrides:

onInit in class AbstractMessageChannel

subscribe

public boolean subscribe(MessageHandler handler)

Specified by:

subscribe in interface SubscribableChannel

Overrides:

subscribe in class AbstractSubscribableChannel

unsubscribe

public boolean unsubscribe(MessageHandler handle)

Specified by:

unsubscribe in interface SubscribableChannel

Overrides:

unsubscribe in class AbstractSubscribableChannel

getDispatcher

protected MessageDispatcher getDispatcher()

Specified by:

getDispatcher in class AbstractSubscribableChannel

doSend

protected boolean doSend(Message<?> message,
 long timeout)

Description copied from class: AbstractMessageChannel

Subclasses must implement this method. A non-negative timeout indicates how long to wait if the channel is at capacity (if the value is 0, it must return immediately with or without success). A negative timeout value indicates that the method should block until either the message is accepted or the blocking thread is interrupted.

Overrides:

doSend in class AbstractSubscribableChannel

Parameters:

message - The message.

timeout - The timeout.

Returns:

true if the send was successful.

notifyUpdate

public void notifyUpdate()

Description copied from interface: PostgresChannelMessageTableSubscriber.Subscription

Indicate that a message was added to the represented region and group id. Note that this method might also be invoked if there are no new messages to read, for example, if another subscription already read those messages or if a new message might have arrived during a temporary connection loss.

Specified by:

notifyUpdate in interface PostgresChannelMessageTableSubscriber.Subscription

getRegion

public String getRegion()

Description copied from interface: PostgresChannelMessageTableSubscriber.Subscription

Return the region for which this subscription receives notifications.

Specified by:

getRegion in interface PostgresChannelMessageTableSubscriber.Subscription

Returns:

The relevant region of the JdbcChannelMessageStore.

getGroupId

public Object getGroupId()

Description copied from interface: PostgresChannelMessageTableSubscriber.Subscription

Return the group id for which this subscription receives notifications.

Specified by:

getGroupId in interface PostgresChannelMessageTableSubscriber.Subscription

Returns:

The group id of the PostgresSubscribableChannel.