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の生成と同じスコープでリソースの初期化をしているケースでは、こちらの問題が発生する可能性を常に考慮する必要があります。