MessengerImpl implements Messenger, MessageProcessor { @Override public void processMessage( String envelopeString ) throws MessageException { try { Envelope envelope = JsonUtil.fromJson( envelopeString, Envelope.class ); notifyListeners( envelope ); } catch ( NullPointerException | JsonSyntaxException e ) { LOG.error( "Error in processMessage", e ); throw new MessageException( e ); } }  void init(); void destroy(); void setDaoManager( final DaoManager daoManager ); @Override Message createMessage( final Object payload ); @Override void sendMessage( final Peer peer, final Message message, final String recipient, final int timeToLive,
                             final Map<String, String> headers ); @Override MessageStatus getMessageStatus( final UUID messageId ); @Override void processMessage( String envelopeString ); void addMessageListener( final MessageListener listener ); void removeMessageListener( final MessageListener listener );  }
@Test public void testProcessMessage() throws Exception { MessageImpl message = new MessageImpl( LOCAL_PEER_ID, PAYLOAD ); Envelope envelope = new Envelope( message, LOCAL_PEER_ID, RECIPIENT, TIME_TO_LIVE, HEADERS ); MessageListener listener = mock( MessageListener.class ); messenger.addMessageListener( listener ); when( listener.getRecipient() ).thenReturn( RECIPIENT ); messenger.processMessage( JsonUtil.toJson( envelope ) ); verify( notificationExecutor ).execute( isA( MessageNotifier.class ) ); try { messenger.processMessage( null ); fail( "Expected message exception" ); } catch ( MessageException e ) { } }