Package com.hazelcast.jet.core.processor
Class SourceProcessors
java.lang.Object
com.hazelcast.jet.core.processor.SourceProcessors
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 TypeMethodDescriptionstatic <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 theSourceBuilder
.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 theSourceBuilder
.static ProcessorMetaSupplier
readCacheP
(String cacheName) Returns a supplier of processors forSources.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 forFileSources.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 forSources.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 forSources.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 forSources.jdbc(SupplierEx, ToResultSetFunction, FunctionEx)
.static <T> ProcessorMetaSupplier
readJdbcP
(DataConnectionRef dataConnectionRef, ToResultSetFunction resultSetFn, FunctionEx<? super ResultSet, ? extends T> mapOutputFn) Returns a supplier of processors forSources.jdbc(DataConnectionRef, ToResultSetFunction, FunctionEx)
.static <T> ProcessorMetaSupplier
readJdbcP
(String connectionURL, String query, FunctionEx<? super ResultSet, ? extends T> mapOutputFn) Returns a supplier of processors forSources.jdbc(String, String, FunctionEx)
.static <T> ProcessorMetaSupplier
readJdbcP
(String connectionURL, String query, Properties properties, FunctionEx<? super ResultSet, ? extends T> mapOutputFn) static ProcessorMetaSupplier
Returns a supplier of processors forSources.list(String)
.static ProcessorMetaSupplier
Returns a supplier of processors forSources.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 forSources.map(String, Predicate, Projection)
.static ProcessorSupplier
readRemoteCacheP
(String cacheName, ClientConfig clientConfig) Returns a supplier of processors forSources.remoteCache(String, ClientConfig)
.static ProcessorMetaSupplier
readRemoteListP
(String listName, ClientConfig clientConfig) Returns a supplier of processors forSources.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 forSources.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 forSources.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 forSources.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 forSources.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 forSources.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 forSources.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 forSources.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 forSources.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 forSources.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 forSources.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 forSources.remoteMapJournal(String, ClientConfig, JournalInitialPosition)
.static ProcessorMetaSupplier
streamSocketP
(String host, int port, Charset charset) Returns a supplier of processors forSources.socket(String, int, Charset)
.
-
Method Details
-
readMapP
Returns a supplier of processors forSources.map(String)
. -
readMapP
@Nonnull public static <T,K, ProcessorMetaSupplier readMapPV> (@Nonnull String mapName, @Nonnull Predicate<K, V> predicate, @Nonnull Projection<? super Map.Entry<K, V>, ? extends T> projection) Returns a supplier of processors forSources.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 forSources.mapJournal(String, JournalInitialPosition)
. -
streamMapP
@Nonnull public static <T,K, ProcessorMetaSupplier streamMapPV> (@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 forSources.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 forSources.remoteMapJournal(String, ClientConfig, JournalInitialPosition)
. -
streamRemoteMapP
@Nonnull public static <T,K, ProcessorMetaSupplier streamRemoteMapPV> (@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 forSources.remoteMapJournal(String, ClientConfig, JournalInitialPosition, FunctionEx, PredicateEx)
. -
readCacheP
Returns a supplier of processors forSources.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 forSources.cacheJournal(String, JournalInitialPosition)
. -
streamCacheP
@Nonnull public static <T,K, ProcessorMetaSupplier streamCachePV> (@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 forSources.cacheJournal(String, JournalInitialPosition, FunctionEx, PredicateEx)
. -
readRemoteCacheP
@Nonnull public static ProcessorSupplier readRemoteCacheP(@Nonnull String cacheName, @Nonnull ClientConfig clientConfig) Returns a supplier of processors forSources.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 forSources.remoteCacheJournal(String, ClientConfig, JournalInitialPosition)
. -
streamRemoteCacheP
@Nonnull public static <T,K, ProcessorMetaSupplier streamRemoteCachePV> (@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 forSources.remoteCacheJournal(String, ClientConfig, JournalInitialPosition, FunctionEx, PredicateEx)
. -
readListP
Returns a supplier of processors forSources.list(String)
. -
readRemoteListP
@Nonnull public static ProcessorMetaSupplier readRemoteListP(@Nonnull String listName, @Nonnull ClientConfig clientConfig) Returns a supplier of processors forSources.remoteList(String, ClientConfig)
. -
streamSocketP
@Nonnull public static ProcessorMetaSupplier streamSocketP(@Nonnull String host, int port, @Nonnull Charset charset) Returns a supplier of processors forSources.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 forSources.filesBuilder(java.lang.String)
. SeeFileSourceBuilder.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 forSources.filesBuilder(java.lang.String)
. SeeFileSourceBuilder.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 forFileSources.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 forSources.filesBuilder(java.lang.String)
. SeeFileSourceBuilder.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 forSources.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 forSources.jmsTopicBuilder(com.hazelcast.function.SupplierEx<? extends jakarta.jms.ConnectionFactory>)
.- Parameters:
isSharedConsumer
- true, ifcreateSharedConsumer()
orcreateSharedDurableConsumer()
was used to create the consumer in theconsumerFn
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 forSources.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 forSources.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 forSources.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, ProcessorMetaSupplier convenientSourcePS> (@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 theSourceBuilder
. This variant creates a source that emits items without timestamps.- Type Parameters:
C
- type of the source's context objectT
- type of items the source emitsS
- type of object saved to state snapshot- Parameters:
createFn
- function that creates the source's context objectfillBufferFn
- function that fills Jet's buffer with items to emitcreateSnapshotFn
- function that returns a snapshot of the context object's staterestoreSnapshotFn
- function that restores the context object's state from a snapshotdestroyFn
- function that cleans up the resources held by the context objectpreferredLocalParallelism
- 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 callbuffer.close()
, that is whether the source reads a bounded or unbounded set of data
-
convenientTimestampedSourceP
@Nonnull public static <C,T, ProcessorMetaSupplier convenientTimestampedSourcePS> (@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 theSourceBuilder
. This variant creates a source that emits timestamped events.- Type Parameters:
C
- type of the context objectT
- type of items the source emitsS
- type of the object saved to state snapshot- Parameters:
createFn
- function that creates the source's context objectfillBufferFn
- function that fills Jet's buffer with items to emiteventTimePolicy
- parameters for watermark generationcreateSnapshotFn
- function that returns a snapshot of the context object's staterestoreSnapshotFn
- function that restores the context object's state from a snapshotdestroyFn
- function that cleans up the resources held by the context objectpreferredLocalParallelism
- 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)
-