Implementation of a pipe combinator system that supports carrying non-semantic info through pipeline elements.

Pipes.java

import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;

/**
 * Created by Lukas Härtel on 14.02.14.
 */
public class Pipes {
    /**
     * Interface describing a tuple
     *
     * @param <A> The type of the first item
     * @param <B> The type of the second item
     */
    public static interface Tuple<A, B> {
        public A getA();

        public B getB();
    }

    /**
     * Special tuple that can substitute the individual components and remain state
     *
     * @param <A> The type of the first item
     * @param <B> The type of the second item
     */
    public static interface Sub<A, B> extends Tuple<A, B> {
        public <C> Sub<C, B> subA(C a);

        public <C> Sub<A, C> subB(C b);

        public default <X, Y> Sub<X, Y> sub(X a, Y b) {
            return subA(a).subB(b);
        }
    }

    /**
     * Single is a special case implementation of a substitutable tuple. It does not carry any secondary item
     *
     * @param <A> The type of the value
     */
    public static class Single<A> implements Sub<A, Void> {
        public final A a;

        public Single(A a) {
            this.a = a;
        }

        @Override
        public <C> Sub<C, Void> subA(C a) {
            return new Single<>(a);
        }

        @Override
        public <C> Sub<A, C> subB(C b) {
            return new Item<>(a, b);
        }

        @Override
        public <X, Y> Sub<X, Y> sub(X a, Y b) {
            return new Item<>(a, b);
        }

        @Override
        public A getA() {
            return a;
        }

        @Override
        public Void getB() {
            return null;
        }
    }

    /**
     * The item is a basic implementation of the substitutable tuple
     *
     * @param <A> The type of the first item
     * @param <B> The type of the second item
     */
    public static class Item<A, B> implements Sub<A, B> {
        public final A a;

        public final B b;

        public Item(A a, B b) {
            this.a = a;
            this.b = b;
        }

        @Override
        public <C> Sub<C, B> subA(C a) {
            return new Item<>(a, b);
        }

        @Override
        public <C> Sub<A, C> subB(C b) {
            return new Item<>(a, b);
        }

        @Override
        public <X, Y> Sub<X, Y> sub(X a, Y b) {
            return new Item<>(a, b);
        }

        @Override
        public A getA() {
            return a;
        }

        @Override
        public B getB() {
            return b;
        }

        @Override
        public String toString() {
            return "(" + a + ", " + b + ")";
        }
    }

    /**
     * Abstract class for pipes that take an item of type A and produce an item of type B
     *
     * @param <A> The type of the input items
     * @param <B> The type of the output items
     */
    public static abstract class Pipe<A, B> implements Consumer<Sub<A, ?>> {
        public Consumer<Sub<B, ?>> consumer;

        @Override
        public void accept(Sub<A, ?> stream) {
            apply(stream.getA(), (b) -> consumer.accept(stream.subA(b)));
        }

        /**
         * Takes the item and passes a transformation to the consumer out
         *
         * @param item The source item
         * @param out  The output consumer
         */
        protected abstract void apply(A item, Consumer<B> out);
    }

    /**
     * Abstract class for pipes that take an element of type A with a carried element of type X and produce an element
     * of type B with carried element of type X
     *
     * @param <A> The type of the input element
     * @param <X> The type of the carried element
     * @param <B> The type of the output element
     */
    public static abstract class XPipe<A, X, B> implements Consumer<Sub<A, ?>> {
        public Consumer<Sub<B, ?>> consumer;

        public final Class<X> xClass;

        protected XPipe(Class<X> xClass) {
            this.xClass = xClass;
        }

        @Override
        public void accept(Sub<A, ?> stream) {
            apply(stream.getA(), xClass.cast(stream.getB()), b -> consumer.accept(stream.subA(b)));
        }

        protected abstract void apply(A a, X x, Consumer<B> out);
    }

    /**
     * Abstract class for pipes that take an element of type A with a carried element of type X and produce an element
     * of type B with carried element of type Y
     *
     * @param <A> The type of the input element
     * @param <X> The type of the input carried element
     * @param <B> The type of the output element
     * @param <Y> The type of the output carried element
     */
    public static abstract class XYPipe<A, X, B, Y> implements Consumer<Sub<A, ?>> {
        public Consumer<Sub<B, ?>> consumer;

        public final Class<X> xClass;

        protected XYPipe(Class<X> xClass) {
            this.xClass = xClass;
        }

        @Override
        public void accept(Sub<A, ?> stream) {
            apply(stream.getA(), xClass.cast(stream.getB()), (b, y) -> consumer.accept(stream.sub(b, y)));
        }

        protected abstract void apply(A a, X x, BiConsumer<B, Y> out);
    }

    /**
     * Filter is a pipe element that filters based on a predicate
     *
     * @param <A> The type of the elements to filter
     */
    public static class Filter<A> extends Pipe<A, A> {
        public final Predicate<A> filter;

        public Filter(Predicate<A> filter) {
            this.filter = filter;
        }

        @Override
        protected void apply(A item, Consumer<A> out) {
            if (filter.test(item)) {
                out.accept(item);
            }
        }
    }

    /**
     * Mapping is a pipe element that transforms based on a function
     *
     * @param <A> The type of the source elements
     * @param <B> The type of the target elements
     */
    public static class Mapping<A, B> extends Pipe<A, B> {
        public final Function<A, B> mapping;

        public Mapping(Function<A, B> mapping) {
            this.mapping = mapping;
        }

        @Override
        protected void apply(A item, Consumer<B> out) {
            out.accept(mapping.apply(item));
        }
    }

    /**
     * Swap is a pipe element that swaps the carrier lane with the element lane
     *
     * @param <A> The type of the element lane
     * @param <B> The type of the carrier lane
     */
    public static class Swap<A, B> extends XYPipe<A, B, B, A> {

        protected Swap(Class<B> bClass) {
            super(bClass);
        }

        @Override
        protected void apply(A a, B b, BiConsumer<B, A> out) {
            out.accept(b, a);
        }
    }

    /**
     * The induce pipe element combines the element and the carrier-lane
     *
     * @param <A> The type of the element lane
     * @param <X> The type of the carrier lane
     */
    public static class Induce<A, X> extends XPipe<A, X, Tuple<A, X>> {
        public Induce(Class<X> xClass) {
            super(xClass);
        }

        @Override
        protected void apply(A a, X x, Consumer<Tuple<A, X>> out) {
            out.accept(new Item<>(a, x));
        }
    }

    public static void main(String... args) {
        final Filter<String> startsWith2 = new Filter<>(x -> x.startsWith("2"));
        final Mapping<String, Integer> parse = new Mapping<>(Integer::valueOf);
        final Swap<Integer, String> swap = new Swap<>(String.class);
        final Filter<String> startsWithH = new Filter<>(x -> x.startsWith("H"));
        final Induce<String, Integer> induce = new Induce<>(Integer.class);

        // Element starts with 2 -> parse element as int -> swap element and carrier -> carrier starts with H -> induce
        startsWith2.consumer = parse;
        parse.consumer = swap;
        swap.consumer = startsWithH;
        startsWithH.consumer = induce;
        induce.consumer = System.out::println;

        startsWith2.accept(new Item<>("21", ""));
        startsWith2.accept(new Item<>("245", "Hallo"));
        startsWith2.accept(new Item<>("2467", "Hallo Welt"));
        startsWith2.accept(new Item<>("256", "Welt"));
        startsWith2.accept(new Item<>("356", "Mairier"));
    }
}