ReactiveInvocationHandler implements InvocationHandler { @Override public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { if ("equals".equals(method.getName())) { try { Object otherHandler = args.length > 0 && args[0] != null ? Proxy.getInvocationHandler(args[0]) : null; return equals(otherHandler); } catch (IllegalArgumentException e) { return false; } } else if ("hashCode".equals(method.getName())) { return hashCode(); } else if ("toString".equals(method.getName())) { return toString(); } return this.invoke(method, this.dispatch.get(method), args); } ReactiveInvocationHandler(Target<?> target,
      Map<Method, MethodHandler> dispatch); @Override Object invoke(Object proxy, Method method, Object[] args); @Override int hashCode(); @Override boolean equals(Object obj); @Override String toString();  }
@SuppressWarnings("unchecked") @Test public void invokeOnSubscribeReactor() throws Throwable { given(this.methodHandler.invoke(any())).willReturn("Result"); ReactorInvocationHandler handler = new ReactorInvocationHandler(this.target, Collections.singletonMap(method, this.methodHandler), Schedulers.elastic()); Object result = handler.invoke(method, this.methodHandler, new Object[] {}); assertThat(result).isInstanceOf(Mono.class); verifyZeroInteractions(this.methodHandler); StepVerifier.create((Mono) result) .expectNext("Result") .expectComplete() .verify(); verify(this.methodHandler, times(1)).invoke(any()); }
@Test public void invokeOnSubscribeEmptyReactor() throws Throwable { given(this.methodHandler.invoke(any())).willReturn(null); ReactorInvocationHandler handler = new ReactorInvocationHandler(this.target, Collections.singletonMap(method, this.methodHandler), Schedulers.elastic()); Object result = handler.invoke(method, this.methodHandler, new Object[] {}); assertThat(result).isInstanceOf(Mono.class); verifyZeroInteractions(this.methodHandler); StepVerifier.create((Mono) result) .expectComplete() .verify(); verify(this.methodHandler, times(1)).invoke(any()); }
@Test public void invokeFailureReactor() throws Throwable { given(this.methodHandler.invoke(any())).willThrow(new IOException("Could Not Decode")); ReactorInvocationHandler handler = new ReactorInvocationHandler(this.target, Collections.singletonMap(this.method, this.methodHandler), Schedulers.elastic()); Object result = handler.invoke(this.method, this.methodHandler, new Object[] {}); assertThat(result).isInstanceOf(Mono.class); verifyZeroInteractions(this.methodHandler); StepVerifier.create((Mono) result) .expectError(IOException.class) .verify(); verify(this.methodHandler, times(1)).invoke(any()); }
@SuppressWarnings("unchecked") @Test public void invokeOnSubscribeRxJava() throws Throwable { given(this.methodHandler.invoke(any())).willReturn("Result"); RxJavaInvocationHandler handler = new RxJavaInvocationHandler(this.target, Collections.singletonMap(this.method, this.methodHandler), io.reactivex.schedulers.Schedulers.trampoline()); Object result = handler.invoke(this.method, this.methodHandler, new Object[] {}); assertThat(result).isInstanceOf(Flowable.class); verifyZeroInteractions(this.methodHandler); StepVerifier.create((Flowable) result) .expectNext("Result") .expectComplete() .verify(); verify(this.methodHandler, times(1)).invoke(any()); }
@Test public void invokeOnSubscribeEmptyRxJava() throws Throwable { given(this.methodHandler.invoke(any())).willReturn(null); RxJavaInvocationHandler handler = new RxJavaInvocationHandler(this.target, Collections.singletonMap(this.method, this.methodHandler), io.reactivex.schedulers.Schedulers.trampoline()); Object result = handler.invoke(this.method, this.methodHandler, new Object[] {}); assertThat(result).isInstanceOf(Flowable.class); verifyZeroInteractions(this.methodHandler); StepVerifier.create((Flowable) result) .expectComplete() .verify(); verify(this.methodHandler, times(1)).invoke(any()); }
@Test public void invokeFailureRxJava() throws Throwable { given(this.methodHandler.invoke(any())).willThrow(new IOException("Could Not Decode")); RxJavaInvocationHandler handler = new RxJavaInvocationHandler(this.target, Collections.singletonMap(this.method, this.methodHandler), io.reactivex.schedulers.Schedulers.trampoline()); Object result = handler.invoke(this.method, this.methodHandler, new Object[] {}); assertThat(result).isInstanceOf(Flowable.class); verifyZeroInteractions(this.methodHandler); StepVerifier.create((Flowable) result) .expectError(IOException.class) .verify(); verify(this.methodHandler, times(1)).invoke(any()); }