Sinks { @Nonnull public static <T extends Entry> Sink<T> remoteCache( @Nonnull String cacheName, @Nonnull ClientConfig clientConfig ) { return fromProcessor("remoteCacheSink(" + cacheName + ')', writeRemoteCacheP(cacheName, clientConfig)); } private  Sinks(); @Nonnull static Sink<T> fromProcessor(
            @Nonnull String sinkName,
            @Nonnull ProcessorMetaSupplier metaSupplier
    ); @Nonnull static Sink<T> fromProcessor(
            @Nonnull String sinkName,
            @Nonnull ProcessorMetaSupplier metaSupplier,
            @Nullable FunctionEx<? super T, ?> partitionKeyFn
    ); @Nonnull static Sink<Entry<K, V>> map(@Nonnull String mapName); @Nonnull static Sink<Entry<K, V>> map(@Nonnull IMap<? super K, ? super V> map); @Nonnull static Sink<T> map(
            @Nonnull String mapName,
            @Nonnull FunctionEx<? super T, ? extends K> toKeyFn,
            @Nonnull FunctionEx<? super T, ? extends V> toValueFn

    ); @Nonnull static Sink<T> map(
            @Nonnull IMap<? super K, ? super V> map,
            @Nonnull FunctionEx<? super T, ? extends K> toKeyFn,
            @Nonnull FunctionEx<? super T, ? extends V> toValueFn

    ); @Nonnull static Sink<Entry<K, V>> remoteMap(@Nonnull String mapName, @Nonnull ClientConfig clientConfig); @Nonnull static Sink<T> remoteMap(
            @Nonnull String mapName,
            @Nonnull ClientConfig clientConfig,
            @Nonnull FunctionEx<? super T, ? extends K> toKeyFn,
            @Nonnull FunctionEx<? super T, ? extends V> toValueFn
    ); @Nonnull static Sink<T> mapWithMerging(
            @Nonnull String mapName,
            @Nonnull FunctionEx<? super T, ? extends K> toKeyFn,
            @Nonnull FunctionEx<? super T, ? extends V> toValueFn,
            @Nonnull BinaryOperatorEx<V> mergeFn
    ); @Nonnull static Sink<T> mapWithMerging(
            @Nonnull IMap<? super K, ? super V> map,
            @Nonnull FunctionEx<? super T, ? extends K> toKeyFn,
            @Nonnull FunctionEx<? super T, ? extends V> toValueFn,
            @Nonnull BinaryOperatorEx<V> mergeFn
    ); @Nonnull static Sink<T> remoteMapWithMerging(
            @Nonnull String mapName,
            @Nonnull ClientConfig clientConfig,
            @Nonnull FunctionEx<? super T, ? extends K> toKeyFn,
            @Nonnull FunctionEx<? super T, ? extends V> toValueFn,
            @Nonnull BinaryOperatorEx<V> mergeFn
    ); @Nonnull static Sink<Entry<K, V>> mapWithMerging(
            @Nonnull String mapName,
            @Nonnull BinaryOperatorEx<V> mergeFn
    ); @Nonnull static Sink<Entry<K, V>> mapWithMerging(
            @Nonnull IMap<? super K, V> map,
            @Nonnull BinaryOperatorEx<V> mergeFn
    ); @Nonnull static Sink<Entry<K, V>> remoteMapWithMerging(
            @Nonnull String mapName,
            @Nonnull ClientConfig clientConfig,
            @Nonnull BinaryOperatorEx<V> mergeFn
    ); @Nonnull static Sink<T> mapWithUpdating(
            @Nonnull String mapName,
            @Nonnull FunctionEx<? super T, ? extends K> toKeyFn,
            @Nonnull BiFunctionEx<? super V, ? super T, ? extends V> updateFn
    ); @Nonnull static Sink<T> mapWithUpdating(
            @Nonnull IMap<? super K, ? super V> map,
            @Nonnull FunctionEx<? super T, ? extends K> toKeyFn,
            @Nonnull BiFunctionEx<? super V, ? super T, ? extends V> updateFn
    ); @Nonnull static Sink<T> remoteMapWithUpdating(
            @Nonnull String mapName,
            @Nonnull ClientConfig clientConfig,
            @Nonnull FunctionEx<? super T, ? extends K> toKeyFn,
            @Nonnull BiFunctionEx<? super V, ? super T, ? extends V> updateFn
    ); @Nonnull static Sink<E> mapWithUpdating(
            @Nonnull String mapName,
            @Nonnull BiFunctionEx<? super V, ? super E, ? extends V> updateFn
    ); @Nonnull static Sink<E> mapWithUpdating(
            @Nonnull IMap<? super K, ? super V> map,
            @Nonnull BiFunctionEx<? super V, ? super E, ? extends V> updateFn
    ); @Nonnull static Sink<E> remoteMapWithUpdating(
            @Nonnull String mapName,
            @Nonnull ClientConfig clientConfig,
            @Nonnull BiFunctionEx<? super V, ? super E, ? extends V> updateFn
    ); @Nonnull static Sink<E> mapWithEntryProcessor(
            @Nonnull String mapName,
            @Nonnull FunctionEx<? super E, ? extends K> toKeyFn,
            @Nonnull FunctionEx<? super E, ? extends EntryProcessor<K, V, R>> toEntryProcessorFn
    ); @Nonnull static Sink<E> mapWithEntryProcessor(
            int maxParallelAsyncOps,
            @Nonnull String mapName,
            @Nonnull FunctionEx<? super E, ? extends K> toKeyFn,
            @Nonnull FunctionEx<? super E, ? extends EntryProcessor<K, V, R>> toEntryProcessorFn
    ); @Nonnull static Sink<T> mapWithEntryProcessor(
            @Nonnull IMap<? super K, ? super V> map,
            @Nonnull FunctionEx<? super T, ? extends K> toKeyFn,
            @Nonnull FunctionEx<? super T, ? extends EntryProcessor<K, V, R>> toEntryProcessorFn
    ); @Nonnull static Sink<E> remoteMapWithEntryProcessor(
            @Nonnull String mapName,
            @Nonnull ClientConfig clientConfig,
            @Nonnull FunctionEx<? super E, ? extends K> toKeyFn,
            @Nonnull FunctionEx<? super E, ? extends EntryProcessor<K, V, R>> toEntryProcessorFn
    ); @Nonnull static Sink<T> cache(@Nonnull String cacheName); @Nonnull static Sink<T> remoteCache(
            @Nonnull String cacheName,
            @Nonnull ClientConfig clientConfig
    ); @Nonnull static Sink<T> list(@Nonnull String listName); @Nonnull static Sink<T> list(@Nonnull IList<? super T> list); @Nonnull static Sink<T> remoteList(@Nonnull String listName, @Nonnull ClientConfig clientConfig); @Nonnull static Sink<T> reliableTopic(@Nonnull String reliableTopicName); @Nonnull static Sink<T> reliableTopic(@Nonnull ITopic<Object> reliableTopic); @Nonnull static Sink<T> remoteReliableTopic(@Nonnull String reliableTopicName, @Nonnull ClientConfig clientConfig); @Nonnull static Sink<T> socket(
            @Nonnull String host,
            int port,
            @Nonnull FunctionEx<? super T, ? extends String> toStringFn,
            @Nonnull Charset charset
    ); @Nonnull static Sink<T> socket(
            @Nonnull String host,
            int port,
            @Nonnull FunctionEx<? super T, ? extends String> toStringFn
    ); @Nonnull static Sink<T> socket(@Nonnull String host, int port); @Nonnull static FileSinkBuilder<T> filesBuilder(@Nonnull String directoryName); @Nonnull static Sink<T> files(@Nonnull String directoryName); @Nonnull static Sink<T> json(@Nonnull String directoryName); @Nonnull static Sink<T> logger(@Nonnull FunctionEx<? super T, String> toStringFn); @Nonnull static Sink<T> logger(); @Nonnull static Sink<T> noop(); @Nonnull static Sink<T> jmsQueue(
            @Nonnull String queueName,
            @Nonnull SupplierEx<ConnectionFactory> factorySupplier
    ); @Nonnull static JmsSinkBuilder<T> jmsQueueBuilder(@Nonnull SupplierEx<ConnectionFactory> factorySupplier); @Nonnull static Sink<T> jmsTopic(
            @Nonnull String topicName,
            @Nonnull SupplierEx<ConnectionFactory> factorySupplier
    ); @Nonnull static JmsSinkBuilder<T> jmsTopicBuilder(@Nonnull SupplierEx<ConnectionFactory> factorySupplier); @Nonnull static Sink<T> jdbc(
            @Nonnull String updateQuery,
            @Nonnull SupplierEx<? extends CommonDataSource> dataSourceSupplier,
            @Nonnull BiConsumerEx<PreparedStatement, T> bindFn
    ); @Nonnull static Sink<T> jdbc(
            @Nonnull String updateQuery,
            @Nonnull String jdbcUrl,
            @Nonnull BiConsumerEx<PreparedStatement, T> bindFn
    ); @Nonnull static JdbcSinkBuilder<T> jdbcBuilder(); @Nonnull static Sink<T> observable(String name); @Nonnull static Sink<T> observable(Observable<? super T> observable);  }
@Test public void remoteCache() { List<Integer> input = sequence(itemCount); putToBatchSrcCache(input); Sink<Entry<String, Integer>> sink = Sinks.remoteCache(sinkName, clientConfig); p.readFrom(Sources.<String, Integer>cache(srcName)).writeTo(sink); execute(); List<Entry<String, Integer>> expected = input.stream() .map(i -> entry(String.valueOf(i), i)) .collect(toList()); ICache<String, Integer> remoteCache = remoteHz.getCacheManager().getCache(sinkName); assertEquals(expected.size(), remoteCache.size()); expected.forEach(entry -> assertEquals(entry.getValue(), remoteCache.get(entry.getKey()))); }