GafferMapFunction implements FlatMapFunction<T, Element> { @Override public void flatMap(final T item, final Collector<Element> out) throws Exception { if (null == out) { throw new IllegalArgumentException("Element collector is required"); } if (null == elementGenerator) { elementGenerator = generatorClassName.newInstance(); } if (elementGenerator instanceof OneToOneElementGenerator) { out.collect(((OneToOneElementGenerator<T>) elementGenerator)._apply(item)); } else if (elementGenerator instanceof OneToManyElementGenerator) { final Iterable<Element> elements = ((OneToManyElementGenerator<T>) elementGenerator)._apply(item); if (null != elements) { elements.forEach(out::collect); } } else { final Iterable<? extends Element> elements = (elementGenerator).apply(Collections.singleton(item)); if (null != elements) { elements.forEach(out::collect); } } } GafferMapFunction();  GafferMapFunction(final Class<T> consumeAs, final Class<? extends Function<Iterable<? extends T>, Iterable<? extends Element>>> generatorClassName); void setConsumeAs(final Class<T> consumeAs); @Override void flatMap(final T item, final Collector<Element> out); DeserializationSchema<T> getSerialisationType();  }
@Test public void shouldDelegateToGafferElementGenerator() throws Exception { final String csv = "1,2,3,4"; final GafferMapFunction function = new GafferMapFunction<>(String.class, MockedGenerator.class); final Iterable expectedResults = Arrays.asList( mock(Element.class), mock(Element.class) ); final Collector<Element> collector = mock(Collector.class); given(mockGenerator.apply(Collections.singleton(csv))).willReturn(expectedResults); function.flatMap(csv, collector); for (Element expectedResult : (Iterable<Element>) expectedResults) { verify(collector).collect(expectedResult); } }
@Test public void shouldDelegateToGafferOneToOneElementGenerator() throws Exception { final String csv = "1,2,3,4"; final GafferMapFunction function = new GafferMapFunction<>(String.class, MockedOneToOneGenerator.class); final Element expectedResult = mock(Element.class); final Collector<Element> collector = mock(Collector.class); given(mockOneToOneGenerator._apply(csv)).willReturn(expectedResult); function.flatMap(csv, collector); verify(collector).collect(expectedResult); }
@Test public void shouldDelegateToGafferOneToManyElementGenerator() throws Exception { final String csv = "1,2,3,4"; final GafferMapFunction function = new GafferMapFunction<>(String.class, MockedOneToManyGenerator.class); final Iterable expectedResults = Arrays.asList( mock(Element.class), mock(Element.class) ); final Collector<Element> collector = mock(Collector.class); given(mockOneToManyGenerator._apply(csv)).willReturn(expectedResults); function.flatMap(csv, collector); for (Element expectedResult : (Iterable<Element>) expectedResults) { verify(collector).collect(expectedResult); } }