org.adroitlogic.x.base.trp

Class ScheduledMessageInjector<C extends ScheduledIngressConfig,R extends FetchReference<T>,T>

java.lang.Object

org.adroitlogic.x.base.trp.ScheduledMessageInjector<C,R,T>

All Implemented Interfaces:

org.quartz.Job

Direct Known Subclasses:

ListenerAwareScheduledMessageInjector

public abstract class ScheduledMessageInjector<C extends ScheduledIngressConfig,R extends FetchReference<T>,T>
extends Object
implements org.quartz.Job

Abstract base class for scheduled message injecting. This implements the quartz Job interface and handle the functionality of the scheduled job. This class should be a stateless class since for each job iteration new instance of this class will be created to execute the job functionality.

In each job iteration this will fetch the existing resources list in the configured location and inject each resource one by one as a new message to the engine. Before injecting a new message to the engine this will compare the existing polling count and execution count to decide whether is it allowed to proceed based on the configured concurrent polling count and concurrent execution count. Each resource will be processed after getting lock for that resource id to make sure that resource won't be picked up and process by another polling thread.

Since:

17.01

Author:

Ruwan Linton, Manjula Piyumal

Field Summary

Fields

Modifier and Type	Field and Description
(package private) static String	INGRESS_CONFIG_KEY
(package private) static String	JOB_CONTEXT_KEY
(package private) static String	LISTENER_KEY
protected org.adroitlogic.x.logging.Logger	logger
(package private) static String	MESSAGE_RECEIVER_KEY
(package private) static String	POLLING_EXECUTOR_KEY

Constructor Summary

Constructors

Constructor and Description
ScheduledMessageInjector()

Method Summary

Modifier and Type	Method and Description
protected abstract void	afterProcess(C config, T fetchRef, XMessageContext mc) Abstract method to handle all the tasks which should be executed after successfully injecting the polling resource to the engine as a new message
void	execute(org.quartz.JobExecutionContext ctx)
protected abstract List<R>	fetchList(C config) Abstract method to implement the logic for fetching the existing list of resources in the configured location for the scheduled job.
protected abstract XMessageContext	fetchMessage(C config, T fetchRef) Abstract method to construct a new message from the given FetchReference object.
protected abstract void	onExecutionLimitExceeded(XMessageContext messageContext, C config, R fetchReference) Abstract method which is exposed as an extension point to handle execution count exceed event for polling message injectors.
protected abstract void	onFailure(C config, T fetchRef, XMessageContext mc, Throwable t) Abstract method to handle all the tasks which should be executed after getting any failures while trying to inject the resources as a new message to the engine
protected boolean	shouldAcquireLock(C config) This method defines whether the polling thread should acquire a lock for each of the fetch references returned by the fetchList(C) method, before invoking the fetchMessage(C, T) method on that fetch reference.

Methods inherited from class java.lang.Object

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

Field Detail

INGRESS_CONFIG_KEY

static final String INGRESS_CONFIG_KEY

See Also:

Constant Field Values

LISTENER_KEY

static final String LISTENER_KEY

See Also:

Constant Field Values

JOB_CONTEXT_KEY

static final String JOB_CONTEXT_KEY

See Also:

Constant Field Values

MESSAGE_RECEIVER_KEY

static final String MESSAGE_RECEIVER_KEY

See Also:

Constant Field Values

POLLING_EXECUTOR_KEY

static final String POLLING_EXECUTOR_KEY

See Also:

Constant Field Values

logger

protected final org.adroitlogic.x.logging.Logger logger

Constructor Detail

ScheduledMessageInjector

public ScheduledMessageInjector()

Method Detail

execute

public final void execute(org.quartz.JobExecutionContext ctx)
                   throws org.quartz.JobExecutionException

Specified by:

execute in interface org.quartz.Job

Throws:

org.quartz.JobExecutionException

shouldAcquireLock

protected boolean shouldAcquireLock(C config)

This method defines whether the polling thread should acquire a lock for each of the fetch references returned by the fetchList(C) method, before invoking the fetchMessage(C, T) method on that fetch reference. In a clustered deployment, this would be a cluster lock, while this will be a JVM level lock otherwise.

Parameters:

config - ScheduledIngressConfig instance of the connector

Returns:

true if a lock should be acquired, or false otherwise

onExecutionLimitExceeded

protected abstract void onExecutionLimitExceeded(XMessageContext messageContext,
                                                 C config,
                                                 R fetchReference)

Abstract method which is exposed as an extension point to handle execution count exceed event for polling message injectors. All the tasks including, resource handling of the already constructed messageContext, should be handled within this method

Parameters:

messageContext - prepared message context for the polling iteration

config - connector config for the polling connector

fetchReference - reference for the current polling resource

fetchList

protected abstract List<R> fetchList(C config)

Abstract method to implement the logic for fetching the existing list of resources in the configured location for the scheduled job. Resources in the returned list from this method will be injected one by one as a new message to the engine if it's allowed to process for the current thread.

Parameters:

config - connector configuration of the relevant polling connector

Returns:

the list of existing resources as FetchReference objects in the configured location

fetchMessage

protected abstract XMessageContext fetchMessage(C config,
                                                T fetchRef)

Abstract method to construct a new message from the given FetchReference object. Implementation of this method should be specific to the type of the fetched resource.

Parameters:

config - connector configuration of the relevant polling connector

fetchRef - the reference for the resource to be fetched and injected to the engine after constructing a new message

Returns:

the constructed message context from the given resource

afterProcess

protected abstract void afterProcess(C config,
                                     T fetchRef,
                                     XMessageContext mc)

Abstract method to handle all the tasks which should be executed after successfully injecting the polling resource to the engine as a new message

Parameters:

config - connector configuration of the relevant polling connector

fetchRef - the reference for the resource which has been fetched and injected to the engine

mc - the message context constructed and injected to the engine from the polling resource

onFailure

protected abstract void onFailure(C config,
                                  T fetchRef,
                                  XMessageContext mc,
                                  Throwable t)

Abstract method to handle all the tasks which should be executed after getting any failures while trying to inject the resources as a new message to the engine

Parameters:

config - connector configuration of the relevant polling connector

fetchRef - the reference for the resource which has been tried to fetch and inject to the engine

mc - the message context constructed and injected to the engine from the polling resource

t - the throwable object related to the failure