com.blacklocus.qs

Class MessageProviders

java.lang.Object

com.blacklocus.qs.MessageProviders

public class MessageProviders
extends java.lang.Object

Provide some common functionality to wrap around existing MessageProvider implementations.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static interface	MessageProviders.NextCallback

Constructor Summary

Constructors

Constructor and Description
MessageProviders()

Method Summary

Modifier and Type	Method and Description
static MessageProvider	filtered(MessageProvider messageProvider, com.google.common.base.Predicate<Message> filter) Filters the values returned by QueueItemProvider.next() by some filter.
static MessageProvider	newDefaultMessageProvider(MessageProvider messageProvider, long maxTimesNextEmpty) Chain together a safe and an empty message return tracking MessageProvider.
static MessageProvider	newDefaultMessageProvider(MessageProvider messageProvider, long maxTimesNextEmpty, int maxMessages) Chain together a safe and an empty message return tracking MessageProvider.
static MessageProvider	newSafeMessageProvider(MessageProvider messageProvider) Wrap each call of the given MessageProvider with a try/catch such that they always return correctly without throwing an Exception.
static MessageProvider	newUntilNextEmptyMessageProvider(MessageProvider messageProvider, long maxTimesNextEmpty) Wrap an existing MessageProvider such that after a specified number of calls to next() that results in 0 messages being returned, hasNext() flips over to return false.
static MessageProvider	newUntilNextEmptyOrMaximumMessageProvider(MessageProvider messageProvider, long maxTimesNextEmpty, int maxMessages) Wrap an existing MessageProvider such that after a specified number of calls to next() that results in 0 messages being returned, or the maximum limit of messages is reached, hasNext() flips over to return false.
static MessageProvider	newUntilSlowOrMaximumMessageProvider(MessageProvider messageProvider, long maxTimesSlow, int slowThreshold, int maxMessages) Like newUntilNextEmptyOrMaximumMessageProvider(MessageProvider, long, int) but instead of detecting numerous "empties" considers numerous "slows", where that is defined to be calls to next that produce less than the specified batch threshold.
static MessageProvider	scanningCallback(MessageProvider messageProvider, MessageProviders.NextCallback nextCallback) Applies an arbitrary callback to the results of calls to QueueItemProvider.next() before bubbling onward.

Methods inherited from class java.lang.Object

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

Constructor Detail

MessageProviders

public MessageProviders()

Method Detail

newDefaultMessageProvider

public static MessageProvider newDefaultMessageProvider(MessageProvider messageProvider,
                                                        long maxTimesNextEmpty)

Chain together a safe and an empty message return tracking MessageProvider.

Parameters:

messageProvider - the base MessageProvider to wrap

maxTimesNextEmpty - the number of times next() can return with 0 messages until hasNext() begins to return false

Returns:

the wrapped MessageProvider

newDefaultMessageProvider

public static MessageProvider newDefaultMessageProvider(MessageProvider messageProvider,
                                                        long maxTimesNextEmpty,
                                                        int maxMessages)

Chain together a safe and an empty message return tracking MessageProvider.

Parameters:

messageProvider - the base MessageProvider to wrap

maxTimesNextEmpty - the number of times next() can return with 0 messages until hasNext() begins to return false

maxMessages - An maximum limit of messages to provide until hasNext() begins to return false

Returns:

the wrapped MessageProvider

newSafeMessageProvider

public static MessageProvider newSafeMessageProvider(MessageProvider messageProvider)

Wrap each call of the given MessageProvider with a try/catch such that they always return correctly without throwing an Exception. Caught Exception's are swallowed and logged.

Parameters:

messageProvider - the base MessageProvider to wrap

Returns:

the safely wrapped MessageProvider

newUntilNextEmptyMessageProvider

public static MessageProvider newUntilNextEmptyMessageProvider(MessageProvider messageProvider,
                                                               long maxTimesNextEmpty)

Wrap an existing MessageProvider such that after a specified number of calls to next() that results in 0 messages being returned, hasNext() flips over to return false. If the underlying MessageProvider's hasNext() returns false before the threshold is exceeded, the hasNext() will also return false.

Parameters:

messageProvider - the base MessageProvider to wrap

maxTimesNextEmpty - the number of times next() can return with 0 messages until hasNext() begins to return false

Returns:

the empty message handling wrapped MessageProvider

newUntilNextEmptyOrMaximumMessageProvider

public static MessageProvider newUntilNextEmptyOrMaximumMessageProvider(MessageProvider messageProvider,
                                                                        long maxTimesNextEmpty,
                                                                        int maxMessages)

Wrap an existing MessageProvider such that after a specified number of calls to next() that results in 0 messages being returned, or the maximum limit of messages is reached, hasNext() flips over to return false. If the underlying MessageProvider's hasNext() returns false before the threshold is exceeded, the hasNext() will also return false.

Since the existing MessageProvider's next() method might return more than one message, the maxMessages parameter can be exceeded for that particular next() call. However, the hasNext() method will return false on the following call.

Parameters:

messageProvider - the base MessageProvider to wrap

maxTimesNextEmpty - the number of times next() can return with 0 messages until hasNext() begins to return false

maxMessages - An maximum limit of messages to provide until hasNext() begins to return false

Returns:

the empty message handling wrapped MessageProvider

newUntilSlowOrMaximumMessageProvider

public static MessageProvider newUntilSlowOrMaximumMessageProvider(MessageProvider messageProvider,
                                                                   long maxTimesSlow,
                                                                   int slowThreshold,
                                                                   int maxMessages)

Like newUntilNextEmptyOrMaximumMessageProvider(MessageProvider, long, int) but instead of detecting numerous "empties" considers numerous "slows", where that is defined to be calls to next that produce less than the specified batch threshold. This is useful in the case that a queue is being written to while being dumped. This can cause endless, slow queue reading as a messages continuously trickle in before "timesEmpty" is achieved. Logic is identical except that instead of requiring empty, "timesSlow" is counted.

newUntilNextEmptyOrMaximumMessageProvider(MessageProvider, long, int) is actually a specialized form of this wrapper, where 'slowThreshold = 0'

Parameters:

slowThreshold - calls to next that contribute less than or equal to this threshold will contribute one increment to 'timesSlow' where when 'timesSlow == maxTimesSlow', the MessageProvider will terminate. For AmazonSQS, 10 is the usual returned message clump. So anything less than 10 may be a valid consideration for slow, e.g. 9, but in any case depends on the underlying MessageProvider.

filtered

public static MessageProvider filtered(MessageProvider messageProvider,
                                       com.google.common.base.Predicate<Message> filter)

Filters the values returned by QueueItemProvider.next() by some filter. This does not reduce the scan count of the given message provider, but does reduce what bubbles out of it throught `next`. Note that this may cause some calls to `next` to contain 0 elements even though there are more messages to scan.

scanningCallback

public static MessageProvider scanningCallback(MessageProvider messageProvider,
                                               MessageProviders.NextCallback nextCallback)

Applies an arbitrary callback to the results of calls to QueueItemProvider.next() before bubbling onward.