FlatMapperOperator implements Operator { @Override public void invoke ( final InvocationCtx ctx ) { for ( Tuple input : ctx.getInputTuplesByDefaultPort() ) { flatMapper.accept( input, outputTupleSupplier, output -> { output.attachTo( input ); ctx.output( output ); } ); } }  @Override SchedulingStrategy init( final InitCtx ctx ); @Override void invoke( final InvocationCtx ctx ); static final String FLAT_MAPPER_CONFIG_PARAMETER; }
@Test public void shouldFlatMapValues () { final int value = 5; final Tuple tuple = Tuple.of( "val", value ); input.add( tuple ); operator.invoke( invocationCtx ); final List<Tuple> outputTuples = output.getTuples( 0 ); assertEquals( 2, outputTuples.size() ); final Tuple output1 = outputTuples.get( 0 ); final Tuple output2 = outputTuples.get( 1 ); assertEquals( 1, output1.size() ); assertEquals( value + 1, output1.getIntegerValueOrDefault( "val", 0 ) ); assertEquals( 1, output2.size() ); assertEquals( value + 2, output2.getIntegerValueOrDefault( "val", 0 ) ); }