MessageResponseListener extends MessageListener implements Disposable { public MessageResponse waitResponse( MessageRequest request, int requestTimeout, int responseTimeout ) throws PeerException { MessageStatus messageStatus; try { long start = System.currentTimeMillis(); messageStatus = messenger.getMessageStatus( request.getMessageId() ); while ( messageStatus == MessageStatus.IN_PROCESS && System.currentTimeMillis() - start < ( requestTimeout + 5 ) * 1000 ) { Thread.sleep( 100 ); messageStatus = messenger.getMessageStatus( request.getMessageId() ); } } catch ( Exception e ) { throw new PeerException( e ); } if ( messageStatus != MessageStatus.SENT ) { throw new PeerException( "Failed to send message" ); } semaphoreMap.put( request.getId(), new Semaphore( 0 ) ); try { if ( !semaphoreMap.get( request.getId() ).tryAcquire( responseTimeout + 5L, TimeUnit.SECONDS ) ) { LOG.warn( "Could not obtain response within timeout interval" ); } } catch ( InterruptedException e ) { Thread.currentThread().interrupt(); } semaphoreMap.remove( request.getId() ); return responses.remove( request.getId() ); } MessageResponseListener( Messenger messenger ); @Override void dispose(); MessageResponse waitResponse( MessageRequest request, int requestTimeout, int responseTimeout ); @Override void onMessage( final Message message );  }
@Test public void testWaitResponse() throws Exception { listener.waitResponse( messageRequest, REQUEST_TIMEOUT, RESPONSE_TIMEOUT ); verify( responses ).remove( any( UUID.class ) ); when( messenger.getMessageStatus( any( UUID.class ) ) ).thenReturn( MessageStatus.EXPIRED ); try { listener.waitResponse( messageRequest, REQUEST_TIMEOUT, RESPONSE_TIMEOUT ); fail( "Expected PeerException" ); } catch ( PeerException e ) { } when( messenger.getMessageStatus( any( UUID.class ) ) ).thenReturn( MessageStatus.SENT ); doThrow( new MessageException( "" ) ).when( messenger ).getMessageStatus( any( UUID.class ) ); try { listener.waitResponse( messageRequest, REQUEST_TIMEOUT, RESPONSE_TIMEOUT ); fail( "Expected PeerException" ); } catch ( PeerException e ) { } reset( messenger ); when( messenger.getMessageStatus( any( UUID.class ) ) ).thenReturn( MessageStatus.SENT ); doThrow( interruptedException ).when( semaphore ).tryAcquire( anyLong(), any( TimeUnit.class ) ); listener.waitResponse( messageRequest, REQUEST_TIMEOUT, RESPONSE_TIMEOUT ); assertTrue( Thread.interrupted() ); }