<T> Stream.fromForEach(Consumer<Consumer<T>>)

[Peter Levart]

Hi,

On 23/05/2022 14:53, Brian Goetz wrote:
> ...
>
> Are there any useful Consumer<Consumer<T>> in the wild, that are 
> actually typed like that?  I doubt it (there are many things 
> convertible to it, though.)  Which suggests it might be fruitful to 
> define a FI:
>
>     interface PushSource<T> {
>         void accept(Consumer<? extends Consumer<T>> pusher);
>     }
>
>     static<T> Stream<T> fromPush(PushSource<T> source) { ... }
>
> and Iterable::forEachRemaining and Optional::ifPresent will convert to 
> it.
>
>
...the PushSource would probably be just the following (a replacement 
for Consumer<Consumer<T>>):

     interface PushSource<T> {
         void accept(Consumer<T> consumer);
     }

The outer Consumer is not that problematic, IMO, since it is subject of 
conversion from any similar signature via lambdas/method references as 
easily as PushSource. The inner Consumer type is more problematic. For 
example, suppose one would have the following API:

     interface Listener {
         void listen(String token);
     }

     record Parser(String text) {
         void parse(Listener listener) {
             for (var token : text.split(" ")) {
                 listener.listen(token);
             }
         }
     }


...to convert a Parser instance `parser` to a Stream of tokens, one 
would have to do something like this:

var parser = new Parser("1 2 3 4");

var stream = Stream.fromForEach(consumer -> parser.parse(consumer::accept));

...and then the type of stream would be inferred as Stream<Object>, not 
Stream<String> as one would like. Type hinting would be necessary:

var stream = Stream.<String>fromForEach(consumer -> 
parser.parse(consumer::accept));


I don't believe Java type system allows doing such things more elegantly.

One point to highlight is also that such method is only suitable for 
"finite" generators. A generator that emits infinite number of elements 
would have no way to stop emitting them although the resulting Stream 
was later shortened with .limit(max) or .takeWhile(predicate), ... Such 
Stream would never stop. This could be solved by a warning in the 
javadoc though. Or by something that project Loom could provide in the 
form of continuation?

As to the name, `generate` is taken by the generate(Supplier<T> 
supplier) form, but overload is technically possible, since it differs 
by parameters of unrelated type and number of parameters of the 
@FunctionalInterface methods (Supplier vs. Consumer). No conflicts 
should be expected.


Regards, Peter