com.hazelcast.jet.core.test

Class TestOutbox

java.lang.Object

com.hazelcast.jet.core.test.TestOutbox

All Implemented Interfaces:

Outbox

public final class TestOutbox
extends Object
implements Outbox

Outbox implementation suitable to be used in tests.

Constructor Summary

Constructors

Constructor and Description
TestOutbox(int... capacities)
TestOutbox(int[] edgeCapacities, int snapshotCapacity)

Method Summary

Modifier and Type	Method and Description
int	bucketCount() Returns the number of buckets in this outbox.
<T> void	drainQueueAndReset(int queueOrdinal, Collection<T> target, boolean logItems) Move all items from the queue to the target collection and make the outbox available to accept more items.
<T> void	drainQueuesAndReset(List<? extends Collection<T>> target, boolean logItems) Move all items from all queues (except the snapshot queue) to the target list of collections.
<K,V> void	drainSnapshotQueueAndReset(Collection<? super Map.Entry<K,V>> target, boolean logItems) Deserialize and move all items from the snapshot queue to the target collection and make the outbox available to accept more items.
boolean	offer(int[] ordinals, Object item) Offers the item to all supplied edge ordinals.
boolean	offer(int ordinal, Object item) Offers the supplied item to the bucket with the supplied ordinal.
boolean	offer(Object item) Offers the item to all edges.
boolean	offerToSnapshot(Object key, Object value) Offers the given key and value pair to the processor's snapshot storage.
<T> Queue<T>	queue(int ordinal) Exposes individual output queues to the testing code.
void	reset() Call this method after any of the offer() methods returned false to be able to offer again.
Queue<Map.Entry<Data,Data>>	snapshotQueue() Returns the queue to which snapshot is written.
String	toString()

Methods inherited from class java.lang.Object

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

Constructor Detail

TestOutbox

public TestOutbox(int... capacities)

Parameters:

capacities - Capacities of individual buckets. Number of buckets is determined by the number of provided capacities. There is no snapshot bucket.

TestOutbox

public TestOutbox(int[] edgeCapacities,
                  int snapshotCapacity)

Parameters:

edgeCapacities - Capacities of individual buckets. Number of buckets is determined by the number of provided capacities.

snapshotCapacity - Capacity of snapshot bucket. If 0, snapshot queue is not present.

Method Detail

bucketCount

public int bucketCount()

Description copied from interface: Outbox

Returns the number of buckets in this outbox. This is equal to the number of output edges of the vertex and does not include the snapshot bucket.

Specified by:

bucketCount in interface Outbox

offer

public boolean offer(int ordinal,
                     @Nonnull
                     Object item)

Description copied from interface: Outbox

Offers the supplied item to the bucket with the supplied ordinal.

Items offered to outbox should not be subsequently mutated because the same instance might be used by a downstream processor in different thread, causing concurrent access.

Specified by:

offer in interface Outbox

Parameters:

ordinal - output ordinal number or -1 to offer to all ordinals

Returns:

true if the outbox accepted the item

offer

public boolean offer(@Nonnull
                     Object item)

Description copied from interface: Outbox

Offers the item to all edges. See Outbox.offer(int, Object) for more details.

Specified by:

offer in interface Outbox

Returns:

true if the outbox accepted the item

offer

public boolean offer(@Nonnull
                     int[] ordinals,
                     @Nonnull
                     Object item)

Description copied from interface: Outbox

Offers the item to all supplied edge ordinals. See Outbox.offer(int, Object) for more details.

Specified by:

offer in interface Outbox

Returns:

true if the outbox accepted the item

offerToSnapshot

public boolean offerToSnapshot(@Nonnull
                               Object key,
                               @Nonnull
                               Object value)

Description copied from interface: Outbox

Offers the given key and value pair to the processor's snapshot storage.

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.

This method may only be called from the Processor.saveToSnapshot() method.

Keys and values offered to snapshot are serialized and can be further mutated as soon as this method returns.

Specified by:

offerToSnapshot in interface Outbox

Returns:

true if the outbox accepted the item

queue

public <T> Queue<T> queue(int ordinal)

Exposes individual output queues to the testing code.

Parameters:

ordinal - ordinal of the bucket

snapshotQueue

public Queue<Map.Entry<Data,Data>> snapshotQueue()

Returns the queue to which snapshot is written. It contains serialized data, if you need deserialized data, you might prefer to use drainSnapshotQueueAndReset(java.util.Collection<? super java.util.Map.Entry<K, V>>, boolean).

drainQueueAndReset

public <T> void drainQueueAndReset(int queueOrdinal,
                                   Collection<T> target,
                                   boolean logItems)

Move all items from the queue to the target collection and make the outbox available to accept more items. Also calls reset(). If you have a limited capacity outbox, you need to call this method regularly.

Parameters:

queueOrdinal - the queue from Outbox to drain

target - target collection

logItems - whether to log drained items to System.out

drainQueuesAndReset

public <T> void drainQueuesAndReset(List<? extends Collection<T>> target,
                                    boolean logItems)

Move all items from all queues (except the snapshot queue) to the target list of collections. Queue N is moved to collection at target N etc. Also calls reset(). If you have a limited capacity outbox, you need to call this method regularly.

Parameters:

target - list of target collections

logItems - whether to log drained items to System.out

drainSnapshotQueueAndReset

public <K,V> void drainSnapshotQueueAndReset(Collection<? super Map.Entry<K,V>> target,
                                             boolean logItems)

Deserialize and move all items from the snapshot queue to the target collection and make the outbox available to accept more items. Also calls reset(). If you have a limited capacity outbox, you need to call this method regularly.

Parameters:

target - target list

logItems - whether to log drained items to System.out

reset

public void reset()

Call this method after any of the offer() methods returned false to be able to offer again. Method is called automatically from drainQueueAndReset(int, java.util.Collection<T>, boolean) and drainSnapshotQueueAndReset(java.util.Collection<? super java.util.Map.Entry<K, V>>, boolean) methods.

toString

public String toString()

Overrides:

toString in class Object