functional-reactive
Made in the European Union

Streams meet functions

Stop inlining lambdas into every Stream pipeline. Treat filters and pipelines as values you can name, reuse and compose.

May 17, 2026

A typical first attempt at filtering a Stream looks like this:

Stream.of(
    new Car("BMW", BLUE,  200, 2017, 10000f),
    new Car("BMW", GREEN, 215, 2017, 10432f),
    new Car("VW",  BLUE,  180, 2017, 10000f),
    new Car("VW",  WHITE, 220, 2017, 10000f))
    .filter(c -> c.brand().equals("BMW"))
    .forEach(System.out::println);

Stream.of(/* same six cars again */)
    .filter(c -> c.brand().equals("VW"))
    .forEach(System.out::println);

The data is recreated on every pipeline. The filter is duplicated. The “VW” version is identical to the “BMW” version save for one literal. We’ll refactor this in four passes, ending with a fully reusable pipeline that you can describe before you have a stream at all.

Pass 1: separate data from stream

A Stream can be consumed only once. If you want to run two pipelines, you need two streams from the same source. So expose the data once and a factory for fresh streams:

public static final List<Car> cars = List.of(
    new Car("BMW", BLUE,  200, 2017, 10000f),
    new Car("BMW", GREEN, 215, 2017, 10432f),
    new Car("BMW", UNDEFINED, 200, 2017, 10000f),
    new Car("VW",  BLUE,  180, 2017, 10000f),
    new Car("VW",  WHITE, 220, 2017, 10000f),
    new Car("VW",  RED,   200, 2017, 10000f)
);

public static Stream<Car> cars() { return cars.stream(); }

cars().filter(c -> c.brand().equals("BMW")).forEach(System.out::println);
cars().filter(c -> c.brand().equals("VW")).forEach(System.out::println);

Better. Two pipelines, one source of truth, but the filters are still hand-rolled.

Pass 2: lift the predicate into a function

The two filters differ only in the brand string. Lift that variation into a function:

Function<String, Predicate<Car>> brandFilter =
    brand -> car -> car.brand().equals(brand);

cars().filter(brandFilter.apply("BMW")).forEach(System.out::println);
cars().filter(brandFilter.apply("VW")).forEach(System.out::println);

brandFilter is one value that produces any number of brand-specific predicates. Three more brands? Three more apply calls — no new lambdas.

Pass 3: describe the filter before you have a stream

Once you can build predicates as values, you can describe the entire filter stage of a pipeline without ever touching a Stream. functional-reactive ships StreamFunctions.streamFilter() for exactly this:

import com.svenruppert.functional.StreamFunctions;

// Function<Predicate<T>, Function<Stream<T>, Stream<T>>>
var streamFilter = StreamFunctions.<Car>streamFilter();

streamFilter.apply(brandFilter.apply("BMW"))
            .apply(cars())
            .forEach(System.out::println);

The signature reads dense, but the idea is small: feed it a predicate, it gives you back a stage that turns one Stream<Car> into another. The stream itself shows up only at the end.

andThen makes this fully fluent:

brandFilter
    .andThen(streamFilter)
    .apply("BMW")
    .apply(cars())
    .forEach(System.out::println);

Now brandFilter.andThen(streamFilter) is a Function<String, Function<Stream<Car>, Stream<Car>>> — a “brand-keyed pipeline factory”. A test, a UI component, a CLI handler can each call it with their own brand string and own stream and get a working pipeline.

Pass 4: bridge errors with Result.stream()

Real pipelines have steps that can fail — parsing, lookups, I/O. Those steps return Result<T>, which has a free bridge into Stream:

Stream.of("1", "2", "Hi", "3")
      .map((CheckedFunction<String, Integer>) Integer::valueOf)   // Stream<Result<Integer>>
      .flatMap(Result::stream)                                    // Stream<Integer>
      .reduce(Integer::sum)
      .ifPresent(System.out::println);                            // 6

Result::stream emits zero or one element — so failed parses ("Hi") silently drop out, and the rest of the pipeline keeps running. No try/catch, no sentinel -1, no orphaned null.

When this approach pays off

It’s overkill for one-off filters. It earns its keep when:

  • the same filter shows up in three or more places (registries, UIs, tests),
  • you want a pipeline you can hand to someone else who supplies the data,
  • you need to combine filters at runtime (“all of these, in this order”),
  • failure modes are part of the pipeline, not an exception thrown out of it.

Recap

  • Treat the data source and the stream factory as separate things.
  • Promote predicates from inline lambdas to Function<X, Predicate<T>> values.
  • StreamFunctions.streamFilter() lets you describe a stage that consumes a stream, before the stream exists.
  • Result::stream is the cleanest way to drop failed elements out of a pipeline.

Next

Edit on GitHub →