Stream APIの拡張(2) - 委譲方式による拡張

前回の記事では、Stream APIを拡張する方法として、特殊な性質を持つStreamを生成する方法を紹介した。
今回は、委譲を使ってStreamを拡張する方法を紹介する。

委譲によるStream APIの拡張

委譲を使ってStreamを拡張する場合、元のStreamを内部に保持するStreamのラッパークラスを用意する。
リクエストはすべてラッパーが受け付け、Stream標準の操作は元のStreamに委譲し、拡張操作はStreamのラッパー内で処理するようにする。

Streamのラッパークラス

具体的なコードで説明しよう。

ラッパーのStreamWrapperクラスでは、コンストラクタの引数で渡されたStreamのインスタンスを内部に保持しておく。
Streamが標準で持つ中間操作が呼ばれた場合、実際の処理はStreamインスタンスに委譲し、戻り値には新しく生成したStreamWrapperを返す。

public class StreamWrapper<T> {
    // 元になるストリーム
    private final Stream<T> source;
    // ファクトリメソッド
    public static <T> StreamWrapper<T> of(Stream<T> base) {
        return new StreamWrapper(base);
    }
    private StreamWrapper(Stream<T> source) {
        this.source = source;
    }
    // 中間操作の委譲例
    public <R> StreamWrapper<R> map(Function<T,R> f ) {
       return of(source.map(f));
    }
    public StreamWrapper<T> filter(Predicate<T> p) {
        return of(source.filter(p));
    }
    
    // 終端操作の委譲例
    public <A,R> R  collect(Collector<T,A,R> collector) {
        return source.collect(collector);
    }

takeWhileやdropWhileのような拡張操作は、次のようにStreamWrapper上に独自メソッドとして定義する。
（takeWhileとdropWhileについては、Stream APIの拡張(1) - 生成処理の拡張を参照のこと）。

    // 拡張したメソッド
    public StreamWrapper<T> takeWhile(Predicate<T> p) {
        
        Spliterator<T> splt = source.spliterator();
        Iterator<T> base = Spliterators.iterator(splt);
        
        return makeStream(splt.characteristics(), new Iterator<T>(){
            private boolean end;
            private T next;
            @Override
            public boolean hasNext() {
                if (!base.hasNext() || end) {
                    return false;
                }
                next = base.next();
                if (p.test(next)) {
                    return true;
                } else {
                    end = true;
                    return false;
                }
           }
            @Override
            public T next() {
                return next;
            }
        });
    }
    
    public StreamWrapper<T> dropWhile(Predicate<T> p) {
        
        Spliterator<T> splt = source.spliterator();
        Iterator<T> base = Spliterators.iterator(splt);
        
        return makeStream(splt.characteristics(), new Iterator<T>(){
            private boolean start;
            private T next;
            @Override
            public boolean hasNext() {
                if (!base.hasNext()) {
                    return false;
                }
                
                next = base.next();
                // 述語が成立しなくなるまで、要素を読み飛ばす。
                if (!start) {
                    while(p.test(next) && base.hasNext()) {
                        next = base.next();
                    }
                    if (base.hasNext()) {
                        start = true;
                    } else {
                        return false;
                    }
                }
                return true;
           }
            @Override
            public T next() {
                return next;
            }
        });
    }
    
    // 共通メソッド
    private static <T> StreamWrapper<T> makeStream(
                int characteristics, Iterator<T> itr) {

        return of(StreamSupport.stream(
                Spliterators.spliteratorUnknownSize(
                        itr,characteristics), false));
    }

zipをStreamWrapperに定義すると、次のようになる。
（zipについては、Java8 Stream APIの基本(5) - zip の実装を参照のこと）。

    /** 2つのストリームの要素をペアとしたストリームを生成。 */
    public <U> StreamWrapper<Pair<T,U>> zip(Stream<U> other) {
        return zip(other, Pair::new);
    }
    /** 2つのストリームの要素に、任意の関数を適用した結果のストリームを生成する。 */
    public <U,R> StreamWrapper<R> zip(Stream<U> other, BiFunction<T,U,R> f) {
        Iterator<T> base1 = source.iterator();
        Iterator<U> base2 = other.iterator();
        return makeStream(Spliterator.IMMUTABLE, new Iterator<R>() {
            @Override
            public boolean hasNext() {
                return base1.hasNext() && base2.hasNext();
            }
            @Override
            public R next() {
                return f.apply(base1.next(), base2.next());
            }
        });
    }
    /** 0オリジンのインデックスを付与する */
    public StreamWrapper<Pair<T,Integer>> zipWithIndex() {
        return zip(Stream.iterate(0, n -> n + 1));
    }
    /** ペアをあらわすクラス */
    public static class Pair<T,U> {
        public final T _1;
        public final U _2;
        public Pair(T t, U u) {
            this._1 = t;
            this._2 = u;
        }
        public T _1(){
            return _1;
        }
        public U _2(){
            return _2;
        }
    }
}

呼び出し側のコード

takeWhileとdropWhileを使ったコードは次のようになる。

List<Integer> list =
    StreamWrapper.of(Stream.of("-2","-1","1","2","3","-1","4"))
        .map(Integer::parseInt) // -2,-1,1,2,3,-1,4
        .dropWhile(n -> n <= 0) // -2,-1をスキップ
        .takeWhile(n -> n >= 0) // 1,2,3 まで取得
        .map(n -> n * 2) // 2倍
        .collect(Collectors.toList());
System.out.println(list); // 2,4,6

zipを利用するコードは次の通り。

// 2つの整数のリストから正の整数同士のみを掛け合わせる。
List<Integer> list2 = 
    StreamWrapper.of(Stream.of(-3, 4, -5, 6))
        .zip(Stream.of(-8, -7, 6, 5))
        .filter(p -> p._1 > 0 && p._2 > 0)
        .map(p -> p._1 * p._2)
        .collect(Collectors.toList());
System.out.println(list2); // 30

// 整数リストの偶数番目の値の平均値を取得する。
double avg = StreamWrapper.of(Stream.of(1,2,3,4,5,6))
        .zipWithIndex()
        // ペアの2番目の要素は0オリジンのインデックス
        .filter(p -> p._2 % 2 == 1) // 2,4,6を抽出。
        .collect(Collectors.averagingInt(Pair::_1));
System.out.println(avg); // 4.0

このように、StreamWrapperを利用することで、Streamの標準機能と拡張機能を区別せず扱えるようになる。

委譲方式のメリットとデメリット

委譲方式を採用した場合、Streamが標準で備える中間操作と、拡張した中間操作をメソッドチェーンで任意に連結して呼び出すことができる。

デメリットは、元のStreamへの委譲メソッドを書かなければならないことだ。
特に、次のようにStreamWrapperでStreamインタフェースを実装して、Decorator構造にした場合には、大量の委譲メソッドを書かなければならなくなる。

public class StreamWrapper<T> implements Stream<T> {

まとめ

前回の記事と合わせて、Stream APIを拡張する2つの方法を紹介した。
takeWhileやdropWhile, zipといった、多くの関数型言語で標準的に用意されている機能は、残念ながらJava8では提供されていない。
これらの仕組みは集合データを操作する関数型プログラミングの機能として一般的なもののため、近い将来のバージョンアップで是非対応してもらいたいものである。

[前多 賢太郎]