Sinks { @Nonnull public static <K, V> Sink<Entry<K, V>> map(@Nonnull String mapName) { return map(mapName, Entry::getKey, Entry::getValue); } 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 map_withJsonKeyValue() { BatchStage<Integer> sourceStage = p.readFrom(TestSources.items(0, 1, 2, 3, 4)); sourceStage.writeTo(Sinks.map(sinkName, JsonUtil::hazelcastJsonValue, JsonUtil::hazelcastJsonValue)); execute(); IMap<HazelcastJsonValue, HazelcastJsonValue> sinkMap = jet().getMap(sinkName); assertEquals(5, sinkMap.size()); IntStream.range(0, 5).forEach(i -> assertEquals(hazelcastJsonValue(String.valueOf(i)), sinkMap.get(hazelcastJsonValue(i)))); }
@Test @SuppressWarnings("unchecked") public void test_adaptingPartitionFunction() { Pipeline p = Pipeline.create(); StreamStage<KeyedWindowResult<String, Long>> input1 = p.readFrom(TestSources.items(0)) .addTimestamps(i -> i, 0) .groupingKey(item -> "key0") .window(WindowDefinition.sliding(1, 1)) .aggregate(AggregateOperations.counting()); BatchStage<Entry<String, Long>> input2 = p.readFrom(TestSources.items(entry("key1", 2L))); IMap<String, Long> sinkMap = jet().getMap(randomMapName()); p.writeTo(Sinks.map(sinkMap), input1, input2); jet().newJob(p).join(); assertEquals(2, sinkMap.size()); assertEquals((Long) 1L, sinkMap.get("key0")); assertEquals((Long) 2L, sinkMap.get("key1")); }
@Test public void map_byName() { List<Integer> input = sequence(itemCount); putToBatchSrcMap(input); Sink<Entry<String, Integer>> sink = Sinks.map(sinkName); p.readFrom(Sources.<String, Integer>map(srcName)).writeTo(sink); execute(); List<Entry<String, Integer>> expected = input.stream() .map(i -> entry(String.valueOf(i), i)) .collect(toList()); Set<Entry<String, Integer>> actual = jet().<String, Integer>getMap(sinkName).entrySet(); assertEquals(expected.size(), actual.size()); expected.forEach(entry -> assertTrue(actual.contains(entry))); }
@Test public void map_byRef() { List<Integer> input = sequence(itemCount); putToBatchSrcMap(input); IMap<String, Integer> sinkMap = jet().getMap(sinkName); Sink<Entry<String, Integer>> sink = Sinks.map(sinkMap); p.readFrom(Sources.<String, Integer>map(srcName)).writeTo(sink); execute(); List<Entry<String, Integer>> expected = input.stream() .map(i -> entry(String.valueOf(i), i)) .collect(toList()); Set<Entry<String, Integer>> actual = sinkMap.entrySet(); assertEquals(expected.size(), actual.size()); expected.forEach(entry -> assertTrue(actual.contains(entry))); }
@Test public void map_withToKeyValueFunctions() { BatchStage<Integer> sourceStage = p.readFrom(TestSources.items(0, 1, 2, 3, 4)); sourceStage.writeTo(Sinks.map(sinkName, t -> t, Object::toString)); execute(); IMap<Integer, String> sinkMap = jet().getMap(sinkName); assertEquals(5, sinkMap.size()); IntStream.range(0, 5).forEach(i -> assertEquals(String.valueOf(i), sinkMap.get(i))); }