Class MessageDispatchInterceptor

java.lang.Object
org.apache.catalina.tribes.group.ChannelInterceptorBase
org.apache.catalina.tribes.group.interceptors.MessageDispatchInterceptor
All Implemented Interfaces:
ChannelInterceptor, MessageDispatchInterceptorMBean, Heartbeat, MembershipListener
public class MessageDispatchInterceptor extends ChannelInterceptorBase implements MessageDispatchInterceptorMBean
The message dispatcher is a way to enable asynchronous communication through a channel. The dispatcher will look for the Channel.SEND_OPTIONS_ASYNCHRONOUS flag to be set, if it is, it will queue the message for delivery and immediately return to the sender.

  • Field Details

    • sm

      protected static final StringManager sm
      The string manager for this class.

    • maxQueueSize

      protected long maxQueueSize
      Maximum queue size.

    • run

      protected volatile boolean run
      Whether the queue is running.

    • useDeepClone

      protected boolean useDeepClone
      Whether to use deep clone.

    • alwaysSend

      protected boolean alwaysSend
      Whether to always send.

    • currentSize

      protected final AtomicLong currentSize
      Current queue size.

    • executor

      protected ExecutorService executor
      The executor service.

    • maxThreads

      protected int maxThreads
      Maximum number of threads.

    • maxSpareThreads

      protected int maxSpareThreads
      Maximum number of spare threads.

    • keepAliveTime

      protected long keepAliveTime
      Keep alive time.

  • Constructor Details

    • MessageDispatchInterceptor

      public MessageDispatchInterceptor()
      Default constructor.

  • Method Details

    • sendMessage

      public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException
      Description copied from interface: ChannelInterceptor
      The sendMessage method is called when a message is being sent to one more destinations. The interceptor can modify any of the parameters and then pass on the message down the stack by invoking getNext().sendMessage(destination,msg,payload).

      Alternatively the interceptor can stop the message from being sent by not invoking getNext().sendMessage(destination,msg,payload).

      If the message is to be sent asynchronous the application can be notified of completion and errors by passing in an error handler attached to a payload object.

      The ChannelMessage.getAddress contains Channel.getLocalMember, and can be overwritten to simulate a message sent from another node.

      Specified by:
      sendMessage in interface ChannelInterceptor
      Overrides:
      sendMessage in class ChannelInterceptorBase
      Parameters:
      destination - Member[] - the destination for this message
      msg - ChannelMessage - the message to be sent
      payload - InterceptorPayload - the payload, carrying an error handler and future useful data, can be null
      Throws:
      ChannelException - if a serialization error happens.
      See Also:

    • addToQueue

      public boolean addToQueue(ChannelMessage msg, Member[] destination, InterceptorPayload payload)
      Add a message to the queue.
      Parameters:
      msg - the message
      destination - the destination
      payload - the payload
      Returns:
      true if added

    • startQueue

      public void startQueue()
      Start the dispatch queue.

    • stopQueue

      public void stopQueue()
      Stop the dispatch queue.

    • setOptionFlag

      public void setOptionFlag(int flag)
      Description copied from interface: ChannelInterceptor
      Sets the option flag
      Specified by:
      setOptionFlag in interface ChannelInterceptor
      Overrides:
      setOptionFlag in class ChannelInterceptorBase
      Parameters:
      flag - int
      See Also:

    • setMaxQueueSize

      public void setMaxQueueSize(long maxQueueSize)
      Set the maximum queue size.
      Parameters:
      maxQueueSize - the maximum queue size

    • setUseDeepClone

      public void setUseDeepClone(boolean useDeepClone)
      Set whether to use deep clone.
      Parameters:
      useDeepClone - whether to use deep clone

    • getMaxQueueSize

      public long getMaxQueueSize()
      Description copied from interface: MessageDispatchInterceptorMBean
      Get the maximum queue size.
      Specified by:
      getMaxQueueSize in interface MessageDispatchInterceptorMBean
      Returns:
      the maximum queue size

    • getUseDeepClone

      public boolean getUseDeepClone()
      Get whether deep clone is used.
      Returns:
      whether deep clone is used

    • getCurrentSize

      public long getCurrentSize()
      Description copied from interface: MessageDispatchInterceptorMBean
      Get the current queue size.
      Specified by:
      getCurrentSize in interface MessageDispatchInterceptorMBean
      Returns:
      the current queue size

    • addAndGetCurrentSize

      public long addAndGetCurrentSize(long inc)
      Add to the current size and return the new value.
      Parameters:
      inc - the increment
      Returns:
      the new size

    • setAndGetCurrentSize

      public long setAndGetCurrentSize(long value)
      Set the current size and return the value.
      Parameters:
      value - the value
      Returns:
      the value

    • getKeepAliveTime

      public long getKeepAliveTime()
      Description copied from interface: MessageDispatchInterceptorMBean
      Get the keep alive time.
      Specified by:
      getKeepAliveTime in interface MessageDispatchInterceptorMBean
      Returns:
      the keep alive time

    • getMaxSpareThreads

      public int getMaxSpareThreads()
      Description copied from interface: MessageDispatchInterceptorMBean
      Get the maximum spare threads.
      Specified by:
      getMaxSpareThreads in interface MessageDispatchInterceptorMBean
      Returns:
      the maximum spare threads

    • getMaxThreads

      public int getMaxThreads()
      Description copied from interface: MessageDispatchInterceptorMBean
      Get the maximum threads.
      Specified by:
      getMaxThreads in interface MessageDispatchInterceptorMBean
      Returns:
      the maximum threads

    • setKeepAliveTime

      public void setKeepAliveTime(long keepAliveTime)
      Set the keep alive time.
      Parameters:
      keepAliveTime - the keep alive time

    • setMaxSpareThreads

      public void setMaxSpareThreads(int maxSpareThreads)
      Set the maximum spare threads.
      Parameters:
      maxSpareThreads - the maximum spare threads

    • setMaxThreads

      public void setMaxThreads(int maxThreads)
      Set the maximum threads.
      Parameters:
      maxThreads - the maximum threads

    • isAlwaysSend

      public boolean isAlwaysSend()
      Description copied from interface: MessageDispatchInterceptorMBean
      Check if always send is enabled.
      Specified by:
      isAlwaysSend in interface MessageDispatchInterceptorMBean
      Returns:
      whether always send is enabled

    • setAlwaysSend

      public void setAlwaysSend(boolean alwaysSend)
      Description copied from interface: MessageDispatchInterceptorMBean
      Set always send.
      Specified by:
      setAlwaysSend in interface MessageDispatchInterceptorMBean
      Parameters:
      alwaysSend - whether to always send

    • start

      public void start(int svc) throws ChannelException
      Description copied from interface: ChannelInterceptor
      Starts up the channel. This can be called multiple times for individual services to start The svc parameter can be the logical or value of any constants
      Specified by:
      start in interface ChannelInterceptor
      Overrides:
      start in class ChannelInterceptorBase
      Parameters:
      svc - one of:
      • Channel.DEFAULT - will start all services
      • Channel.MBR_RX_SEQ - starts the membership receiver
      • Channel.MBR_TX_SEQ - starts the membership broadcaster
      • Channel.SND_TX_SEQ - starts the replication transmitter
      • Channel.SND_RX_SEQ - starts the replication receiver
      Throws:
      ChannelException - if a startup error occurs or the service is already started.
      See Also:

    • stop

      public void stop(int svc) throws ChannelException
      Description copied from interface: ChannelInterceptor
      Shuts down the channel. This can be called multiple times for individual services to shut down. The svc parameter can be the logical or value of any constants
      Specified by:
      stop in interface ChannelInterceptor
      Overrides:
      stop in class ChannelInterceptorBase
      Parameters:
      svc - one of:
      • Channel.DEFAULT - will shut down all services
      • Channel.MBR_RX_SEQ - stops the membership receiver
      • Channel.MBR_TX_SEQ - stops the membership broadcaster
      • Channel.SND_TX_SEQ - stops the replication transmitter
      • Channel.SND_RX_SEQ - stops the replication receiver
      Throws:
      ChannelException - if a startup error occurs or the service is already started.
      See Also:

    • sendAsyncData

      protected void sendAsyncData(ChannelMessage msg, Member[] destination, InterceptorPayload payload)
      Send async data.
      Parameters:
      msg - the message
      destination - the destination
      payload - the payload

    • getPoolSize

      public int getPoolSize()
      Return the current number of threads that are managed by the pool.
      Specified by:
      getPoolSize in interface MessageDispatchInterceptorMBean
      Returns:
      the current number of threads that are managed by the pool

    • getActiveCount

      public int getActiveCount()
      Return the current number of threads that are in use.
      Specified by:
      getActiveCount in interface MessageDispatchInterceptorMBean
      Returns:
      the current number of threads that are in use

    • getTaskCount

      public long getTaskCount()
      Return the total number of tasks that have ever been scheduled for execution by the pool.
      Specified by:
      getTaskCount in interface MessageDispatchInterceptorMBean
      Returns:
      the total number of tasks that have ever been scheduled for execution by the pool

    • getCompletedTaskCount

      public long getCompletedTaskCount()
      Return the total number of tasks that have completed execution by the pool.
      Specified by:
      getCompletedTaskCount in interface MessageDispatchInterceptorMBean
      Returns:
      the total number of tasks that have completed execution by the pool