@Test public void shouldNotAllowValueTransformerToCallInternalProcessorContextMethods() { final KStreamTransformValues<Integer, Integer, Integer> transformValue = new KStreamTransformValues<>(new ValueTransformerSupplier<Integer, Integer>() { @Override public ValueTransformer<Integer, Integer> get() { return new BadValueTransformer(); } }); final Processor transformValueProcessor = transformValue.get(); transformValueProcessor.init(null); try { transformValueProcessor.process(null, 0); fail("should not allow call to context.forward() within ValueTransformer"); } catch (final StreamsException e) { } try { transformValueProcessor.process(null, 1); fail("should not allow call to context.forward() within ValueTransformer"); } catch (final StreamsException e) { } try { transformValueProcessor.process(null, 2); fail("should not allow call to context.forward() within ValueTransformer"); } catch (final StreamsException e) { } try { transformValueProcessor.punctuate(0); fail("should not allow ValueTransformer#puntuate() to return not-null value"); } catch (final StreamsException e) { } }
@Override public Processor<K, V> get() { return new KStreamTransformValuesProcessor<>(valueTransformerSupplier.get()); }
KStreamTransformValues implements ProcessorSupplier<K, V> { @Override public Processor<K, V> get() { return new KStreamTransformValuesProcessor<>(valueTransformerSupplier.get()); } }
KStreamTransformValues implements ProcessorSupplier<K, V> { @Override public Processor<K, V> get() { return new KStreamTransformValuesProcessor<>(valueTransformerSupplier.get()); } KStreamTransformValues(ValueTransformerSupplier<V, R> valueTransformerSupplier); }
KStreamTransformValues implements ProcessorSupplier<K, V> { @Override public Processor<K, V> get() { return new KStreamTransformValuesProcessor<>(valueTransformerSupplier.get()); } KStreamTransformValues(ValueTransformerSupplier<V, R> valueTransformerSupplier); @Override Processor<K, V> get(); }
KStreamTransformValues implements ProcessorSupplier<K, V> { @Override public Processor<K, V> get() { return new KStreamTransformValuesProcessor<>(valueTransformerSupplier.get()); } KStreamTransformValues(ValueTransformerSupplier<V, R> valueTransformerSupplier); @Override Processor<K, V> get();  }