public static <InputT, OutputT> FlatMapElements<InputT, OutputT> via(
      InferableFunction<? super InputT, ? extends Iterable<OutputT>> fn) {
    TypeDescriptor<OutputT> outputType =
        TypeDescriptors.extractFromTypeParameters(
            (TypeDescriptor<Iterable<OutputT>>) fn.getOutputTypeDescriptor(),
            Iterable.class,
            new TypeDescriptors.TypeVariableExtractor<Iterable<OutputT>, OutputT>() {});
    TypeDescriptor<InputT> inputType = (TypeDescriptor<InputT>) fn.getInputTypeDescriptor();
    return new FlatMapElements<>(fn, inputType, outputType);
  }
@Test
  @Category(NeedsRunner.class)
  public void testFlatMapSimpleFunction() throws Exception {
    PCollection<Integer> output =
        pipeline
            .apply(Create.of(1, 2, 3))

            // Note that FlatMapElements takes an InferableFunction<InputT, ? extends
            // Iterable<OutputT>>
            // so the use of List<Integer> here (as opposed to Iterable<Integer>) deliberately
            // exercises
            // the use of an upper bound.
            .apply(
                FlatMapElements.via(
                    new SimpleFunction<Integer, List<Integer>>() {
                      @Override
                      public List<Integer> apply(Integer input) {
                        return ImmutableList.of(-input, input);
                      }
                    }));

    PAssert.that(output).containsInAnyOrder(1, -2, -1, -3, 2, 3);
    pipeline.run();
  }