public int sendAll ( Iterable < Observable < Integer > > batches , final int numMetrics , long timeoutMillis ) { final AtomicBoolean err = new AtomicBoolean ( false ) ; final AtomicInteger updated = new AtomicInteger ( 0 ) ; LOGGER . debug ( "Got {} ms to send {} metrics" , timeoutMillis , numMetrics ) ; try { final CountDownLatch completed = new CountDownLatch ( 1 ) ; final Subscription s = Observable . mergeDelayError ( Observable . from ( batches ) ) . timeout ( timeoutMillis , TimeUnit . MILLISECONDS ) . subscribeOn ( Schedulers . immediate ( ) ) . subscribe ( updated :: addAndGet , exc -> { logErr ( "onError caught" , exc , updated . get ( ) , numMetrics ) ; err . set ( true ) ; completed . countDown ( ) ; } , completed :: countDown ) ; try { completed . await ( timeoutMillis , TimeUnit . MILLISECONDS ) ; } catch ( InterruptedException interrupted ) { err . set ( true ) ; s . unsubscribe ( ) ; LOGGER . warn ( "Timed out sending metrics. {}/{} sent" , updated . get ( ) , numMetrics ) ; } } catch ( Exception e ) { err . set ( true ) ; logErr ( "Unexpected " , e , updated . get ( ) , numMetrics ) ; } if ( updated . get ( ) < numMetrics && ! err . get ( ) ) { LOGGER . warn ( "No error caught, but only {}/{} sent." , updated . get ( ) , numMetrics ) ; } return updated . get ( ) ; }
Attempt to send all the batches totalling numMetrics in the allowed time .