单元测试中Mono.just的map方法未执行，如何正确触发执行？

现有代码 #

JUnit测试代码 #

@Mock
Client client;

@Test
public void test() {
    doAnswer(r -> {
        response.setData(data);
        return Mono.just(response);
    }).when(client).doSomething(any(Response.class));
    ...
    foo();
}

业务代码 #

public void foo() {
    Response result = new Response();
    client.doSomething(result).map( resp -> {
        // 这段代码在JUnit测试中未执行
        if (somelag) {
           ...
        }
        return resp;
    });
}

问题说明 #

• 该map内的代码在运行时可正常执行，但JUnit测试中完全不执行
• 问题可能与线程或Reactor的订阅机制相关
• 运行时使用reactor.core.publisher.Mono实现异步逻辑

相关Mono源码细节 #

map方法实现 #

public final <R> Mono<R> map(Function<? super T, ? extends R> mapper) {
    return this instanceof Fuseable ? onAssembly(new MonoMapFuseable(this, mapper)) : onAssembly(new MonoMap(this, mapper));
}

• 运行时：this instanceof Fuseable == false
• JUnit测试中：this instanceof Fuseable == true

onAssembly方法实现 #

protected static <T> Mono<T> onAssembly(Mono<T> source) {
    Function<Publisher, Publisher> hook = Hooks.onEachOperatorHook;
    if (hook != null) {
        source = (Mono)hook.apply(source);
    }

    if (Hooks.GLOBAL_TRACE) {
        FluxOnAssembly.AssemblySnapshot stacktrace = new FluxOnAssembly.AssemblySnapshot((String)null, (Supplier)Traces.callSiteSupplierFactory.get());
        source = (Mono)Hooks.addAssemblyInfo(source, stacktrace);
    }

    return source;
}

• JUnit测试中，reactor.core.publisher.Mono的hook == null且Hooks.GLOBAL_TRACE == false

解决方案 #

核心原因是Reactor的数据流是惰性执行的：Mono的操作符（比如map）只是定义了数据处理逻辑，只有当调用subscribe()或block()触发订阅时，整个数据流才会开始执行。业务代码中只构建了数据流但没有触发订阅，所以测试中map内的代码不会运行。

方案1：修改业务代码触发订阅 #

如果业务逻辑需要立即执行这段map内的逻辑，在map之后添加订阅操作：

public void foo() {
    Response result = new Response();
    client.doSomething(result)
          .map( resp -> {
              if (somelag) {
                 ...
              }
              return resp;
          })
          .subscribe(); // 触发订阅，执行map逻辑
}

如果需要等待执行完成再继续后续逻辑，可以用block()（适合同步场景）：

public void foo() {
    Response result = new Response();
    client.doSomething(result)
          .map( resp -> {
              if (somelag) {
                 ...
              }
              return resp;
          })
          .block(); // 阻塞等待执行完成
}

方案2：在测试中手动触发订阅 #

如果不想修改业务代码，可以在测试中获取到该Mono实例并触发订阅。比如调整业务代码返回Mono，或者通过Mock捕获返回值：

// 修改业务代码返回Mono
public Mono<Response> foo() {
    Response result = new Response();
    return client.doSomething(result)
          .map( resp -> {
              if (somelag) {
                 ...
              }
              return resp;
          });
}

// 在测试中触发订阅
@Test
public void test() {
    doAnswer(r -> {
        response.setData(data);
        return Mono.just(response);
    }).when(client).doSomething(any(Response.class));
    
    // 获取Mono并触发订阅
    foo().block(); // 或者subscribe()
}

关于Fuseable差异的说明 #

测试中返回的Mono.just()是Fuseable实现，而运行时的Mono不是，但这不是核心问题——无论是否是Fuseable，只要没有触发订阅，map内的代码都不会执行。解决订阅问题后，Fuseable的差异不会影响逻辑执行。

内容的提问来源于stack exchange，提问作者THM