public abstract class AbstractProcessor extends Object implements Processor
Processor with several levels of convenience:
Processor.init(Outbox, Context) retains the supplied outbox
and the logger retrieved from the context.
process(n, inbox) delegates to the matching
tryProcessN() with each item received in the inbox.
tryProcess(int, Object) to which
the tryProcessN methods delegate by default. It is convenient
to override it when the processor doesn't care which edge an item
originates from. Another convenient idiom is to override tryProcessN() for one or two specially treated edges and override
tryProcess(int, Object) to process the rest of the edges, which
are treated uniformly.
tryEmit(...) methods avoid the need to deal with Outbox
directly.
emitFromTraverser(...) methods handle the boilerplate of
emission from a traverser. They are especially useful in the
Processor.complete() step when there is a collection of items to emit.
The Traversers class contains traversers tailored to simplify
the implementation of complete().
FlatMapper class additionally simplifies the
usage of emitFromTraverser() inside tryProcess(), in
a scenario where an input item results in a collection of output
items. FlatMapper is obtained from one of the factory methods
flatMapper(...).
| Modifier and Type | Class and Description |
|---|---|
protected class |
AbstractProcessor.FlatMapper<T,R>
A helper that simplifies the implementation of
tryProcess(ordinal, item) for emitting collections. |
Processor.Context| Constructor and Description |
|---|
AbstractProcessor() |
| Modifier and Type | Method and Description |
|---|---|
protected <E> boolean |
emitFromTraverser(int[] ordinals,
Traverser<E> traverser)
Obtains items from the traverser and offers them to the outbox's buckets
identified in the supplied array.
|
protected <E> boolean |
emitFromTraverser(int ordinal,
Traverser<E> traverser)
Obtains items from the traverser and offers them to the outbox's buckets
identified in the supplied array.
|
protected boolean |
emitFromTraverser(Traverser<?> traverser)
Convenience for
emitFromTraverser(int, Traverser)
which emits to all ordinals. |
protected <T extends Map.Entry<?,?>> |
emitFromTraverserToSnapshot(Traverser<T> traverser)
Obtains items from the traverser and offers them to the snapshot bucket
of the outbox.
|
protected <T,R> AbstractProcessor.FlatMapper<T,R> |
flatMapper(Function<? super T,? extends Traverser<? extends R>> mapper)
Factory of
AbstractProcessor.FlatMapper. |
protected <T,R> AbstractProcessor.FlatMapper<T,R> |
flatMapper(int[] ordinals,
Function<? super T,? extends Traverser<? extends R>> mapper)
Factory of
AbstractProcessor.FlatMapper. |
protected <T,R> AbstractProcessor.FlatMapper<T,R> |
flatMapper(int ordinal,
Function<? super T,? extends Traverser<? extends R>> mapper)
Factory of
AbstractProcessor.FlatMapper. |
protected ILogger |
getLogger()
Returns the logger associated with this processor instance.
|
protected Outbox |
getOutbox() |
void |
init(Outbox outbox,
Processor.Context context)
Initializes this processor with the outbox that the processing
methods must use to deposit their output items.
|
protected void |
init(Processor.Context context)
Method that can be overridden to perform any necessary initialization
for the processor.
|
protected void |
keyedWatermarkCheck(Watermark watermark)
Throws
UnsupportedOperationException if watermark has non-zero
key. |
void |
process(int ordinal,
Inbox inbox)
Implements the boilerplate of dispatching against the ordinal,
taking items from the inbox one by one, and invoking the
processing logic on each.
|
void |
restoreFromSnapshot(Inbox inbox)
Implements the boilerplate of polling the inbox, casting the items to
Map.Entry, and extracting the key and value. |
protected void |
restoreFromSnapshot(Object key,
Object value)
Called to restore one key-value pair from the snapshot to processor's
internal state.
|
protected boolean |
tryEmit(int[] ordinals,
Object item)
Offers the item to the outbox buckets identified in the supplied array.
|
protected boolean |
tryEmit(int ordinal,
Object item)
Offers the item to the outbox bucket at the supplied ordinal.
|
protected boolean |
tryEmit(Object item)
Offers the item to all the outbox buckets (except the snapshot outbox).
|
protected boolean |
tryEmitToSnapshot(Object key,
Object value)
Offers one key-value pair to the snapshot bucket.
|
protected boolean |
tryProcess(int ordinal,
Object item)
Tries to process the supplied input item, which was received from the
edge with the supplied ordinal.
|
protected boolean |
tryProcess0(Object item)
Tries to process the supplied input item, which was received from the
edge with ordinal 0.
|
protected boolean |
tryProcess1(Object item)
Tries to process the supplied input item, which was received from the
edge with ordinal 1.
|
protected boolean |
tryProcess2(Object item)
Tries to process the supplied input item, which was received from the
edge with ordinal 2.
|
protected boolean |
tryProcess3(Object item)
Tries to process the supplied input item, which was received from the
edge with ordinal 3.
|
protected boolean |
tryProcess4(Object item)
Tries to process the supplied input item, which was received from the
edge with ordinal 4.
|
boolean |
tryProcessWatermark(Watermark watermark)
This basic implementation only forwards the passed watermark.
|
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, waitclose, closeIsCooperative, complete, completeEdge, finishSnapshotRestore, isCooperative, saveToSnapshot, snapshotCommitFinish, snapshotCommitPrepare, tryProcess, tryProcessWatermarkpublic final void init(@Nonnull Outbox outbox, @Nonnull Processor.Context context) throws Exception
ProcessorProcessor.isCooperative() method).
Even if this processor is cooperative, this method is allowed to do blocking operations.
The default implementation does nothing.
public final void restoreFromSnapshot(@Nonnull Inbox inbox)
Map.Entry, and extracting the key and value. Forwards each
key-value pair to restoreFromSnapshot(Object, Object).restoreFromSnapshot in interface Processorpublic void process(int ordinal,
@Nonnull
Inbox inbox)
protected void init(@Nonnull Processor.Context context) throws Exception
process() and
Processor.complete()), but after the outbox and logger have been initialized.
Subclasses are not required to call this superclass method, it does nothing.
protected boolean tryProcess(int ordinal,
@Nonnull
Object item)
throws Exception
false, in which case it will be called again later
with the same (ordinal, item) combination before any other
processing method is called.
The default implementation throws an UnsupportedOperationException.
NOTE: unless the processor doesn't differentiate between
its inbound edges, the first choice should be leaving this method alone
and instead overriding the specific tryProcessN() methods for
each ordinal the processor expects.
ordinal - ordinal of the edge that delivered the itemitem - item to be processedtrue if this item has now been processed,
false otherwise.Exceptionprotected boolean tryProcess0(@Nonnull Object item) throws Exception
false, in which case it will be called again later with the same
item before any other processing method is called.
The default implementation delegates to tryProcess(0, item).
item - item to be processedtrue if this item has now been processed,
false otherwise.Exceptionprotected boolean tryProcess1(@Nonnull Object item) throws Exception
false, in which case it will be called again later with the same
item before any other processing method is called.
The default implementation delegates to tryProcess(1, item).
item - item to be processedtrue if this item has now been processed,
false otherwise.Exceptionprotected boolean tryProcess2(@Nonnull Object item) throws Exception
false, in which case it will be called again later with the same
item before any other processing method is called.
The default implementation delegates to tryProcess(2, item).
item - item to be processedtrue if this item has now been processed,
false otherwise.Exceptionprotected boolean tryProcess3(@Nonnull Object item) throws Exception
false, in which case it will be called again later with the same
item before any other processing method is called.
The default implementation delegates to tryProcess(3, item).
item - item to be processedtrue if this item has now been processed,
false otherwise.Exceptionprotected boolean tryProcess4(@Nonnull Object item) throws Exception
false, in which case it will be called again later with the same
item before any other processing method is called.
The default implementation delegates to tryProcess(4, item).
item - item to be processedtrue if this item has now been processed,
false otherwise.Exceptionprotected void restoreFromSnapshot(@Nonnull Object key, @Nonnull Object value)
The default implementation throws an UnsupportedOperationException, but it will not be called unless you
override Processor.saveToSnapshot().
key - key of the entry from the snapshotvalue - value of the entry from the snapshotpublic boolean tryProcessWatermark(@Nonnull Watermark watermark)
tryProcessWatermark in interface Processorwatermark - watermark to be processedtrue if this watermark has now been processed,
false to call this method again with the same watermarkprotected final ILogger getLogger()
protected final Outbox getOutbox()
@CheckReturnValue protected final boolean tryEmit(int ordinal, @Nonnull Object item)
Emitted items should not be subsequently mutated because the same instance might be used by a downstream processor in a different thread, causing concurrent access.
true, if the item was accepted. If false is
returned, the call must be retried later with the same (or equal) item.@CheckReturnValue protected final boolean tryEmit(@Nonnull Object item)
Emitted items should not be subsequently mutated because the same instance might be used by a downstream processor in a different thread, causing concurrent access.
true, if the item was accepted. If false is
returned, the call must be retried later with the same (or equal) item.@CheckReturnValue protected final boolean tryEmit(@Nonnull int[] ordinals, @Nonnull Object item)
Emitted items should not be subsequently mutated because the same instance might be used by a downstream processor in a different thread, causing concurrent access.
true, if the item was accepted. If false is
returned, the call must be retried later with the same (or equal) item.protected final <E> boolean emitFromTraverser(@Nonnull int[] ordinals, @Nonnull Traverser<E> traverser)
false.
Emitted items should not be subsequently mutated because the same instance might be used by a downstream processor in a different thread, causing concurrent access.
If this method returns false, then the caller must retain the
traverser and pass it again in the subsequent invocation of this method,
so as to resume emitting where it left off.
For simplified usage from tryProcess(ordinal, item) methods, see AbstractProcessor.FlatMapper.
ordinals - ordinals of the target buckettraverser - traverser over items to emitprotected final <E> boolean emitFromTraverser(int ordinal,
@Nonnull
Traverser<E> traverser)
false.
Do not mutate the items you emit because the downstream processor may be using them in a different thread, resulting in concurrent access.
If this method returns false, then you must retain the traverser
and pass it again in the subsequent invocation of this method, so as to
resume emitting where you left off.
For simplified usage in tryProcess(ordinal, item) methods, see AbstractProcessor.FlatMapper.
ordinal - ordinal of the target buckettraverser - traverser over items to emitprotected final boolean emitFromTraverser(@Nonnull Traverser<?> traverser)
emitFromTraverser(int, Traverser)
which emits to all ordinals.@CheckReturnValue protected final boolean tryEmitToSnapshot(@Nonnull Object key, @Nonnull Object value)
The type of the offered key determines which processors receive the key
and value pair when it is restored. If the key is of type BroadcastKey, the entry will be restored to all processor instances.
Otherwise, the key will be distributed according to default partitioning
and only a single processor instance will receive the key.
Keys and values offered to snapshot are serialized and can be further mutated as soon as this method returns.
true, if the item was accepted. If false is
returned, the call must be retried later with the same (or equal) key
and value.protected final <T extends Map.Entry<?,?>> boolean emitFromTraverserToSnapshot(@Nonnull Traverser<T> traverser)
Map.Entry and its key and value
are passed as the two arguments of tryEmitToSnapshot(Object, Object).
If the outbox refuses an item, it backs off and returns false.
Keys and values offered to snapshot are serialized and can be further mutated as soon as this method returns.
If this method returns false, then the caller must retain the
traverser and pass it again in the subsequent invocation of this method,
so as to resume emitting where it left off.
The type of the offered key determines which processors receive the key
and value pair when it is restored. If the key is of type BroadcastKey, the entry will be restored to all processor instances.
Otherwise, the key will be distributed according to default partitioning
and only a single processor instance will receive the key.
traverser - traverser over the items to emit to the snapshot@Nonnull protected final <T,R> AbstractProcessor.FlatMapper<T,R> flatMapper(int ordinal, @Nonnull Function<? super T,? extends Traverser<? extends R>> mapper)
AbstractProcessor.FlatMapper. The FlatMapper will emit items to
the given output ordinal.@Nonnull protected final <T,R> AbstractProcessor.FlatMapper<T,R> flatMapper(@Nonnull Function<? super T,? extends Traverser<? extends R>> mapper)
AbstractProcessor.FlatMapper. The FlatMapper will emit items to
all defined output ordinals.@Nonnull protected final <T,R> AbstractProcessor.FlatMapper<T,R> flatMapper(@Nonnull int[] ordinals, @Nonnull Function<? super T,? extends Traverser<? extends R>> mapper)
AbstractProcessor.FlatMapper. The FlatMapper will emit items to
the ordinals identified in the array.protected void keyedWatermarkCheck(Watermark watermark)
UnsupportedOperationException if watermark has non-zero
key.
Supposed to be used by processors that don't function properly with keyed watermarks.
Copyright © 2022 Hazelcast, Inc.. All rights reserved.