Package org.frankframework.receivers

Class Receiver<M>

java.lang.Object
org.frankframework.core.TransactionAttributes
org.frankframework.receivers.Receiver<M>
All Implemented Interfaces:
FrankElement, HasApplicationContext, HasName, HasSender, HasTransactionAttribute, IbisExceptionListener, IConfigurable, IHasProcessState<M>, IMessageHandler<M>, IProvidesMessageBrowsers<M>, IScopeProvider, IThreadCountControllable, ManagableLifecycle, NameAware, ConfigurableLifecycle, EventThrowing, org.springframework.beans.factory.Aware, org.springframework.context.ApplicationContextAware, org.springframework.context.Lifecycle, org.springframework.context.Phased, org.springframework.context.SmartLifecycle
@Category(BASIC) @FrankDocGroup(OTHER) public class Receiver<M> extends TransactionAttributes implements ManagableLifecycle, IMessageHandler<M>, IProvidesMessageBrowsers<M>, EventThrowing, IbisExceptionListener, HasSender, FrankElement, IThreadCountControllable, NameAware
Wrapper for a listener that specifies a channel for the incoming messages of a specific Adapter. By choosing a listener, the Frank developer determines how the messages are received. For example, an ApiListener receives RESTful HTTP requests and a JavaListener receives messages from direct Java calls.

Apart from wrapping the listener, a Receiver can be configured to store received messages and to keep track of the processed / failed status of these messages.

There are two kinds of listeners: synchronous listeners and asynchronous listeners. Synchronous listeners are expected to return a response. The system that triggers the receiver typically waits for a response before proceeding its operation. When a ApiListener receives a HTTP request, the listener is expected to return a HTTP response. Asynchronous listeners are not expected to return a response. The system that triggers the listener typically continues without waiting for the adapter to finish. When a receiver contains an asynchronous listener, it can have a sender that sends the transformed message to its destination. Receivers with an asynchronous listener can also have an error sender that is used by the receiver to send error messages. In other words: if the result state is SUCCESS then the message is sent by the ordinary sender, while the error sender is used if the result state is ERROR.

Transaction control

If transacted is set to true, messages will be received and processed under transaction control. This means that after a message has been read and processed and the transaction has ended, one of the following apply:
situationinput listenerPipelineinProcess storagesummary of effect
successfulmessage read and committedmessage processedunchangedmessage processed
procesing failedmessage read and committedmessage processing failed and rolled backunchangedmessage only transferred from listener to errroSender
listening failedunchanged: listening rolled backno processing performedunchangedno changes, input message remains on input available for listener
transfer to inprocess storage failedunchanged: listening rolled backno processing performedunchangedno changes, input message remains on input available for listener
If the application or the server crashes in the middle of one or more transactions, these transactions will be recovered and rolled back after the server/application is restarted. Then always exactly one of the following applies for any message touched at any time by Ibis by a transacted receiver:
  • It is processed correctly by the pipeline and removed from the input-queue, not present in inProcess storage
  • It is not processed at all by the pipeline, or processing by the pipeline has been rolled back; the message is removed from the input queue and either (one of) still in inProcess storage

commit or rollback
If transacted is set to true, messages will be either committed or rolled back. All message-processing transactions are committed, unless one or more of the following apply:

  • The PipeLine is transacted and the exitState of the pipeline is not equal to SUCCESS
  • a PipeRunException or another runtime-exception has been thrown by any Pipe or by the PipeLine
  • the setRollBackOnly() method has been called on the userTransaction (not accessible by Pipes)

Since:
4.2
Author:
Gerrit van Brakel

  • Field Details

    • TXSUPPORTED

      public static final org.springframework.transaction.TransactionDefinition TXSUPPORTED

    • TXREQUIRED

      public static final org.springframework.transaction.TransactionDefinition TXREQUIRED

    • TXNEW_CTRL

      public static final org.springframework.transaction.TransactionDefinition TXNEW_CTRL

    • THREAD_CONTEXT_KEY_TYPE

      public static final String THREAD_CONTEXT_KEY_TYPE
      See Also:

    • RCV_CONFIGURED_MONITOR_EVENT

      public static final String RCV_CONFIGURED_MONITOR_EVENT
      See Also:

    • RCV_CONFIGURATIONEXCEPTION_MONITOR_EVENT

      public static final String RCV_CONFIGURATIONEXCEPTION_MONITOR_EVENT
      See Also:

    • RCV_STARTED_RUNNING_MONITOR_EVENT

      public static final String RCV_STARTED_RUNNING_MONITOR_EVENT
      See Also:

    • RCV_SHUTDOWN_MONITOR_EVENT

      public static final String RCV_SHUTDOWN_MONITOR_EVENT
      See Also:

    • RCV_SUSPENDED_MONITOR_EVENT

      public static final String RCV_SUSPENDED_MONITOR_EVENT
      See Also:

    • RCV_RESUMED_MONITOR_EVENT

      public static final String RCV_RESUMED_MONITOR_EVENT
      See Also:

    • RCV_THREAD_EXIT_MONITOR_EVENT

      public static final String RCV_THREAD_EXIT_MONITOR_EVENT
      See Also:

    • RCV_MESSAGE_TO_ERRORSTORE_EVENT

      public static final String RCV_MESSAGE_TO_ERRORSTORE_EVENT
      See Also:

    • RCV_MESSAGE_LOG_COMMENTS

      public static final String RCV_MESSAGE_LOG_COMMENTS
      See Also:

    • RCV_SUSPENSION_MESSAGE_THRESHOLD

      public static final int RCV_SUSPENSION_MESSAGE_THRESHOLD
      See Also:

    • DEFAULT_MAX_BACKOFF_DELAY_KEY

      public static final String DEFAULT_MAX_BACKOFF_DELAY_KEY
      See Also:

    • DEFAULT_MAX_BACKOFF_DELAY

      public static final int DEFAULT_MAX_BACKOFF_DELAY
      Should be smaller than the transaction timeout as the delay takes place within the transaction. WebSphere default transaction timeout is 120.
      See Also:

    • RETRY_FLAG_SESSION_KEY

      public static final String RETRY_FLAG_SESSION_KEY
      See Also:

    • runState

      protected final RunStateManager runState

  • Constructor Details

    • Receiver

      public Receiver()

  • Method Details

    • getLogPrefix

      protected String getLogPrefix()

    • info

      protected void info(String msg)
      sends an informational message to the log and to the messagekeeper of the adapter

    • warn

      protected void warn(String msg)
      sends a warning to the log and to the messagekeeper of the adapter

    • error

      protected void error(@Nonnull String msg, @Nullable Throwable t)
      sends a error message to the log and to the messagekeeper of the adapter

    • setApplicationContext

      public final void setApplicationContext(@Nonnull org.springframework.context.ApplicationContext context)
      Specified by:
      setApplicationContext in interface org.springframework.context.ApplicationContextAware

    • openAllResources

      protected void openAllResources() throws ListenerException, TimeoutException
      Throws:
      ListenerException
      TimeoutException

    • tellResourcesToStop

      protected void tellResourcesToStop()
      must lead to a 'closeAllResources()' and runstate must be 'STOPPING' if IPushingListener -> call closeAllResources() if IPullingListener -> PullingListenerContainer has to call closeAllResources();

    • closeAllResources

      protected void closeAllResources()
      Should only close resources when in state stopping (or error)! this should be the only trigger to change the state to stopped On exit resources must be 'closed' so the receiver RunState can be set to 'STOPPED'

    • propagateName

      protected void propagateName()

    • configure

      public void configure() throws ConfigurationException
      This method is called by the Adapter to let the receiver do things to initialize itself before the startListening method is called.
      Specified by:
      configure in interface IConfigurable
      Overrides:
      configure in class TransactionAttributes
      Throws:
      ConfigurationException - when initialization did not succeed.
      See Also:

    • calculateAdjustedMaxBackoffDelay

      protected int calculateAdjustedMaxBackoffDelay(Integer configuredMaxBackoffDelay)

    • start

      public void start()
      Specified by:
      start in interface ConfigurableLifecycle
      Specified by:
      start in interface org.springframework.context.Lifecycle

    • stop

      public void stop()
      Specified by:
      stop in interface org.springframework.context.Lifecycle

    • knownProcessStates

      public Set<ProcessState> knownProcessStates()
      Description copied from interface: IHasProcessState
      Provides the set of ProcessStates used by this listener.
      Specified by:
      knownProcessStates in interface IHasProcessState<M>

    • targetProcessStates

      public Map<ProcessState,Set<ProcessState>> targetProcessStates()
      Description copied from interface: IHasProcessState
      Provides the set of ProcessStates that a message in the specified state can be moved to, e.g. from a MessageBrowser for that state.
      Specified by:
      targetProcessStates in interface IHasProcessState<M>

    • changeProcessState

      public RawMessageWrapper<M> changeProcessState(RawMessageWrapper<M> message, ProcessState toState, String reason) throws ListenerException
      Change message process state, via the listener of the receiver. The listener must support IHasProcessState. As an extra side-effect of this method implementation, if the toState is ProcessState.AVAILABLE then any recorded processing history status is reset so that a message can be manually retried.
      Specified by:
      changeProcessState in interface IHasProcessState<M>
      Parameters:
      message - Message for which to change state.
      toState - Desired state of the message
      reason - Reason for changing the state
      Returns:
      Updated message
      Throws:
      ListenerException - thrown if changing the state by the listener fails.
      IllegalStateException - thrown if the listener does not support changing process states.

    • getMessageBrowser

      public IMessageBrowser<M> getMessageBrowser(ProcessState state)
      Description copied from interface: IProvidesMessageBrowsers
      returns a browser of messages that are in ProcessState 'state', and are stored in a storage managed by the listener itself (as opposed to a storage configured as a messageLog or errorStorage in the configuration).
      Specified by:
      getMessageBrowser in interface IProvidesMessageBrowsers<M>

    • startProcessingMessage

      protected void startProcessingMessage()

    • finishProcessingMessage

      protected void finishProcessingMessage(long processingDuration)

    • moveInProcessToError

      public void moveInProcessToError(RawMessageWrapper<M> rawMessageWrapper, PipeLineSession session, Instant receivedDate, String comments, org.springframework.transaction.TransactionDefinition txDef)
      Move a message from the "in process" state or storage, to the error state or storage.
      Parameters:
      rawMessageWrapper - Wrapper for the raw message, may be an instance of RawMessageWrapper or MessageWrapper. If an instance of RawMessageWrapper then the IListener will be used to extract the full Message object to be sent to the error storage.
      session - PipeLineSession of the process.
      receivedDate - Timestamp of when the message was received.
      comments - Processing comments and error message regarding the reason the message was rejected.
      txDef - TransactionDefinition for the transaction to be used for moving the message to error state / storage.

    • processRequest

      public Message processRequest(IPushingListener<M> origin, @Nonnull MessageWrapper<M> messageWrapper, @Nonnull PipeLineSession session) throws ListenerException
      Process the received message with IMessageHandler.processRequest(IPushingListener, MessageWrapper, PipeLineSession).
      A messageId is generated that is unique and consists of the name of this listener and a GUID.

      If the receiver is transactional, then this method will enforce the transactional requirements specified by HasTransactionAttribute.setTransactionAttribute(TransactionAttribute).

      N.B. callers of this method should clear the remaining ThreadContext if it's not to be returned to their callers.

      Specified by:
      processRequest in interface IMessageHandler<M>
      Throws:
      ListenerException

    • processRawMessage

      public void processRawMessage(IListener<M> origin, RawMessageWrapper<M> rawMessage, @Nonnull PipeLineSession session, boolean retryStatusAlreadyChecked) throws ListenerException
      This method processes the raw message from the listener.
      The method assumes that a transaction has been started where necessary.
      Specified by:
      processRawMessage in interface IMessageHandler<M>
      Throws:
      ListenerException

    • retryMessage

      public void retryMessage(String storageKey) throws ListenerException
      Throws:
      ListenerException

    • updateMessageReceiveCount

      @Nonnull protected Receiver.ProcessStatusCacheItem updateMessageReceiveCount(@Nonnull RawMessageWrapper<M> rawMessageWrapper)
      Call this method exactly once per every try to process a message to initialize the internally cached receive-count for the messageId. If the listener implements IKnowsDeliveryCount then the receiveCount will be set to the IKnowsDeliveryCount.getDeliveryCount(RawMessageWrapper).
      Parameters:
      rawMessageWrapper - The raw message for which to set the receiveCount.

    • getCachedErrorMessage

      @Nullable public String getCachedErrorMessage(@Nonnull RawMessageWrapper<M> rawMessageWrapper)

    • getDeliveryCount

      public int getDeliveryCount(RawMessageWrapper<M> rawMessage)
      Get the delivery-count for the message. If the listener implements IKnowsDeliveryCount then get the delivery count from the listener, otherwise get it from the internal process-result cache.
      Parameters:
      rawMessage - RawMessageWrapper for which to retrieve the delivery count
      Returns:
      Number of times the message has been delivered, minimum is always 1 (for the first delivery of the message).

    • isDeliveryRetryLimitExceededAfterMessageProcessed

      protected boolean isDeliveryRetryLimitExceededAfterMessageProcessed(@Nonnull RawMessageWrapper<M> messageWrapper)
      Returns true if the message should no longer be retried after it has failed in processing.
      Parameters:
      messageWrapper - Message for which to check delivery count
      Returns:
      true if message should no longer be retried, false if it should.

    • isDeliveryRetryLimitExceededBeforeMessageProcessing

      protected boolean isDeliveryRetryLimitExceededBeforeMessageProcessing(RawMessageWrapper<M> rawMessageWrapper, PipeLineSession session, boolean manualRetry) throws ListenerException
      returns true if message should not be processed when it is delivered.
      Throws:
      ListenerException

    • exceptionThrown

      public void exceptionThrown(HasName object, Throwable t)
      Description copied from interface: IbisExceptionListener
      Inform the implementing class that the exception t occurred in object.
      Specified by:
      exceptionThrown in interface IbisExceptionListener

    • exceptionThrown

      public void exceptionThrown(String errorMessage, Throwable t)

    • getEventSourceName

      public String getEventSourceName()
      Specified by:
      getEventSourceName in interface EventThrowing

    • throwEvent

      protected void throwEvent(String eventCode)

    • resetBackoffDelay

      public void resetBackoffDelay()

    • increaseBackoffIntervalAndWait

      public void increaseBackoffIntervalAndWait(Throwable t, String description)

    • suspendReceiverThread

      public void suspendReceiverThread(int delayTimeInSeconds)
      Suspend the receiver for delayTimeInSeconds seconds
      Parameters:
      delayTimeInSeconds - Number of seconds the receiver thread should be suspended from processing new messages.

    • isThreadCountReadable

      public boolean isThreadCountReadable()
      Specified by:
      isThreadCountReadable in interface IThreadCountControllable

    • isThreadCountControllable

      public boolean isThreadCountControllable()
      Specified by:
      isThreadCountControllable in interface IThreadCountControllable

    • getCurrentThreadCount

      public int getCurrentThreadCount()
      Specified by:
      getCurrentThreadCount in interface IThreadCountControllable

    • getMaxThreadCount

      public int getMaxThreadCount()
      Specified by:
      getMaxThreadCount in interface IThreadCountControllable

    • increaseThreadCount

      public void increaseThreadCount()
      Specified by:
      increaseThreadCount in interface IThreadCountControllable

    • decreaseThreadCount

      public void decreaseThreadCount()
      Specified by:
      decreaseThreadCount in interface IThreadCountControllable

    • setRunState

      @Protected public void setRunState(RunState state)
      Changes runstate. Always stops the receiver when state is `**ERROR**`

    • getRunState

      public RunState getRunState()
      Get the runstate of this receiver.
      Specified by:
      getRunState in interface ManagableLifecycle

    • getPhase

      public int getPhase()
      Specified by:
      getPhase in interface ConfigurableLifecycle
      Specified by:
      getPhase in interface org.springframework.context.Phased
      Specified by:
      getPhase in interface org.springframework.context.SmartLifecycle

    • isInRunState

      public boolean isInRunState(RunState someRunState)

    • getProcessStatistics

      protected io.micrometer.core.instrument.DistributionSummary getProcessStatistics(int threadsProcessing)

    • isOnErrorContinue

      public boolean isOnErrorContinue()

    • getMessagesReceived

      public double getMessagesReceived()
      get the number of messages received by this receiver.

    • getMessagesRetried

      public double getMessagesRetried()
      get the number of duplicate messages received this receiver.

    • getMessagesRejected

      public double getMessagesRejected()
      Get the number of messages rejected (discarded or put in errorStorage).

    • getLastMessageDate

      public long getLastMessageDate()

    • resetNumberOfExceptionsCaughtWithoutMessageBeingReceived

      public void resetNumberOfExceptionsCaughtWithoutMessageBeingReceived()

    • toString

      public String toString()
      Returns a toString of this class by introspection and the toString() value of its listener.
      Overrides:
      toString in class Object
      Returns:
      Description of the Return Value

    • setListener

      public void setListener(IListener<M> newListener)
      Sets the listener used to receive messages from.

    • setSender

      public void setSender(ICorrelatedSender sender)
      Sender to which the response (output of PipeLine) should be sent. Applies if the receiver has an asynchronous listener. N.B. Sending correlated responses via this sender is not supported.

    • setErrorStorage

      public void setErrorStorage(ITransactionalStorage<Serializable> errorStorage)
      Storage to keep track of messages that failed processing

    • setMessageLog

      public void setMessageLog(ITransactionalStorage<Serializable> messageLog)
      Storage to keep track of all messages processed correctly

    • setName

      public void setName(String newName)
      Sets the name of the Receiver, as known to the Adapter. If the listener getName() is empty, the name of this object is given to the listener.
      Specified by:
      setName in interface NameAware

    • setOnError

      public void setOnError(Receiver.OnError value)
      One of 'continue', 'recover' or 'close'. Controls the behaviour of the Receiver, when it encounters an error during processing of a message.
      Default value
      CONTINUE

    • setNumThreads

      public void setNumThreads(int newNumThreads)
      The number of threads that may execute a Pipeline concurrently (only for pulling listeners)
      Default value
      1

    • setNumThreadsPolling

      public void setNumThreadsPolling(int i)
      The number of threads that are actively polling for messages concurrently. '0' means 'limited only by numthreads' (only for pulling listeners)
      Default value
      1

    • setPollInterval

      public void setPollInterval(int i)
      The number of seconds waited after an unsuccessful poll attempt, before another poll attempt is made. Only for polling listeners, not for e.g. jms, webservice or javaListeners
      Default value
      10

    • setStartTimeout

      public void setStartTimeout(int i)
      timeout (in seconds) to start receiver. If this timeout is exceeded, the Receiver startup is aborted and all resources closed and the receiver will be in state EXCEPTION_STARTING and a new start command may be issued again.

    • setStopTimeout

      public void setStopTimeout(int i)
      timeout (in seconds) to stop receiver. If this timeout is exceeded, stopping will be aborted and the receiver will be in state EXCEPTION_STOPPING. The receiver will no longer be running but some resources might not have been cleaned up properly.

    • setCheckForDuplicates

      public void setCheckForDuplicates(boolean b)
      If set to true, each message is checked for presence in the messageLog. If already present, it is not processed again. Only required for non XA compatible messaging. Requires messageLog!
      Default value
      false

    • setCheckForDuplicatesMethod

      public void setCheckForDuplicatesMethod(Receiver.CheckForDuplicatesMethod method)
      (Only used when checkForDuplicates=true) Indicates whether the messageid or the correlationid is used for checking presence in the message log
      Default value
      MESSAGEID

    • setMaxDeliveries

      @Deprecated(forRemoval=true, since="9.0") @ConfigurationWarning("This property has been deprecated, please use maxRetries instead.") public void setMaxDeliveries(Integer i)
      Deprecated, for removal: This API element is subject to removal in a future version.
      The maximum retry count after which to stop processing the message. If equal to or lower than 0, the retry count is ignored. This property is deprecated -- use maxRetries instead. Until removal of this property, the code will treat this property as the same as maxRetries. If both are set in a configuration, then the highest value is used.

    • setMaxRetries

      public void setMaxRetries(Integer i)
      The number of times a processing attempt is automatically retried after an exception is caught or rollback is experienced. If maxRetries < 0 the number of attempts is infinite
      Default value
      1, or 3 for JMS Listeners or other listeners implementing IKnowsDeliveryCount.

    • setProcessResultCacheSize

      public void setProcessResultCacheSize(int processResultCacheSize)
      Size of the cache to keep process results, used by maxRetries
      Default value
      100

    • setReturnedSessionKeys

      @Deprecated(forRemoval=true, since="7.9.0") @ConfigurationWarning("attribute is no longer used. Please use attribute returnedSessionKeys of the JavaListener if the set of sessionsKeys that can be returned to callers session must be limited.") public void setReturnedSessionKeys(String string)
      Deprecated, for removal: This API element is subject to removal in a future version.

    • setCorrelationIDXPath

      public void setCorrelationIDXPath(String string)
      XPath expression to extract correlationId from message

    • setCorrelationIDNamespaceDefs

      public void setCorrelationIDNamespaceDefs(String correlationIDNamespaceDefs)
      Namespace definitions for correlationIDXPath. Must be in the form of a comma or space separated list of prefix=namespaceuri-definitions

    • setCorrelationIDStyleSheet

      public void setCorrelationIDStyleSheet(String string)
      Stylesheet to extract correlationID from message

    • setLabelXPath

      public void setLabelXPath(String string)
      XPath expression to extract label from message

    • setLabelNamespaceDefs

      public void setLabelNamespaceDefs(String labelNamespaceDefs)
      Namespace definitions for labelXPath. Must be in the form of a comma or space separated list of prefix=namespaceuri-definitions

    • setLabelStyleSheet

      public void setLabelStyleSheet(String string)
      Stylesheet to extract label from message

    • setChompCharSize

      public void setChompCharSize(String string)
      If set (>=0) and the character data length inside a xml element exceeds this size, the character data is chomped (with a clear comment)

    • setElementToMove

      public void setElementToMove(String string)
      If set, the character data in this XML element is stored inside a session key and in the message it is replaced by a reference to this session key: {sessionKey: elementToMoveSessionKey}

    • setElementToMoveSessionKey

      public void setElementToMoveSessionKey(String string)
      (Only used when elementToMove or elementToMoveChain is set) Name of the session key wherein the character data is stored
      Default value
      ref_ + the name of the element

    • setElementToMoveChain

      public void setElementToMoveChain(String string)
      Like elementToMove but element is preceded with all ancestor elements and separated by semicolons (e.g. adapter;pipeline;pipe)

    • setRemoveCompactMsgNamespaces

      public void setRemoveCompactMsgNamespaces(boolean b)

    • setHideRegex

      public void setHideRegex(String hideRegex)
      Regular expression to mask strings in the errorStore/logStore and logfiles. Every character between to the strings in this expression will be replaced by a '*'. For example, the regular expression (?<=<party>).*?(?=</party>) will replace every character between keys <party> and </party>

    • setHideMethod

      public void setHideMethod(IMessageBrowser.HideMethod hideMethod)
      Only used when hideRegex is not empty
      Default value
      all

    • setHiddenInputSessionKeys

      public void setHiddenInputSessionKeys(String string)
      Comma separated list of keys of session variables which are available when the PipelineSession is created and of which the value will not be shown in the log (replaced by asterisks)

    • setForceRetryFlag

      public void setForceRetryFlag(boolean b)
      If set to true, every message read will be processed as if it is being retried, by setting a session variable to "retry"<M>.
      Default value
      false

    • setNumberOfExceptionsCaughtWithoutMessageBeingReceivedThreshold

      public void setNumberOfExceptionsCaughtWithoutMessageBeingReceivedThreshold(int number)
      Number of connection attempts to put the adapter in warning status
      Default value
      5

    • setMaxBackoffDelay

      public void setMaxBackoffDelay(Integer maxBackoffDelaySeconds)
      After a message has an error in processing, there is a small delay before processing the next message before processing the next message or retrying the failed message. This is so that errors coming from external systems so not overload those external systems.

      The delay doubles after every failure, until the maximum set here is reached. See: https://en.wikipedia.org/wiki/Exponential_backoff.

      If the transaction timeout can be determined, then the backoff-delay is capped by half the transaction timeout to avoid messages automatically timing out.

      There is no backoff-time after a message is successfully processed. After a message is successfully processed, the actual backoff-time is reset to 1 second.

      If set to 0, then there is no delay after messages that had an error.

      If this is not set on the receiver, then a default is taken from the configuration property ${receiver.defaultMaxBackoffDelay} which defaults to 60 seconds.

      Parameters:
      maxBackoffDelaySeconds - Maximum backoff-time in seconds before retrying a message, after an error occurred during processing.