Reactorのdeferの使い方

reactor-coreのMonoやFluxを使う上で、よくやってしまうリソース解放に関する誤りを、Mono#deferの使い方とともに説明します。
reactor-coreのその他の使い方については、Reactor 3 Reference Guidを参考にして下さい。

前準備

reactor-coreに従って、reactor-coreを使ったJavaプログラミングが出来る環境を準備します。

問題のあるコード

MonoやFluxの処理は、subscribeやblock等の終端処理が呼ばれた際に実行されます。このため、終端処理が呼ばれるたびに、doOnTerminate等のハンドラーが実行されます。以下のように、Monoの外側で作成したリソースをdoOnTerminateでcloseすると、2回目の終端処理の呼び出しはすでにcloseされているリソースに対して実行されてしまいます。

    private static class _SampleCloseable implements Closeable {

        private boolean _closed;

        @Override
        public synchronized void close() {
            // 2回以上、closeが呼ばれるとエラーになる。
            if (_closed) {
                throw new RuntimeException("Already closed.");
            }
            _closed = true;
        }

    }

    @Test
    public void testInvalidClose() throws Exception {
        var closeable = new _SampleCloseable();
        var mono = Mono.just("good").doOnTerminate(() -> {
            closeable.close();
        });
        // 内部でcloseが呼ばれる
        var value = mono.block();
        Assertions.assertEquals("good", value);
        // 内部でcloseが呼ばれる(2回目なのでエラーになる)
        Assertions.assertThrows(RuntimeException.class, () -> mono.block());
    }

deferを使用した修正

deferを使用することで、Monoの内部でリソースの初期化を行う事が出来ます。

    @Test
    public void testDefer() throws Exception {
        var mono = Mono.defer(() -> {
            // リソースの初期化をdefer内で行います。
            var closeable = new _SampleCloseable();
            return Mono.just("good").doOnTerminate(() -> {
                closeable.close();
            });
        });
        // block実行時に、リソースが初期化されて、閉じられます
        var value = mono.block();
        Assertions.assertEquals("good", value);
        // block実行時に、リソースが初期化されて、閉じられます
        value = mono.block();
        Assertions.assertEquals("good", value);
    }

結論

deferを使うことで、リソースの初期化と解放をMonoやFluxに閉じて行う事が出来ます。
MonoやFluxの生成と同じスコープでリソースの初期化をしているケースでは、こちらの問題が発生する可能性を常に考慮する必要があります。