@Test public void testTransform() { KStreamBuilder builder = new KStreamBuilder(); TransformerSupplier<Number, Number, KeyValue<Integer, Integer>> transformerSupplier = new TransformerSupplier<Number, Number, KeyValue<Integer, Integer>>() { public Transformer<Number, Number, KeyValue<Integer, Integer>> get() { return new Transformer<Number, Number, KeyValue<Integer, Integer>>() { private int total = 0; @Override public void init(ProcessorContext context) { } @Override public KeyValue<Integer, Integer> transform(Number key, Number value) { total += value.intValue(); return KeyValue.pair(key.intValue() * 2, total); } @Override public KeyValue<Integer, Integer> punctuate(long timestamp) { return KeyValue.pair(-1, (int) timestamp); } @Override public void close() { } }; } }; final int[] expectedKeys = {1, 10, 100, 1000}; MockProcessorSupplier<Integer, Integer> processor = new MockProcessorSupplier<>(); KStream<Integer, Integer> stream = builder.stream(intSerde, intSerde, topicName); stream.transform(transformerSupplier).process(processor); driver = new KStreamTestDriver(builder); for (int expectedKey : expectedKeys) { driver.process(topicName, expectedKey, expectedKey * 10); } driver.punctuate(2); driver.punctuate(3); assertEquals(6, processor.processed.size()); String[] expected = {"2:10", "20:110", "200:1110", "2000:11110", "-1:2", "-1:3"}; for (int i = 0; i < expected.length; i++) { assertEquals(expected[i], processor.processed.get(i)); } }
@Override public Processor<K, V> get() { return new KStreamTransformProcessor<>(transformerSupplier.get()); }
KStreamTransform implements ProcessorSupplier<K, V> { @Override public Processor<K, V> get() { return new KStreamTransformProcessor<>(transformerSupplier.get()); } }
KStreamTransform implements ProcessorSupplier<K, V> { @Override public Processor<K, V> get() { return new KStreamTransformProcessor<>(transformerSupplier.get()); } KStreamTransform(TransformerSupplier<? super K, ? super V, ? extends KeyValue<? extends K1, ? extends V1>> transformerSupplier); }
KStreamTransform implements ProcessorSupplier<K, V> { @Override public Processor<K, V> get() { return new KStreamTransformProcessor<>(transformerSupplier.get()); } KStreamTransform(TransformerSupplier<? super K, ? super V, ? extends KeyValue<? extends K1, ? extends V1>> transformerSupplier); @Override Processor<K, V> get(); }
KStreamTransform implements ProcessorSupplier<K, V> { @Override public Processor<K, V> get() { return new KStreamTransformProcessor<>(transformerSupplier.get()); } KStreamTransform(TransformerSupplier<? super K, ? super V, ? extends KeyValue<? extends K1, ? extends V1>> transformerSupplier); @Override Processor<K, V> get();  }