Sources { @Nonnull public static StreamSource<String> fileWatcher(@Nonnull String watchedDirectory) { return filesBuilder(watchedDirectory).buildWatcher(); } private  Sources(); @Nonnull static BatchSource<T> batchFromProcessor(
            @Nonnull String sourceName,
            @Nonnull ProcessorMetaSupplier metaSupplier
    ); @Nonnull static StreamSource<T> streamFromProcessorWithWatermarks(
            @Nonnull String sourceName,
            boolean supportsNativeTimestamps,
            @Nonnull FunctionEx<EventTimePolicy<? super T>, ProcessorMetaSupplier> metaSupplierFn
    ); @Nonnull static StreamSource<T> streamFromProcessor(
            @Nonnull String sourceName,
            @Nonnull ProcessorMetaSupplier metaSupplier
    ); @Nonnull static BatchSource<Entry<K, V>> map(@Nonnull String mapName); @Nonnull static BatchSource<Entry<K, V>> map(@Nonnull IMap<? extends K, ? extends V> map); @Nonnull static BatchSource<T> map(
            @Nonnull String mapName,
            @Nonnull Predicate<K, V> predicate,
            @Nonnull Projection<? super Entry<K, V>, ? extends T> projection
    ); @Nonnull static BatchSource<T> map(
            @Nonnull IMap<? extends K, ? extends V> map,
            @Nonnull Predicate<K, V> predicate,
            @Nonnull Projection<? super Entry<K, V>, ? extends T> projection
    ); @Nonnull static StreamSource<T> mapJournal(
            @Nonnull String mapName,
            @Nonnull JournalInitialPosition initialPos,
            @Nonnull FunctionEx<? super EventJournalMapEvent<K, V>, ? extends T> projectionFn,
            @Nonnull PredicateEx<? super EventJournalMapEvent<K, V>> predicateFn
    ); @Nonnull static StreamSource<T> mapJournal(
            @Nonnull IMap<? extends K, ? extends V> map,
            @Nonnull JournalInitialPosition initialPos,
            @Nonnull FunctionEx<? super EventJournalMapEvent<K, V>, ? extends T> projectionFn,
            @Nonnull PredicateEx<? super EventJournalMapEvent<K, V>> predicateFn
    ); @Nonnull static StreamSource<Entry<K, V>> mapJournal(
            @Nonnull String mapName,
            @Nonnull JournalInitialPosition initialPos
    ); @Nonnull static StreamSource<Entry<K, V>> mapJournal(
            @Nonnull IMap<? extends K, ? extends V> map,
            @Nonnull JournalInitialPosition initialPos
    ); @Nonnull static BatchSource<Entry<K, V>> remoteMap(
            @Nonnull String mapName,
            @Nonnull ClientConfig clientConfig
    ); @Nonnull static BatchSource<T> remoteMap(
            @Nonnull String mapName,
            @Nonnull ClientConfig clientConfig,
            @Nonnull Predicate<K, V> predicate,
            @Nonnull Projection<? super Entry<K, V>, ? extends T> projection
    ); @Nonnull static StreamSource<T> remoteMapJournal(
            @Nonnull String mapName,
            @Nonnull ClientConfig clientConfig,
            @Nonnull JournalInitialPosition initialPos,
            @Nonnull FunctionEx<? super EventJournalMapEvent<K, V>, ? extends T> projectionFn,
            @Nonnull PredicateEx<? super EventJournalMapEvent<K, V>> predicateFn
    ); @Nonnull static StreamSource<Entry<K, V>> remoteMapJournal(
            @Nonnull String mapName,
            @Nonnull ClientConfig clientConfig,
            @Nonnull JournalInitialPosition initialPos
    ); @Nonnull static BatchSource<Entry<K, V>> cache(@Nonnull String cacheName); @Nonnull static StreamSource<T> cacheJournal(
            @Nonnull String cacheName,
            @Nonnull JournalInitialPosition initialPos,
            @Nonnull FunctionEx<? super EventJournalCacheEvent<K, V>, ? extends T> projectionFn,
            @Nonnull PredicateEx<? super EventJournalCacheEvent<K, V>> predicateFn
    ); @Nonnull static StreamSource<Entry<K, V>> cacheJournal(
            @Nonnull String cacheName,
            @Nonnull JournalInitialPosition initialPos
    ); @Nonnull static BatchSource<Entry<K, V>> remoteCache(
            @Nonnull String cacheName,
            @Nonnull ClientConfig clientConfig
    ); @Nonnull static StreamSource<T> remoteCacheJournal(
            @Nonnull String cacheName,
            @Nonnull ClientConfig clientConfig,
            @Nonnull JournalInitialPosition initialPos,
            @Nonnull FunctionEx<? super EventJournalCacheEvent<K, V>, ? extends T> projectionFn,
            @Nonnull PredicateEx<? super EventJournalCacheEvent<K, V>> predicateFn
    ); @Nonnull static StreamSource<Entry<K, V>> remoteCacheJournal(
            @Nonnull String cacheName,
            @Nonnull ClientConfig clientConfig,
            @Nonnull JournalInitialPosition initialPos
    ); @Nonnull static BatchSource<T> list(@Nonnull String listName); @Nonnull static BatchSource<T> list(@Nonnull IList<? extends T> list); @Nonnull static BatchSource<T> remoteList(@Nonnull String listName, @Nonnull ClientConfig clientConfig); @Nonnull static StreamSource<String> socket(
            @Nonnull String host, int port, @Nonnull Charset charset
    ); @Nonnull static StreamSource<String> socket(@Nonnull String host, int port); @Nonnull static FileSourceBuilder filesBuilder(@Nonnull String directory); @Nonnull static BatchSource<String> files(@Nonnull String directory); @Nonnull static BatchSource<T> json(@Nonnull String directory, @Nonnull Class<T> type); @Nonnull static BatchSource<Map<String, Object>> json(@Nonnull String directory); @Nonnull static StreamSource<String> fileWatcher(@Nonnull String watchedDirectory); @Nonnull static StreamSource<T> jsonWatcher(@Nonnull String watchedDirectory, @Nonnull Class<T> type); @Nonnull static StreamSource<Map<String, Object>> jsonWatcher(@Nonnull String watchedDirectory); @Nonnull @Deprecated static StreamSource<Message> jmsQueue(
            @Nonnull SupplierEx<? extends ConnectionFactory> factorySupplier,
            @Nonnull String name
    ); @Nonnull static StreamSource<Message> jmsQueue(
            @Nonnull String name,
            @Nonnull SupplierEx<? extends ConnectionFactory> factorySupplier
    ); @Nonnull static JmsSourceBuilder jmsQueueBuilder(SupplierEx<? extends ConnectionFactory> factorySupplier); @Nonnull @Deprecated static StreamSource<Message> jmsTopic(
            @Nonnull SupplierEx<? extends ConnectionFactory> factorySupplier,
            @Nonnull String name
    ); @Nonnull static StreamSource<Message> jmsTopic(
            @Nonnull String name,
            @Nonnull SupplierEx<? extends ConnectionFactory> factorySupplier
    ); @Nonnull static JmsSourceBuilder jmsTopicBuilder(SupplierEx<? extends ConnectionFactory> factorySupplier); static BatchSource<T> jdbc(
            @Nonnull SupplierEx<? extends Connection> newConnectionFn,
            @Nonnull ToResultSetFunction resultSetFn,
            @Nonnull FunctionEx<? super ResultSet, ? extends T> createOutputFn
    ); static BatchSource<T> jdbc(
            @Nonnull String connectionURL,
            @Nonnull String query,
            @Nonnull FunctionEx<? super ResultSet, ? extends T> createOutputFn
    );  }
@Test @Ignore("Changes on the file is not reflected as an event from the File System, needs more investigation") public void fileChanges() throws Exception { File directory = createTempDirectory(); File file = new File(directory, randomName()); appendToFile(file, "hello", "pre-existing"); sleepAtLeastMillis(50); StreamSource<String> source = Sources.fileWatcher(directory.getPath()); p.readFrom(source).withoutTimestamps().writeTo(sink); Job job = jet().newJob(p); assertJobStatusEventually(job, JobStatus.RUNNING); assertEquals(0, sinkList.size()); appendToFile(file, "third line"); int nodeCount = jet().getCluster().getMembers().size(); assertTrueEventually(() -> assertEquals(nodeCount, sinkList.size())); }