ActorReceiver implements IMessageReceiver { @Override public void receive(final IMessageProcessor processor) throws MessageReceiveException, AsynchronousOperationException { Throwable syncException = null; if (isBusy.compareAndSet(false, true)) { try { childReceiver.receive(processor); } catch (Throwable e) { syncException = e; } finally { isBusy.set(false); } } else { processor.pauseProcess(); queue.add(processor); } executeDelayed(); if (null != syncException) { throw new MessageReceiveException("Failed to execute actor receiver.", syncException); } } ActorReceiver(final IMessageReceiver childReceiver); @Override void receive(final IMessageProcessor processor);  }
@Test public void Should_enqueueMessageProcessor_When_ActorIsBusy() throws Exception { receiverFlag.set(true); when(receiverQueueMock.isEmpty()).thenReturn(false); ActorReceiver actorReceiver = new ActorReceiver(childReceiverMock); actorReceiver.receive(processorMock); verify(receiverQueueMock).add(same(processorMock)); }
@Test public void Should_executeReceiverImmediately_When_ActorIsNotBusy() throws Exception { receiverFlag.set(false); doAnswer(invocationOnMock -> { assertTrue(receiverFlag.get()); return null; }).when(childReceiverMock).receive(any()); when(receiverQueueMock.isEmpty()).thenReturn(true); ActorReceiver actorReceiver = new ActorReceiver(childReceiverMock); actorReceiver.receive(processorMock); verify(childReceiverMock).receive(same(processorMock)); }
@Test public void Should_executeReceiversFromQueue_When_QueueIsNotEmptyAndActorIsNotBusy() throws Exception { IMessageProcessor[] processorMocks = new IMessageProcessor[] { mock(IMessageProcessor.class), mock(IMessageProcessor.class) }; receiverFlag.set(false); doAnswer(invocationOnMock -> { assertTrue(receiverFlag.get()); return null; }).when(childReceiverMock).receive(any()); when(receiverQueueMock.isEmpty()) .thenAnswer(invocationOnMock -> { assertFalse(receiverFlag.get()); return false; }) .thenAnswer(invocationOnMock -> { assertFalse(receiverFlag.get()); return true; }); when(receiverQueueMock.poll()) .thenReturn(processorMocks[0]) .thenReturn(processorMocks[1]) .thenReturn(null); ActorReceiver actorReceiver = new ActorReceiver(childReceiverMock); actorReceiver.receive(processorMock); verify(childReceiverMock).receive(processorMock); verify(childReceiverMock).receive(processorMocks[0]); verify(childReceiverMock).receive(processorMocks[1]); }
@Test public void Should_rethrowExceptionAfterCheckingQueue_When_nestedReceiverThrows() throws Exception { MessageReceiveException exception = mock(MessageReceiveException.class); when(receiverQueueMock.isEmpty()).thenReturn(true); doAnswer(invocationOnMock -> { assertTrue(receiverFlag.get()); throw exception; }).when(childReceiverMock).receive(same(processorMock)); ActorReceiver actorReceiver = new ActorReceiver(childReceiverMock); try { actorReceiver.receive(processorMock); fail(); } catch (MessageReceiveException e) { assertSame(exception, e.getCause()); } }
@Test public void Should_handleExceptionOccurredWhileCompletingAsynchronousOperationOnDelayedProcessor() throws Exception { MessageReceiveException messageReceiveExceptionMock = mock(MessageReceiveException.class); AsynchronousOperationException asynchronousOperationExceptionMock = mock(AsynchronousOperationException.class); when(receiverQueueMock.isEmpty()).thenAnswer(invocationOnMock -> { receiverFlag.set(false); return false; }).thenReturn(true); when(receiverQueueMock.poll()).thenReturn(processorMock).thenReturn(null); doThrow(messageReceiveExceptionMock).when(childReceiverMock).receive(same(processorMock)); doThrow(asynchronousOperationExceptionMock).when(processorMock).continueProcess(same(messageReceiveExceptionMock)); ActorReceiver actorReceiver = new ActorReceiver(childReceiverMock); receiverFlag.set(true); actorReceiver.receive(processorMock); verify(asynchronousOperationExceptionMock).printStackTrace(); verify(asynchronousOperationExceptionMock).addSuppressed(messageReceiveExceptionMock); }