Package com.hazelcast.jet.core.processor

Class SourceProcessors

java.lang.Object

com.hazelcast.jet.core.processor.SourceProcessors

public final class SourceProcessors
extends Object

Static utility class with factories of source processors (the DAG entry points). For other kinds for a vertices refer to the package-level documentation.

Since:

Jet 3.0

Method Summary

Modifier and Type	Method	Description
static <C, T, S> ProcessorMetaSupplier	convenientSourceP(FunctionEx<? super Processor.Context,? extends C> createFn, BiConsumerEx<? super C,? super SourceBuilder.SourceBuffer<T>> fillBufferFn, FunctionEx<? super C,? extends S> createSnapshotFn, BiConsumerEx<? super C,? super List<S>> restoreSnapshotFn, ConsumerEx<? super C> destroyFn, int preferredLocalParallelism, boolean isBatch, Permission permission)	Returns a supplier of processors for a source that the user can create using the SourceBuilder.
static <C, T, S> ProcessorMetaSupplier	convenientTimestampedSourceP(FunctionEx<? super Processor.Context,? extends C> createFn, BiConsumerEx<? super C,? super SourceBuilder.TimestampedSourceBuffer<T>> fillBufferFn, EventTimePolicy<? super T> eventTimePolicy, FunctionEx<? super C,? extends S> createSnapshotFn, BiConsumerEx<? super C,? super List<S>> restoreSnapshotFn, ConsumerEx<? super C> destroyFn, int preferredLocalParallelism)	Returns a supplier of processors for a source that the user can create using the SourceBuilder.
static ProcessorMetaSupplier	readCacheP(String cacheName)	Returns a supplier of processors for Sources.cache(String).
static <I> ProcessorMetaSupplier	readFilesP(String directory, String glob, boolean sharedFileSystem, boolean ignoreFileNotFound, FunctionEx<? super Path,? extends Stream<I>> readFileFn)	Returns a supplier of processors for FileSources.files(String) to read local files.
static <I> ProcessorMetaSupplier	readFilesP(String directory, String glob, boolean sharedFileSystem, FunctionEx<? super Path,? extends Stream<I>> readFileFn)	Returns a supplier of processors for Sources.filesBuilder(java.lang.String).
static <R> ProcessorMetaSupplier	readFilesP(String directory, Charset charset, String glob, boolean sharedFileSystem, BiFunctionEx<? super String,? super String,? extends R> mapOutputFn)	Returns a supplier of processors for Sources.filesBuilder(java.lang.String).
static <T> ProcessorMetaSupplier	readJdbcP(SupplierEx<? extends Connection> newConnectionFn, ToResultSetFunction resultSetFn, FunctionEx<? super ResultSet,? extends T> mapOutputFn)	Returns a supplier of processors for Sources.jdbc(SupplierEx, ToResultSetFunction, FunctionEx).
static <T> ProcessorMetaSupplier	readJdbcP(DataConnectionRef dataConnectionRef, ToResultSetFunction resultSetFn, FunctionEx<? super ResultSet,? extends T> mapOutputFn)	Returns a supplier of processors for Sources.jdbc(DataConnectionRef, ToResultSetFunction, FunctionEx).
static <T> ProcessorMetaSupplier	readJdbcP(String connectionURL, String query, FunctionEx<? super ResultSet,? extends T> mapOutputFn)	Returns a supplier of processors for Sources.jdbc(String, String, FunctionEx).
static <T> ProcessorMetaSupplier	readJdbcP(String connectionURL, String query, Properties properties, FunctionEx<? super ResultSet,? extends T> mapOutputFn)
static ProcessorMetaSupplier	readListP(String listName)	Returns a supplier of processors for Sources.list(String).
static ProcessorMetaSupplier	readMapP(String mapName)	Returns a supplier of processors for Sources.map(String).
static <T, K, V> ProcessorMetaSupplier	readMapP(String mapName, Predicate<K,V> predicate, Projection<? super Map.Entry<K,V>,? extends T> projection)	Returns a supplier of processors for Sources.map(String, Predicate, Projection).
static ProcessorSupplier	readRemoteCacheP(String cacheName, ClientConfig clientConfig)	Returns a supplier of processors for Sources.remoteCache(String, ClientConfig).
static ProcessorMetaSupplier	readRemoteListP(String listName, ClientConfig clientConfig)	Returns a supplier of processors for Sources.remoteList(String, ClientConfig).
static <T, K, V> ProcessorMetaSupplier	streamCacheP(String cacheName, PredicateEx<? super EventJournalCacheEvent<K,V>> predicateFn, FunctionEx<? super EventJournalCacheEvent<K,V>,? extends T> projectionFn, JournalInitialPosition initialPos, EventTimePolicy<? super T> eventTimePolicy)	Returns a supplier of processors for Sources.cacheJournal(String, JournalInitialPosition, FunctionEx, PredicateEx).
static <K, V> ProcessorMetaSupplier	streamCacheP(String cacheName, JournalInitialPosition initialPos, EventTimePolicy<? super Map.Entry<K,V>> eventTimePolicy)	Returns a supplier of processors for Sources.cacheJournal(String, JournalInitialPosition).
static ProcessorMetaSupplier	streamFilesP(String watchedDirectory, Charset charset, String glob, boolean sharedFileSystem, BiFunctionEx<? super String,? super String,?> mapOutputFn)	Returns a supplier of processors for Sources.filesBuilder(java.lang.String).
static <T> ProcessorMetaSupplier	streamJmsQueueP(String destination, ProcessingGuarantee maxGuarantee, EventTimePolicy<? super T> eventTimePolicy, SupplierEx<? extends jakarta.jms.Connection> newConnectionFn, FunctionEx<? super jakarta.jms.Session,? extends jakarta.jms.MessageConsumer> consumerFn, FunctionEx<? super jakarta.jms.Message,?> messageIdFn, FunctionEx<? super jakarta.jms.Message,? extends T> projectionFn)	Returns a supplier of processors for Sources.jmsQueueBuilder(com.hazelcast.function.SupplierEx<? extends jakarta.jms.ConnectionFactory>).
static <T> ProcessorMetaSupplier	streamJmsTopicP(String destination, boolean isSharedConsumer, ProcessingGuarantee maxGuarantee, EventTimePolicy<? super T> eventTimePolicy, SupplierEx<? extends jakarta.jms.Connection> newConnectionFn, FunctionEx<? super jakarta.jms.Session,? extends jakarta.jms.MessageConsumer> consumerFn, FunctionEx<? super jakarta.jms.Message,?> messageIdFn, FunctionEx<? super jakarta.jms.Message,? extends T> projectionFn)	Returns a supplier of processors for Sources.jmsTopicBuilder(com.hazelcast.function.SupplierEx<? extends jakarta.jms.ConnectionFactory>).
static <T, K, V> ProcessorMetaSupplier	streamMapP(String mapName, PredicateEx<? super EventJournalMapEvent<K,V>> predicateFn, FunctionEx<? super EventJournalMapEvent<K,V>,? extends T> projectionFn, JournalInitialPosition initialPos, EventTimePolicy<? super T> eventTimePolicy)	Returns a supplier of processors for Sources.mapJournal(String, JournalInitialPosition, FunctionEx, PredicateEx).
static <K, V> ProcessorMetaSupplier	streamMapP(String mapName, JournalInitialPosition initialPos, EventTimePolicy<? super Map.Entry<K,V>> eventTimePolicy)	Returns a supplier of processors for Sources.mapJournal(String, JournalInitialPosition).
static <T, K, V> ProcessorMetaSupplier	streamRemoteCacheP(String cacheName, ClientConfig clientConfig, PredicateEx<? super EventJournalCacheEvent<K,V>> predicateFn, FunctionEx<? super EventJournalCacheEvent<K,V>,? extends T> projectionFn, JournalInitialPosition initialPos, EventTimePolicy<? super T> eventTimePolicy)	Returns a supplier of processors for Sources.remoteCacheJournal(String, ClientConfig, JournalInitialPosition, FunctionEx, PredicateEx).
static <K, V> ProcessorMetaSupplier	streamRemoteCacheP(String cacheName, ClientConfig clientConfig, JournalInitialPosition initialPos, EventTimePolicy<? super Map.Entry<K,V>> eventTimePolicy)	Returns a supplier of processors for Sources.remoteCacheJournal(String, ClientConfig, JournalInitialPosition).
static <T, K, V> ProcessorMetaSupplier	streamRemoteMapP(String mapName, ClientConfig clientConfig, PredicateEx<? super EventJournalMapEvent<K,V>> predicateFn, FunctionEx<? super EventJournalMapEvent<K,V>,? extends T> projectionFn, JournalInitialPosition initialPos, EventTimePolicy<? super T> eventTimePolicy)	Returns a supplier of processors for Sources.remoteMapJournal(String, ClientConfig, JournalInitialPosition, FunctionEx, PredicateEx).
static <K, V> ProcessorMetaSupplier	streamRemoteMapP(String mapName, ClientConfig clientConfig, JournalInitialPosition initialPos, EventTimePolicy<? super Map.Entry<K,V>> eventTimePolicy)	Returns a supplier of processors for Sources.remoteMapJournal(String, ClientConfig, JournalInitialPosition).
static ProcessorMetaSupplier	streamSocketP(String host, int port, Charset charset)	Returns a supplier of processors for Sources.socket(String, int, Charset).

Methods inherited from class java.lang.Object

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

Method Details

readMapP

@Nonnull
public static ProcessorMetaSupplier readMapP(@Nonnull
 String mapName)

Returns a supplier of processors for Sources.map(String).

readMapP

@Nonnull
public static <T,
K,
V> ProcessorMetaSupplier readMapP(@Nonnull
 String mapName,
 @Nonnull
 Predicate<K,V> predicate,
 @Nonnull
 Projection<? super Map.Entry<K,V>,? extends T> projection)

Returns a supplier of processors for Sources.map(String, Predicate, Projection).

streamMapP

@Nonnull
public static <K,
V> ProcessorMetaSupplier streamMapP(@Nonnull
 String mapName,
 @Nonnull
 JournalInitialPosition initialPos,
 @Nonnull
 EventTimePolicy<? super Map.Entry<K,V>> eventTimePolicy)

Returns a supplier of processors for Sources.mapJournal(String, JournalInitialPosition).

streamMapP

@Nonnull
public static <T,
K,
V> ProcessorMetaSupplier streamMapP(@Nonnull
 String mapName,
 @Nonnull
 PredicateEx<? super EventJournalMapEvent<K,V>> predicateFn,
 @Nonnull
 FunctionEx<? super EventJournalMapEvent<K,V>,? extends T> projectionFn,
 @Nonnull
 JournalInitialPosition initialPos,
 @Nonnull
 EventTimePolicy<? super T> eventTimePolicy)

Returns a supplier of processors for Sources.mapJournal(String, JournalInitialPosition, FunctionEx, PredicateEx).

streamRemoteMapP

@Nonnull
public static <K,
V> ProcessorMetaSupplier streamRemoteMapP(@Nonnull
 String mapName,
 @Nonnull
 ClientConfig clientConfig,
 @Nonnull
 JournalInitialPosition initialPos,
 @Nonnull
 EventTimePolicy<? super Map.Entry<K,V>> eventTimePolicy)

Returns a supplier of processors for Sources.remoteMapJournal(String, ClientConfig, JournalInitialPosition).

streamRemoteMapP

@Nonnull
public static <T,
K,
V> ProcessorMetaSupplier streamRemoteMapP(@Nonnull
 String mapName,
 @Nonnull
 ClientConfig clientConfig,
 @Nonnull
 PredicateEx<? super EventJournalMapEvent<K,V>> predicateFn,
 @Nonnull
 FunctionEx<? super EventJournalMapEvent<K,V>,? extends T> projectionFn,
 @Nonnull
 JournalInitialPosition initialPos,
 @Nonnull
 EventTimePolicy<? super T> eventTimePolicy)

Returns a supplier of processors for Sources.remoteMapJournal(String, ClientConfig, JournalInitialPosition, FunctionEx, PredicateEx).

readCacheP

@Nonnull
public static ProcessorMetaSupplier readCacheP(@Nonnull
 String cacheName)

Returns a supplier of processors for Sources.cache(String).

streamCacheP

@Nonnull
public static <K,
V> ProcessorMetaSupplier streamCacheP(@Nonnull
 String cacheName,
 @Nonnull
 JournalInitialPosition initialPos,
 @Nonnull
 EventTimePolicy<? super Map.Entry<K,V>> eventTimePolicy)

Returns a supplier of processors for Sources.cacheJournal(String, JournalInitialPosition).

streamCacheP

@Nonnull
public static <T,
K,
V> ProcessorMetaSupplier streamCacheP(@Nonnull
 String cacheName,
 @Nonnull
 PredicateEx<? super EventJournalCacheEvent<K,V>> predicateFn,
 @Nonnull
 FunctionEx<? super EventJournalCacheEvent<K,V>,? extends T> projectionFn,
 @Nonnull
 JournalInitialPosition initialPos,
 @Nonnull
 EventTimePolicy<? super T> eventTimePolicy)

Returns a supplier of processors for Sources.cacheJournal(String, JournalInitialPosition, FunctionEx, PredicateEx).

readRemoteCacheP

@Nonnull
public static ProcessorSupplier readRemoteCacheP(@Nonnull
 String cacheName,
 @Nonnull
 ClientConfig clientConfig)

Returns a supplier of processors for Sources.remoteCache(String, ClientConfig).

streamRemoteCacheP

@Nonnull
public static <K,
V> ProcessorMetaSupplier streamRemoteCacheP(@Nonnull
 String cacheName,
 @Nonnull
 ClientConfig clientConfig,
 @Nonnull
 JournalInitialPosition initialPos,
 @Nonnull
 EventTimePolicy<? super Map.Entry<K,V>> eventTimePolicy)

Returns a supplier of processors for Sources.remoteCacheJournal(String, ClientConfig, JournalInitialPosition).

streamRemoteCacheP

@Nonnull
public static <T,
K,
V> ProcessorMetaSupplier streamRemoteCacheP(@Nonnull
 String cacheName,
 @Nonnull
 ClientConfig clientConfig,
 @Nonnull
 PredicateEx<? super EventJournalCacheEvent<K,V>> predicateFn,
 @Nonnull
 FunctionEx<? super EventJournalCacheEvent<K,V>,? extends T> projectionFn,
 @Nonnull
 JournalInitialPosition initialPos,
 @Nonnull
 EventTimePolicy<? super T> eventTimePolicy)

Returns a supplier of processors for Sources.remoteCacheJournal(String, ClientConfig, JournalInitialPosition, FunctionEx, PredicateEx).

readListP

@Nonnull
public static ProcessorMetaSupplier readListP(@Nonnull
 String listName)

Returns a supplier of processors for Sources.list(String).

readRemoteListP

@Nonnull
public static ProcessorMetaSupplier readRemoteListP(@Nonnull
 String listName,
 @Nonnull
 ClientConfig clientConfig)

Returns a supplier of processors for Sources.remoteList(String, ClientConfig).

streamSocketP

@Nonnull
public static ProcessorMetaSupplier streamSocketP(@Nonnull
 String host,
 int port,
 @Nonnull
 Charset charset)

Returns a supplier of processors for Sources.socket(String, int, Charset).

readFilesP

@Nonnull
public static <R> ProcessorMetaSupplier readFilesP(@Nonnull
 String directory,
 @Nonnull
 Charset charset,
 @Nonnull
 String glob,
 boolean sharedFileSystem,
 @Nonnull
 BiFunctionEx<? super String,? super String,? extends R> mapOutputFn)

Returns a supplier of processors for Sources.filesBuilder(java.lang.String). See FileSourceBuilder.build() for more details.

readFilesP

@Nonnull
public static <I> ProcessorMetaSupplier readFilesP(@Nonnull
 String directory,
 @Nonnull
 String glob,
 boolean sharedFileSystem,
 @Nonnull
 FunctionEx<? super Path,? extends Stream<I>> readFileFn)

Returns a supplier of processors for Sources.filesBuilder(java.lang.String). See FileSourceBuilder.build() for more details.

readFilesP

@Nonnull
public static <I> ProcessorMetaSupplier readFilesP(@Nonnull
 String directory,
 @Nonnull
 String glob,
 boolean sharedFileSystem,
 boolean ignoreFileNotFound,
 @Nonnull
 FunctionEx<? super Path,? extends Stream<I>> readFileFn)

Returns a supplier of processors for FileSources.files(String) to read local files.

streamFilesP

@Nonnull
public static ProcessorMetaSupplier streamFilesP(@Nonnull
 String watchedDirectory,
 @Nonnull
 Charset charset,
 @Nonnull
 String glob,
 boolean sharedFileSystem,
 @Nonnull
 BiFunctionEx<? super String,? super String,?> mapOutputFn)

Returns a supplier of processors for Sources.filesBuilder(java.lang.String). See FileSourceBuilder.buildWatcher() for more details.

streamJmsQueueP

@Nonnull
public static <T> ProcessorMetaSupplier streamJmsQueueP(@Nullable
 String destination,
 @Nonnull
 ProcessingGuarantee maxGuarantee,
 @Nonnull
 EventTimePolicy<? super T> eventTimePolicy,
 @Nonnull
 SupplierEx<? extends jakarta.jms.Connection> newConnectionFn,
 @Nonnull
 FunctionEx<? super jakarta.jms.Session,? extends jakarta.jms.MessageConsumer> consumerFn,
 @Nonnull
 FunctionEx<? super jakarta.jms.Message,?> messageIdFn,
 @Nonnull
 FunctionEx<? super jakarta.jms.Message,? extends T> projectionFn)

Returns a supplier of processors for Sources.jmsQueueBuilder(com.hazelcast.function.SupplierEx<? extends jakarta.jms.ConnectionFactory>).

Parameters:

maxGuarantee - maximum processing guarantee for the source. You can use it to disable acknowledging in transactions to save transaction overhead

streamJmsTopicP

@Nonnull
public static <T> ProcessorMetaSupplier streamJmsTopicP(@Nullable
 String destination,
 boolean isSharedConsumer,
 @Nonnull
 ProcessingGuarantee maxGuarantee,
 @Nonnull
 EventTimePolicy<? super T> eventTimePolicy,
 @Nonnull
 SupplierEx<? extends jakarta.jms.Connection> newConnectionFn,
 @Nonnull
 FunctionEx<? super jakarta.jms.Session,? extends jakarta.jms.MessageConsumer> consumerFn,
 @Nonnull
 FunctionEx<? super jakarta.jms.Message,?> messageIdFn,
 @Nonnull
 FunctionEx<? super jakarta.jms.Message,? extends T> projectionFn)

Returns a supplier of processors for Sources.jmsTopicBuilder(com.hazelcast.function.SupplierEx<? extends jakarta.jms.ConnectionFactory>).

Parameters:

isSharedConsumer - true, if createSharedConsumer() or createSharedDurableConsumer() was used to create the consumer in the consumerFn

maxGuarantee - maximum processing guarantee for the source. You can use it to disable acknowledging in transactions to save transaction overhead

readJdbcP

public static <T> ProcessorMetaSupplier readJdbcP(@Nonnull
 SupplierEx<? extends Connection> newConnectionFn,
 @Nonnull
 ToResultSetFunction resultSetFn,
 @Nonnull
 FunctionEx<? super ResultSet,? extends T> mapOutputFn)

Returns a supplier of processors for Sources.jdbc(SupplierEx, ToResultSetFunction, FunctionEx).

readJdbcP

public static <T> ProcessorMetaSupplier readJdbcP(@Nonnull
 DataConnectionRef dataConnectionRef,
 @Nonnull
 ToResultSetFunction resultSetFn,
 @Nonnull
 FunctionEx<? super ResultSet,? extends T> mapOutputFn)

Returns a supplier of processors for Sources.jdbc(DataConnectionRef, ToResultSetFunction, FunctionEx).

Since:

5.2

readJdbcP

public static <T> ProcessorMetaSupplier readJdbcP(@Nonnull
 String connectionURL,
 @Nonnull
 String query,
 @Nonnull
 FunctionEx<? super ResultSet,? extends T> mapOutputFn)

Returns a supplier of processors for Sources.jdbc(String, String, FunctionEx).

readJdbcP

public static <T> ProcessorMetaSupplier readJdbcP(@Nonnull
 String connectionURL,
 @Nonnull
 String query,
 @Nonnull
 Properties properties,
 @Nonnull
 FunctionEx<? super ResultSet,? extends T> mapOutputFn)

convenientSourceP

@Nonnull
public static <C,
T,
S> ProcessorMetaSupplier convenientSourceP(@Nonnull
 FunctionEx<? super Processor.Context,? extends C> createFn,
 @Nonnull
 BiConsumerEx<? super C,? super SourceBuilder.SourceBuffer<T>> fillBufferFn,
 @Nonnull
 FunctionEx<? super C,? extends S> createSnapshotFn,
 @Nonnull
 BiConsumerEx<? super C,? super List<S>> restoreSnapshotFn,
 @Nonnull
 ConsumerEx<? super C> destroyFn,
 int preferredLocalParallelism,
 boolean isBatch,
 @Nullable
 Permission permission)

Returns a supplier of processors for a source that the user can create using the SourceBuilder. This variant creates a source that emits items without timestamps.

Type Parameters:

C - type of the source's context object

T - type of items the source emits

S - type of object saved to state snapshot

Parameters:

createFn - function that creates the source's context object

fillBufferFn - function that fills Jet's buffer with items to emit

createSnapshotFn - function that returns a snapshot of the context object's state

restoreSnapshotFn - function that restores the context object's state from a snapshot

destroyFn - function that cleans up the resources held by the context object

preferredLocalParallelism - preferred local parallelism of the source vertex. Special values: -1 -> use the cluster's default local parallelism; 0 -> create a single processor for the entire cluster (total parallelism = 1)

isBatch - true, if the fillBufferFn will call buffer.close(), that is whether the source reads a bounded or unbounded set of data

convenientTimestampedSourceP

@Nonnull
public static <C,
T,
S> ProcessorMetaSupplier convenientTimestampedSourceP(@Nonnull
 FunctionEx<? super Processor.Context,? extends C> createFn,
 @Nonnull
 BiConsumerEx<? super C,? super SourceBuilder.TimestampedSourceBuffer<T>> fillBufferFn,
 @Nonnull
 EventTimePolicy<? super T> eventTimePolicy,
 @Nonnull
 FunctionEx<? super C,? extends S> createSnapshotFn,
 @Nonnull
 BiConsumerEx<? super C,? super List<S>> restoreSnapshotFn,
 @Nonnull
 ConsumerEx<? super C> destroyFn,
 int preferredLocalParallelism)

Returns a supplier of processors for a source that the user can create using the SourceBuilder. This variant creates a source that emits timestamped events.

Type Parameters:

C - type of the context object

T - type of items the source emits

S - type of the object saved to state snapshot

Parameters:

createFn - function that creates the source's context object

fillBufferFn - function that fills Jet's buffer with items to emit

eventTimePolicy - parameters for watermark generation

createSnapshotFn - function that returns a snapshot of the context object's state

restoreSnapshotFn - function that restores the context object's state from a snapshot

destroyFn - function that cleans up the resources held by the context object

preferredLocalParallelism - preferred local parallelism of the source vertex. Special values: -1 -> use the cluster's default local parallelism; 0 -> create a single processor for the entire cluster (total parallelism = 1)