【メモ】Google MµGの使い方

Java 8向けのユーティリティーライブラリー、[Google MµG](https://github.com/google/mug)ライブラリーの使い方メモ。

* 読み方は『マグ』でよいのだろうか...?
* なお、Googleの公式ライブラリーではない。
* [https://github.com/google/mug](https://github.com/google/mug)
* [JavaDoc](https://google.github.io/mug/apidocs/index.html)

# 依存関係

```gradle
dependencies {
    compile 'com.google.mug:mug:1.12'
}
```


# `BiStream`

キーと値を持つ`Stream`。


## `BiStream`を作る

`BiStream`の静的メソッドを使う。

#### `biStream()`

```java
var input = Stream.of( "foo", "bar", "buz" );
var result = BiStream.biStream( input )
                     .mapValues( e -> e.toUpperCase() )
                     .toMap();
assertThat( result ).containsExactly( "foo", "FOO", "bar", "BAR", "buz", "BUZ" );
```

通常の`Stream`をラップする。何に使うのかと思ったら、`mapKeys()`のようなオペレーションを後続で呼び出すためらしい。


```java
var input = Stream.of( "foo", "bar", "buz" );
var result = BiStream.biStream( input, Function.identity(), e -> e.toUpperCase() )
                     .toMap();
assertThat( result ).containsExactly( "foo", "FOO", "bar", "BAR", "buz", "BUZ" );
```

マッピング関数を引数にとるバージョン。


#### `from()`

```java
var input = Map.of( "foo", "123", "bar", "456", "buz", "789" );
var result = BiStream.from( input )
                     .map( ( i, v ) -> i + ":" + v )
                     .collect( toList() );
assertThat( result ).containsExactly( "foo:123", "bar:456", "buz:789" );
```

マップのキーと値をペアとして、`BiStream`を作成する。


#### `indexed()`

```java
var input = Stream.of( "foo", "bar", "buz" );
var result = BiStream.indexed( input )
                     .map( ( i, v ) -> i + ":" + v )
                     .collect( toList() );
assertThat( result ).containsExactly( "0:foo", "1:bar", "2:buz" );
```

0始まりのインデックスを付加する。たいていのプログラミング言語の標準APIには組み込まれているので、これは嬉しい。
なお戻り値の型は`BiStream<Integer, V>`。流石にプリミティブ型はサポートされない。


#### `neighbors()`

```java
var input = Stream.of( "foo", "bar", "buz" );
var result = BiStream.neighbors( input )
                     .map( ( i, v ) -> i + ":" + v )
                     .collect( toList() );
assertThat( result ).containsExactly( "foo:bar", "bar:buz" );
```

隣り合った二つの値をペアにする。


#### `of()`

```java
var result = BiStream.of( "foo", "123", "bar", "456" )
                     .map( ( i, v ) -> i + ":" + v )
                     .collect( toList() );
assertThat( result ).containsExactly( "foo:123", "bar:456" );
```

JDK標準の`of()`と同様。


#### `zip()`

```java
var input1 = Stream.of( "foo", "bar", "buz" );
var input2 = Stream.of( "hoge", "piyo", "fuga" );
var result = BiStream.zip( input1, input2 )
                     .map( ( t, u ) -> t + ":" + u )
                     .collect( toList() );
assertThat( result ).containsExactly( "foo:hoge", "bar:piyo", "buz:fuga" );
```

二つの`Stream`を合体させる。これ、いつも自作していた。正直、標準ライブラリーに追加してほしいレベル。
ただし、`Stream`のサイズが異なると、少ないほうに合わせて余りが無視されるので注意。

```java
var input1 = Stream.of( "foo", "bar", "buz" );
var input2 = Stream.of( "hoge", "piyo" );
var result = BiStream.zip( input1, input2 )
                     .map( ( t, u ) -> t + ":" + u )
                     .collect( toList() );

assertThat( result ).containsExactly( "foo:hoge", "bar:piyo" );
```


## `BiStream`を使う

* 通常の`Stream`の操作に加え、`filterKeys()`、`mapValues()`のようなキーや値のためのオペレーションが追加されている。


#### `append()`

```java
var result = BiStream.of( "foo", "123" ).append( "bar", "456" )
                     .map( ( t, u ) -> t + ":" + u )
                     .collect( toList() );
assertThat( result ).containsExactly( "foo:123", "bar:456" );
```

後続に値を追加する。


#### `inverse()`

```java
var result = BiStream.of( "foo", "123", "bar", "456" )
                     .inverse()
                     .map( ( t, u ) -> t + ":" + u )
                     .collect( toList() );
assertThat( result ).containsExactly( "123:foo", "456:bar" );
```

キーと値を交換する。標準`Stream`にも追加してほしい……。


# `MoreStream`

JDKやGuavaにない拡張を提供する。

#### `generate()`

```java
var result = MoreStreams.generate( 1, n -> n >= 9 ? Stream.empty() : Stream.of( n + 1 ) )
                        .collect( toList() );
assertThat( result ).containsExactly( 1, 2, 3, 4, 5, 6, 7, 8, 9 );
```

*有限の*ストリームを生成するためのメソッド。空のストリームが返された時点で、生成を停止する。
使い道はあるのだろうか。


#### `flatten()`

```java
var input = Stream.of(
        Stream.of( "foo", "bar", "buz" ),
        Stream.of( "hoge", "piyo", "fuga" )
);
var result = MoreStreams.flatten( input )
                        .collect( toList() );
assertThat( result ).containsExactly( "foo", "bar", "buz", "hoge", "piyo", "fuga" );
```

`flatMap`で同様の処理は実現できるはずなのだが、Javaのバグで無限ストリームを扱えないためこのメソッドがある。[JDK 10で修正された模様。](https://bugs.java.com/bugdatabase/view_bug.do?bug_id=8075939)


#### `dice()`

```java
var input = Stream.of( "foo", "bar", "buz", "hoge", "piyo" );
var result = MoreStreams.dice( input, 2 )
                        .collect( toList() );
assertThat( result.size() ).isEqualTo( 3 );
var list1 = result.get( 0 );
assertThat( list1 ).containsExactly( "foo", "bar" );
var list2 = result.get( 1 );
assertThat( list2 ).containsExactly( "buz", "hoge" );
var list3 = result.get( 2 );
assertThat( list3 ).containsExactly( "piyo" );
```

`Stream`を指定サイズごとにリストにして、新たな`Stream<List>`にまとめる。
順次ストリームの場合、ストリームの最後以外は指定サイズのリストが返されることが保証される。並列ストリームの場合、保証されない。

ところでこういう操作を`dice`と呼ぶのを初めて知りました。`window`とどう違うのだろう。


#### `iterateOnce()`

```java
var input = Stream.of( "foo", "bar", "buz" );
for ( var e : MoreStreams.iterateOnce( input ) ) {
    assertThat( e ).isAnyOf( "foo", "bar", "buz" );
}
```

ストリームを`for`文でイテレートするためのメソッド。多分一生使うことはない。


#### `iterateThrough()`

```java
expect(() -> {
    MoreStreams.iterateThrough( Stream.of( "foo" ), e -> {
        throw new Exception( "Checked Exception!" );
    });
}).throwsException( e -> {
    assertThat( e ).hasMessageThat().isEqualTo( "Checked Exception!" );
});
```

基本的に`forEach()`と同じだが、チェック例外を投げることができるという違いがある。


# `Retryer`

失敗する可能性のある処理をリトライするための処理。便利かと思いきや、大抵リトライが必要な処理ではそもそもリトライ機能が提供されている場合が多いので、意外と使いどころがない。

1. `Retryer`のインスタンスを作る。
2. `upon()`、`ifReturns()`などで、リトライの条件を指定する。
3. `Delay`で実行間隔を指定。大抵の場合、倍々でリトライ間隔を増やす`exponentialBackoff()`を使えばよい。
4. `retryBlockingly()`、`retry()`、`retryAsync()`でリトライする処理を指定。


#### 同期

```java
var result = new Retryer()
        .upon( RuntimeException.class, Retryer.Delay.ofMillis( 100 ).exponentialBackoff( 2, 2 ) )
        .retryBlockingly( this::mayFail );
assertThat( result ).isEqualTo( "success" );
```


#### 非同期

```java
CompletionStage<String> future = new Retryer()
        .upon( RuntimeException.class, Retryer.Delay.ofMillis( 100 ).exponentialBackoff( 2, 2 ) )
        .retry( this::mayFail, Executors.newSingleThreadScheduledExecutor() );
future.thenAccept( result -> {
    assertThat( result ).isEqualTo( "success" );
});
```

正直非同期APIは微妙。


# Maybe

例外が発生する可能性のある処理をラップするためのクラス。`java.util.Optional`に例外を扱う機能が付いたバージョンのようなもの。Scalaにおける`Try`が比較的近い?

主な目的は、ストリームにおけるチェック例外を容易に扱うことらしい。


#### `maybe()`

`Maybe`インスタンスを作成する基本的なメソッド。各種関数型インターフェースを受け付ける。

```java
var result = IntStream.range( 1, 10 ).boxed()
                      .map( Maybe.maybe( e -> {
                          if ( e <= 5 ) {
                              return e;
                          } else {
                              throw new Exception();
                          }
                      } ) )
                      .map( m -> m.orElse( e -> Integer.valueOf( 0 ) ) )
                      .collect( toList() );
assertThat( result ).containsExactly( 1, 2, 3, 4, 5, 0, 0, 0, 0 );
```

`orElse()`で例外が発生した値を、`0`に変換している。

```java
expect(() -> {
    Maybe.maybe( () -> { throw new RuntimeException( "Unchecked!" );} );
}).throwsException( e -> {
    assertThat( e ).hasMessageThat().isEqualTo( "Unchecked!" );
});
```

非チェック例外はそのまま再スローされる。注意。一応ストリーム以外でも使えないわけではないが、この制限があるので、専用のライブラリーを使ったほうがいい気がする。


#### `byValue()`

```java
var result = IntStream.range( 1, 10 ).boxed()
                      .map( Maybe.maybe( e -> {
                          if ( e <= 5 ) {
                              return e;
                          } else {
                              throw new Exception();
                          }
                      } ) )
                      .filter( Maybe.byValue( n -> n % 2 == 0 ) )
                      .map( m -> m.orElse( e -> Integer.valueOf( 0 ) ) )
                      .collect( toList() );
assertThat( result ).containsExactly( 2, 4, 0, 0, 0, 0 );
```

`filter()`が`Maybe`を受け取るようになってしまうため、アンラップするためのメソッドが用意されている。見てのとおり、例外が発生したものはそのまま後続の処理に移る。

![1.PNG](https://qiita-image-store.s3.amazonaws.com/0/231669/3f00a085-77bc-931e-206c-c55257daf800.png)
![2.PNG](https://qiita-image-store.s3.amazonaws.com/0/231669/15b93ca6-36ed-5656-cc9c-4083de0b7467.png)


#### `catching()`

```java
var result = IntStream.range( 1, 10 ).boxed()
                      .map( Maybe.maybe( e -> {
                          if ( e <= 5 ) {
                              return e;
                          } else {
                              throw new Exception();
                          }
                      } ) )
                      .flatMap( m -> m.catching( e -> {} ) )
                      .collect( toList() );
assertThat( result ).containsExactly( 1, 2, 3, 4, 5 );
```

単に値を無視したいときに使う。


# `Funnel`

使いどころはかなりニッチだがハマれば便利そう。

## ユースケース

* ある一連の値(`Iterable`、`Stream`)を、別の値に変換したい
* 一部の値に対しては、まとめて(Batch)処理を行いたい
* 出力の順番を、入力の順番と同じにしたい(FIFO)


## `Funnel`

```java
var funnel = new Funnel<String>();
var batch = funnel.<String>through( data -> {
    assertThat( data ).containsExactly( "batch1", "batch2" );
    return data.stream().map( String::toUpperCase ).collect( toList() );
} );

var input = List.of( "local1", "batch1", "local2", "batch2" );
for ( var i : input ) {
    if ( i.startsWith( "local" ) ) {
        funnel.add( i );
    } else {
        batch.accept( i );
    }
}
var result = funnel.run();

assertThat( result ).containsExactly( "local1", "BATCH1", "local2", "BATCH2" );
```

1. まず、`Funnel`をインスタンス化する
2. `through()`メソッドで、バッチ変換の処理を指定する。
3. 個別に処理する値は、`Funnel.add()`を呼び出す。
4. まとめて処理する値は、`Funnel.Batch.accept()`に渡す。
5. `Funnel.run()`で実行。

なお、`accept()`は結果の変換などを行う関数を引数にとるオーバーロードがある。

```java
var funnel = new Funnel<String>();
var batch = funnel.<String>through( data -> List.of() );
batch.accept( "batch1" );
expect(() -> {
    funnel.run();
}).throwsException( e -> {
    assertThat( e ).isInstanceOf( IllegalStateException.class );
});
```

当たり前だが、バッチの戻りリストのサイズが入力と異なる場合、エラーになる。

```java
var funnel = new Funnel<String>();
var batch = funnel.<String>through( data -> List.of() );
batch.accept( "batch1" );

expect(() -> {
    funnel.run();
}).throwsException( e -> {
    assertThat( e ).isInstanceOf( IllegalStateException.class );
});
```

また、順番はバッチ処理の順番に依存する。当該処理が戻り順を入力順と同じであることを保証している必要がある。

```java
var funnel = new Funnel<String>();
var batch = funnel.<String>through( data -> {
    var newList = new ArrayList<String>( data );
    Collections.reverse( newList );
    return newList;
} );
batch.accept( "batch1" );
batch.accept( "batch2" );
var result = funnel.run();

assertThat( result ).containsExactly( "batch2", "batch1" );
```


# `Parallelizer`

名前の通り、処理を並列化する。

```java
var input = Stream.of( "foo", "bar", "buz" );
var result = Collections.synchronizedList( new ArrayList<String>() );
new Parallelizer( executor, 3 )
        .parallelize( input, result::add );
assertThat( result ).containsExactly( "foo", "bar", "buz" );
```

1. `Parallelizer`をインスタンス化する。引数は、処理を実行する`ExecutorService`と、同時実行可能なタスク数。基本的に`ExecutorService`のスレッド数と同じ?
2. `parallelize()`もしくは`parallelizeUninterruptibly()`を呼ぶ。入力データとなる`Stream`もしくは`Iterator`を第一引数、行う処理を第二引数にとる。

処理は終了するまでブロックされる。途中で例外が発生した場合、後続処理は中断される。

READMEによれば、

> Parallel streams are for CPU-bound tasks. JDK has built-in magic to optimally use the available cores.
> Parallelizer is for IO-bound tasks.

とあるので、ファイルの入出力、外部APIの呼び出し等に使用されるべきらしい。


# まとめ

* `BiStream`、`MoreStreams`は便利
* それ以外はかなりニッチ
* [サンプルソース](https://bitbucket.org/minebreaker_tf/teachyourselfmug)