Java Stream API にこういう機能があったらいいなあ的な妄想を書きなぐりました。

MyStream.java

package satomaru.sandbox.stream;

import java.util.Optional;
import java.util.function.BiFunction;
import java.util.stream.Stream;

/**
 * Java Stream API にこういう機能があったらいいなあ的な妄想を書きなぐりました。
 *
 * <p>
 * 可用性とか安全性とかそういう要件はとりえず考慮してないぜ Sorry。
 * </p>
 *
 * @param <P> 拡張 Stream から呼び出したい処理部品
 */
public interface MyStream<P> {

	/**
	 * 処理部品を呼び出すことが可能な、拡張 Stream。
	 *
	 * @param <P> 拡張 Stream から呼び出したい処理部品
	 * @param <T> 要素
	 */
	interface StreamEx<P, T> {
		/**
		 * 処理部品を使って値をマッピングします。
		 * 
		 * @param <R> マッピング後の要素
		 * @param mapper 処理部品を使って値をマッピングする関数
		 * @return マッピングされた後の拡張 Stream
		 */
		<R> StreamEx<P, R> map(BiFunction<P, T, R> mapper);

		// 本当はもっといっぱいメソッドがあるの……。

		/**
		 * 処理部品のインスタンスを確定して、通常の Stream を返却します。
		 * 
		 * @param processor 処理部品のインスタンス
		 * @return 処理部品が確定した後の、通常の Stream
		 */
		Stream<T> dependOn(P processor);
	}

	/**
	 * of のみ可能な実装です。
	 *
	 * @param <P> 拡張 Stream から呼び出したい処理部品
	 */
	class MyStreamImpl<P> implements MyStream<P> {
		private MyStreamImpl() {
		}

		/**
		 * {@inheritDoc}
		 */
		@Override
		public <T> StreamEx<P, T> of(@SuppressWarnings("unchecked") T... values) {
			return new StreamExImpl<>(Stream.of(values));
		}
	}

	/**
	 * 処理部品を呼び出すことが可能な、拡張 Stream の実装です。
	 *
	 * @param <P> 拡張 Stream から呼び出したい処理部品
	 * @param <T> 要素
	 */
	class StreamExImpl<P, T> implements StreamEx<P, T> {
		/** 処理を委譲する Stream。 */
		private final Stream<T> stream;

		/** このインスタンスに連結したインスタンス。 */
		private StreamExImpl<P, ?> chained;

		/** 処理部品のインスタンス。 */
		private Optional<P> processor = Optional.empty();

		/**
		 * コンストラクタ。
		 * 
		 * @param stream 処理を委譲する Stream
		 */
		private StreamExImpl(Stream<T> stream) {
			this.stream = stream;
		}

		/**
		 * {@inheritDoc}
		 */
		@Override
		public <R> StreamEx<P, R> map(BiFunction<P, T, R> mapper) {
			StreamExImpl<P, R> instance = new StreamExImpl<>(stream.map(t -> mapper.apply(getProcessor(), t)));
			chained = instance;
			return instance;
		}

		/**
		 * {@inheritDoc}
		 */
		@Override
		public Stream<T> dependOn(P processor) {
			this.processor = Optional.of(processor);
			return stream;
		}

		/**
		 * 拡張 Stream のチェーンを辿って、処理部品を取得します。
		 * 
		 * @return 処理部品
		 */
		private P getProcessor() {
			return processor.orElseGet(chained::getProcessor);
		}
	}

	/**
	 * 処理部品の型を指定して、拡張 Stream を作成します。
	 * 
	 * @param processorType 処理部品の型
	 * @return 拡張 Stream
	 */
	static <P> MyStream<P> with(Class<P> processorType) {
		return new MyStreamImpl<>();
	}

	/**
	 * 指定された要素で拡張 Stream を作成します。
	 * 
	 * @param <T> 要素
	 * @param values 要素のインスタンス
	 * @return 拡張 Stream
	 */
	<T> StreamEx<P, T> of(@SuppressWarnings("unchecked") T... values);
}

MyStreamTest.java

package satomaru.sandbox.stream;

import static org.junit.Assert.*;

import java.util.stream.Collectors;

import org.junit.Test;

public class MyStreamTest {

	public interface Processor {
		String bracketLeft(String string);
		String bracketRight(String string);
	}

	public static class MyProcessor implements Processor {
		@Override
		public String bracketLeft(String string) {
			return "[" + string;
		}

		@Override
		public String bracketRight(String string) {
			return string + "]";
		}
	}

	@Test
	public void test() {
		assertEquals("[FOO],[BAR]",
				MyStream.with(Processor.class).of("FOO", "BAR")
						.map(Processor::bracketLeft)
						.map(Processor::bracketRight)
						.dependOn(new MyProcessor())
						.collect(Collectors.joining(",")));
	}
}