MessageSender { protected void deliverMessages() { Set<Envelope> envelopes = messengerDataService.getEnvelopes(); Map<String, Set<Envelope>> peerEnvelopesMap = Maps.newHashMap(); int maxTimeToLive = 0; for ( Envelope envelope : envelopes ) { if ( envelope.getTimeToLive() > maxTimeToLive ) { maxTimeToLive = envelope.getTimeToLive(); } Set<Envelope> peerEnvelopes = peerEnvelopesMap.get( envelope.getTargetPeerId() ); if ( peerEnvelopes == null ) { peerEnvelopes = new TreeSet<>( new Comparator<Envelope>() { @Override public int compare( final Envelope o1, final Envelope o2 ) { return o1.getCreateDate().compareTo( o2.getCreateDate() ); } } ); peerEnvelopesMap.put( envelope.getTargetPeerId(), peerEnvelopes ); } peerEnvelopes.add( envelope ); } try { for ( Map.Entry<String, Set<Envelope>> envelopsPerPeer : peerEnvelopesMap.entrySet() ) { Peer targetPeer = messenger.getPeerManager().getPeer( envelopsPerPeer.getKey() ); if ( targetPeer.isLocal() ) { completer.submit( new LocalPeerMessageSender( messenger, messengerDataService, envelopsPerPeer.getValue() ) ); } else { completer.submit( new RemotePeerMessageSender( messengerDataService, targetPeer, envelopsPerPeer.getValue() ) ); } } try { for ( int i = 0; i < peerEnvelopesMap.size(); i++ ) { Future<Boolean> future = completer.take(); future.get(); } } catch ( InterruptedException | ExecutionException e ) { LOG.warn( "ignore", e ); } } catch ( PeerException e ) { LOG.error( e.getMessage(), e ); } } MessageSender( final MessengerDataService messengerDataService, final MessengerImpl messenger ); void init(); void dispose(); static final int SLEEP_BETWEEN_ITERATIONS_SEC; }
@Test public void testDeliverMessages() throws Exception { when( messengerDataService.getEnvelopes() ).thenReturn( Sets.newHashSet( envelope ) ); when( envelope.getTimeToLive() ).thenReturn( TIME_TO_LIVE ); when( envelope.getTargetPeerId() ).thenReturn( TARGET_PEER_ID ); when( envelope.getCreateDate() ).thenReturn( CREATE_DATE ); when( peerManager.getPeer( TARGET_PEER_ID ) ).thenReturn( peer ); when( peer.getId() ).thenReturn( TARGET_PEER_ID ); when( peerManager.getLocalPeer() ).thenReturn( localPeer ); when( peerInfo.getId() ).thenReturn( UUID.randomUUID().toString() ); when( localPeer.isLocal() ).thenReturn( false ); when( localPeer.getId() ).thenReturn( UUID.randomUUID().toString() ); when( completer.take() ).thenReturn( future ); when( messenger.getPeerManager() ).thenReturn( peerManager ); messageSender.deliverMessages(); verify( completer ).submit( isA( RemotePeerMessageSender.class ) ); when( peer.isLocal() ).thenReturn( true ); messageSender.deliverMessages(); verify( completer ).submit( isA( LocalPeerMessageSender.class ) ); doThrow( new InterruptedException() ).when( completer ).take(); messageSender.deliverMessages(); verify( logger ).warn( anyString(), isA( InterruptedException.class ) ); }