Package com.tangosol.util.processor

Class StreamingAsynchronousProcessor<K,V,R>

java.lang.Object

com.tangosol.util.AsynchronousAgent<Void>

com.tangosol.util.processor.AbstractAsynchronousProcessor<K,V,R,Void>

com.tangosol.util.processor.StreamingAsynchronousProcessor<K,V,R>

Type Parameters:

K - the type of the Map entry key

V - the type of the Map entry value

R - the type of value returned by the EntryProcessor

All Implemented Interfaces:

FlowControl, InvocableMap.EntryProcessor<K,V,R>, Serializable, Future<Void>

public class StreamingAsynchronousProcessor<K,V,R>
extends AbstractAsynchronousProcessor<K,V,R,Void>

An EntryProcessor wrapper class that allows for an asynchronous invocation of the underlying processor. Unlike AsynchronousProcessor, this implementation does not collect the results of the underlying entry processor execution, but simply streams partial results to the provided partial results callback.

This allows for a much lower memory overhead if the complete result set does not to be realized on the client.

It's very important that the overriding implementations of onComplete(), onResult(java.util.Map.Entry<K, R>) and onException(java.lang.Throwable), and provided callbacks must be non-blocking. For example, any use of NamedCache API is completely disallowed, with the only exception of asynchronous agents with disabled flow control.

The underlying entry processor is guaranteed to have been fully executed when onComplete() is called.

Note 1: Neither this class nor its extensions need to be serializable. Only the underlying processor is serialized and sent to corresponding servers for execution.
Note 2: This feature is not available on Coherence*Extend clients.

Author:

as 2015.01.28

See Also:

• AsynchronousProcessor
• SingleEntryAsynchronousProcessor
• Serialized Form

Nested Class Summary

Nested classes/interfaces inherited from interface java.util.concurrent.Future

Future.State

Field Summary

Fields

Modifier and Type	Field	Description
protected final Consumer<? super Map.Entry<? extends K,? extends R>>	f_onPartial	The user-provided callback that will be invoked for each partial result.

Fields inherited from class com.tangosol.util.processor.AbstractAsynchronousProcessor

f_processor

Fields inherited from class com.tangosol.util.AsynchronousAgent

m_control, m_iOrderId

Constructor Summary

Constructors

Constructor	Description
StreamingAsynchronousProcessor(InvocableMap.EntryProcessor<K,V,R> processor, int iUnitOrderId, Consumer<? super Map.Entry<? extends K,? extends R>> onPartial)	Construct a StreamingAsynchronousProcessor for a given processor and one or more callbacks.
StreamingAsynchronousProcessor(InvocableMap.EntryProcessor<K,V,R> processor, int iUnitOrderId, Consumer<? super Map.Entry<? extends K,? extends R>> onPartial, Executor executor)	Construct a StreamingAsynchronousProcessor for a given processor and one or more callbacks.
StreamingAsynchronousProcessor(InvocableMap.EntryProcessor<K,V,R> processor, Consumer<? super Map.Entry<? extends K,? extends R>> onPartial)	Construct a StreamingAsynchronousProcessor for a given processor and one or more callbacks.
StreamingAsynchronousProcessor(InvocableMap.EntryProcessor<K,V,R> processor, Consumer<? super Map.Entry<? extends K,? extends R>> onPartial, Executor executor)	Construct a StreamingAsynchronousProcessor for a given processor and one or more callbacks.

Method Summary

Modifier and Type	Method	Description
void	onComplete()	Called after the processor has been notified about all possible partial results or failures and no more are forthcoming.
void	onException(Throwable eReason)	Called if any part of the operation failed for any reason.
void	onResult(Map.Entry<K,R> entry)	Called when there is a partial result of the asynchronous execution.

Methods inherited from class com.tangosol.util.processor.AbstractAsynchronousProcessor

getProcessor, getUnitOfOrderId, process, processAll

Methods inherited from class com.tangosol.util.AsynchronousAgent

bind, cancel, checkBacklog, complete, completeExceptionally, drainBacklog, flush, get, get, getCompletableFuture, getException, getResult, isCancelled, isCompletedExceptionally, isDone

Methods inherited from class java.lang.Object

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

Methods inherited from interface java.util.concurrent.Future

exceptionNow, resultNow, state

Field Details

f_onPartial

protected final Consumer<? super Map.Entry<? extends K,? extends R>> f_onPartial

The user-provided callback that will be invoked for each partial result.

Constructor Details

StreamingAsynchronousProcessor

public StreamingAsynchronousProcessor(InvocableMap.EntryProcessor<K,V,R> processor,
 Consumer<? super Map.Entry<? extends K,? extends R>> onPartial)

Construct a StreamingAsynchronousProcessor for a given processor and one or more callbacks.

Important Note: All provided callbacks must be non-blocking. For example, any use of NamedCache API is completely disallowed.

Parameters:

processor - the underlying InvocableMap.EntryProcessor

onPartial - a user-defined callback that will be called for each partial result

StreamingAsynchronousProcessor

public StreamingAsynchronousProcessor(InvocableMap.EntryProcessor<K,V,R> processor,
 Consumer<? super Map.Entry<? extends K,? extends R>> onPartial,
 Executor executor)

Construct a StreamingAsynchronousProcessor for a given processor and one or more callbacks.

Important Note: All provided callbacks must be non-blocking. For example, any use of NamedCache API is completely disallowed.

Parameters:

processor - the underlying InvocableMap.EntryProcessor

onPartial - a user-defined callback that will be called for each partial result

executor - an optional Executor to complete the future on, if not provided the Daemons.commonPool() is used

StreamingAsynchronousProcessor

public StreamingAsynchronousProcessor(InvocableMap.EntryProcessor<K,V,R> processor,
 int iUnitOrderId,
 Consumer<? super Map.Entry<? extends K,? extends R>> onPartial)

Construct a StreamingAsynchronousProcessor for a given processor and one or more callbacks.

Important Note: All provided callbacks must be non-blocking. For example, any use of NamedCache API is completely disallowed.

Parameters:

processor - the underlying InvocableMap.EntryProcessor

iUnitOrderId - the unit-of-order id for this processor

onPartial - a user-defined callback that will be called for each partial result

StreamingAsynchronousProcessor

public StreamingAsynchronousProcessor(InvocableMap.EntryProcessor<K,V,R> processor,
 int iUnitOrderId,
 Consumer<? super Map.Entry<? extends K,? extends R>> onPartial,
 Executor executor)

Construct a StreamingAsynchronousProcessor for a given processor and one or more callbacks.

Important Note: All provided callbacks must be non-blocking. For example, any use of NamedCache API is completely disallowed.

Parameters:

processor - the underlying InvocableMap.EntryProcessor

iUnitOrderId - the unit-of-order id for this processor

onPartial - a user-defined callback that will be called for each partial result

executor - an optional Executor to complete the future on, if not provided the Daemons.commonPool() is used

Method Details

onResult

public void onResult(Map.Entry<K,R> entry)

Description copied from class: AbstractAsynchronousProcessor

Called when there is a partial result of the asynchronous execution.

For a given request, calls to this method and AbstractAsynchronousProcessor.onException(java.lang.Throwable) may come concurrently.

For ordering guarantees across different processor invocations see AbstractAsynchronousProcessor.getUnitOfOrderId().

Note: Overriding implementations of this method must be non-blocking.

Specified by:

onResult in class AbstractAsynchronousProcessor<K,V,R,Void>

Parameters:

entry - an entry holding the key and a result of the operation for the given key

onException

public void onException(Throwable eReason)

Description copied from class: AbstractAsynchronousProcessor

Called if any part of the operation failed for any reason. For operations that span multiple partitions this method could be called more than once. However, unless subclasses override this method, any failure will "complete" the operation.

Note: Overriding implementations of this method must be non-blocking.

Specified by:

onException in class AbstractAsynchronousProcessor<K,V,R,Void>

Parameters:

eReason - the reason of failure

onComplete

public void onComplete()

Description copied from class: AbstractAsynchronousProcessor

Called after the processor has been notified about all possible partial results or failures and no more are forthcoming. As long as this processor was submitted to any of InvocableMap's methods, this method is guaranteed to be called once and only once.

Possible call back sequences are:

  cache.invoke
    ...
    onResult
    onComplete

  cache.invoke
    ...
    onException
    onComplete

  cache.invokeAll
    ...
     onResult
     onException
     onException
     onResult
     onComplete
 

For ordering guarantees across processors see AbstractAsynchronousProcessor.getUnitOfOrderId().

Note: Overriding implementations of this method must be non-blocking.

Specified by:

onComplete in class AbstractAsynchronousProcessor<K,V,R,Void>