ReactorBulkheadAspectExt implements BulkheadAspectExt { @Override public Object handle(ProceedingJoinPoint proceedingJoinPoint, Bulkhead bulkhead, String methodName) throws Throwable { Object returnValue = proceedingJoinPoint.proceed(); if (Flux.class.isAssignableFrom(returnValue.getClass())) { Flux<?> fluxReturnValue = (Flux<?>) returnValue; return fluxReturnValue.transformDeferred(BulkheadOperator.of(bulkhead)); } else if (Mono.class.isAssignableFrom(returnValue.getClass())) { Mono<?> monoReturnValue = (Mono<?>) returnValue; return monoReturnValue.transformDeferred(BulkheadOperator.of(bulkhead)); } else { logger.error("Unsupported type for Reactor BulkHead {}", returnValue.getClass().getTypeName()); throw new IllegalArgumentException( "Not Supported type for the BulkHead in Reactor :" + returnValue.getClass() .getName()); } }  @Override boolean canHandleReturnType(Class returnType); @Override Object handle(ProceedingJoinPoint proceedingJoinPoint, Bulkhead bulkhead,
        String methodName);  }
@Test public void testReactorTypes() throws Throwable { Bulkhead bulkhead = Bulkhead.ofDefaults("test"); when(proceedingJoinPoint.proceed()).thenReturn(Mono.just("Test")); assertThat(reactorBulkheadAspectExt.handle(proceedingJoinPoint, bulkhead, "testMethod")) .isNotNull(); when(proceedingJoinPoint.proceed()).thenReturn(Flux.just("Test")); assertThat(reactorBulkheadAspectExt.handle(proceedingJoinPoint, bulkhead, "testMethod")) .isNotNull(); }