Package com.swiftmq.impl.streams.comp.io

Class QueueWireTapInput

  • All Implemented Interfaces:
    DestinationInput, Input, WireTapSubscriber
    public class QueueWireTapInput
extends java.lang.Object
implements DestinationInput, WireTapSubscriber
    Creates or re-uses a named WireTap on a Queue and adds a subscriber to it.

    Messages are sent to the WireTap when the Message is inserted into the Queue. If the WireTap has multiple subscribers (that is, Streams are using the same WireTap name), Messages are distributed evenly over the subscribers in a round-robin fashion.

    Per default, a deep copy of a Message is performed before it is transferred to a subscriber. So it is save to change it. If the Message will not be changed by a subscriber, the deep copy can be disabled. But be careful! All changes of a Message will have side effects.

    Message transfer to a WireTap subscriber is done outside of a transaction. The Subscriber uses a BlockingQueue to receive the Messages. The BlockingQueue has a bufferSize which is 10 Messages per default. If the size is reached, the Queue waits until there is free space. The time to wait is specified in maxBlockTime with a default of 500 ms. If the timeout is reached, the Message will not be inserted, so a WireTap subscriber may not receive all Messages transferred through the Queue. For that reason, a WireTap is a good solution if Message lost can be tolerated, e.g. in statistic scenarios.

    Author:
    IIT Software GmbH, Muenster/Germany, (c) 2017, All Rights Reserved

    • Method Detail

      • destinationName

        public DestinationInput destinationName​(java.lang.String destinationName)
        Description copied from interface: DestinationInput
        Sets the destinationName if different from the name used in stream.create().input(name)
        Specified by:
        destinationName in interface DestinationInput
        Parameters:
        destinationName - destinationName
        Returns:
        DestinationInput

      • bufferSize

        public QueueWireTapInput bufferSize​(int bufferSize)
        Sets the bufferSize of the internal BlockingQueue. Default is 10 Messages.
        Parameters:
        bufferSize - Buffer Size
        Returns:
        this

      • maxBlockTime

        public QueueWireTapInput maxBlockTime​(long maxBlockTime)
        Sets the maximum block time when the internal BlockingQueue is full. Default is 500 ms. If the time is reached, the Message will not be inserted.
        Parameters:
        maxBlockTime - Max Block Time
        Returns:
        this

      • requieresDeepCopy

        public boolean requieresDeepCopy()
        Returns whether Message inserts requires a deep copy (default is true).
        Specified by:
        requieresDeepCopy in interface WireTapSubscriber
        Returns:
        deepCopy flag

      • requiresDeepCopy

        public QueueWireTapInput requiresDeepCopy​(boolean requiresDeepCopy)
        Sets whether the Message inserts requires a deep copy in order to change it later on. Default is true.
        Parameters:
        requiresDeepCopy - deepCopy flag
        Returns:
        this

      • next

        public Message next()
        Internal use.

      • getName

        public java.lang.String getName()
        Description copied from interface: Input
        Returns the name of this Input.
        Specified by:
        getName in interface Input
        Returns:
        Name

      • getSelector

        public java.lang.String getSelector()
        Internal use.
        Specified by:
        getSelector in interface DestinationInput
        Returns:
        JMS Message Selector

      • selector

        public DestinationInput selector​(java.lang.String selector)
        Description copied from interface: DestinationInput
        Sets the JMS Message Selector
        Specified by:
        selector in interface DestinationInput
        Parameters:
        selector - JMS Message Selector
        Returns:
        DestinationInput

      • current

        public Input current​(Message current)
        Description copied from interface: Input
        Sets the current Message on this Input.
        Specified by:
        current in interface Input
        Parameters:
        current - Message
        Returns:
        Input

      • current

        public Message current()
        Description copied from interface: Input
        Returns the current Message of this Input.
        Specified by:
        current in interface Input
        Returns:
        Message

      • executeCallback

        public void executeCallback()
                     throws java.lang.Exception
        Description copied from interface: Input
        Internal use.
        Specified by:
        executeCallback in interface Input
        Throws:
        java.lang.Exception

      • collect

        public void collect​(long interval)
        Description copied from interface: Input
        Internal use.
        Specified by:
        collect in interface Input

      • start

        public void start()
           throws java.lang.Exception
        Description copied from interface: Input
        Starts this Input. This method is called automatically if an Input is created outside a callback. If it is created inside, it must be called explicitly.
        Specified by:
        start in interface Input
        Throws:
        java.lang.Exception

      • close

        public void close()
        Description copied from interface: Input
        Closes this Input.
        Specified by:
        close in interface Input

      • toString

        public java.lang.String toString()
        Overrides:
        toString in class java.lang.Object