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:
- 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)
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(String connectionURL, String query, FunctionEx<? super ResultSet,? extends T> mapOutputFn)
Returns a supplier of processors forSources.jdbc(String, String, FunctionEx)
.static ProcessorMetaSupplier
readListP(String listName)
Returns a supplier of processors forSources.list(String)
.static ProcessorMetaSupplier
readMapP(String mapName)
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 ProcessorSupplier
readRemoteMapP(String mapName, ClientConfig clientConfig)
Returns a supplier of processors forSources.remoteMap(String, ClientConfig)
.static <T, K, V> ProcessorSupplier
readRemoteMapP(String mapName, ClientConfig clientConfig, Predicate<K,V> predicate, Projection<? super Map.Entry<K,V>,? extends T> projection)
Returns a supplier of processors forSources.remoteMap(String, ClientConfig, Predicate, Projection)
.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(SupplierEx<? extends javax.jms.Connection> newConnectionFn, FunctionEx<? super javax.jms.Session,? extends javax.jms.MessageConsumer> consumerFn, FunctionEx<? super javax.jms.Message,?> messageIdFn, FunctionEx<? super javax.jms.Message,? extends T> projectionFn, EventTimePolicy<? super T> eventTimePolicy, ProcessingGuarantee maxGuarantee)
Returns a supplier of processors forSources.jmsQueueBuilder(com.hazelcast.function.SupplierEx<? extends javax.jms.ConnectionFactory>)
.static <T> ProcessorMetaSupplier
streamJmsTopicP(SupplierEx<? extends javax.jms.Connection> newConnectionFn, FunctionEx<? super javax.jms.Session,? extends javax.jms.MessageConsumer> consumerFn, boolean isSharedConsumer, FunctionEx<? super javax.jms.Message,?> messageIdFn, FunctionEx<? super javax.jms.Message,? extends T> projectionFn, EventTimePolicy<? super T> eventTimePolicy, ProcessingGuarantee maxGuarantee)
Returns a supplier of processors forSources.jmsTopicBuilder(com.hazelcast.function.SupplierEx<? extends javax.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, 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 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, 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 forSources.mapJournal(String, JournalInitialPosition, FunctionEx, PredicateEx)
. -
readRemoteMapP
@Nonnull public static ProcessorSupplier readRemoteMapP(@Nonnull String mapName, @Nonnull ClientConfig clientConfig)Returns a supplier of processors forSources.remoteMap(String, ClientConfig)
. -
readRemoteMapP
@Nonnull public static <T, K, V> ProcessorSupplier readRemoteMapP(@Nonnull String mapName, @Nonnull ClientConfig clientConfig, @Nonnull Predicate<K,V> predicate, @Nonnull Projection<? super Map.Entry<K,V>,? extends T> projection)Returns a supplier of processors forSources.remoteMap(String, ClientConfig, Predicate, Projection)
. -
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, 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 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, 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 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, 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 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(@Nonnull SupplierEx<? extends javax.jms.Connection> newConnectionFn, @Nonnull FunctionEx<? super javax.jms.Session,? extends javax.jms.MessageConsumer> consumerFn, @Nonnull FunctionEx<? super javax.jms.Message,?> messageIdFn, @Nonnull FunctionEx<? super javax.jms.Message,? extends T> projectionFn, @Nonnull EventTimePolicy<? super T> eventTimePolicy, ProcessingGuarantee maxGuarantee)Returns a supplier of processors forSources.jmsQueueBuilder(com.hazelcast.function.SupplierEx<? extends javax.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(@Nonnull SupplierEx<? extends javax.jms.Connection> newConnectionFn, @Nonnull FunctionEx<? super javax.jms.Session,? extends javax.jms.MessageConsumer> consumerFn, boolean isSharedConsumer, @Nonnull FunctionEx<? super javax.jms.Message,?> messageIdFn, @Nonnull FunctionEx<? super javax.jms.Message,? extends T> projectionFn, @Nonnull EventTimePolicy<? super T> eventTimePolicy, ProcessingGuarantee maxGuarantee)Returns a supplier of processors forSources.jmsTopicBuilder(com.hazelcast.function.SupplierEx<? extends javax.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 String connectionURL, @Nonnull String query, @Nonnull FunctionEx<? super ResultSet,? extends T> mapOutputFn)Returns a supplier of processors forSources.jdbc(String, String, FunctionEx)
. -
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)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, 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 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)
-