public Observable < ResponseType > submitRequest ( final RequestArgumentType arg ) { /* * We only want the timer ticking if there are actually things to do so we register it the first time something is added. */ if ( ! timerListenerRegistered . get ( ) && timerListenerRegistered . compareAndSet ( false , true ) ) { /* schedule the collapsing task to be executed every x milliseconds (x defined inside CollapsedTask) */ timerListenerReference . set ( timer . addListener ( new CollapsedTask ( ) ) ) ; } // loop until succeed (compare-and-set spin-loop) while ( true ) { final RequestBatch < BatchReturnType , ResponseType , RequestArgumentType > b = batch . get ( ) ; if ( b == null ) { return Observable . error ( new IllegalStateException ( "Submitting requests after collapser is shutdown" ) ) ; } final Observable < ResponseType > response ; if ( arg != null ) { response = b . offer ( arg ) ; } else { response = b . offer ( ( RequestArgumentType ) NULL_SENTINEL ) ; } // it will always get an Observable unless we hit the max batch size if ( response != null ) { return response ; } else { // this batch can't accept requests so create a new one and set it if another thread doesn't beat us createNewBatchAndExecutePreviousIfNeeded ( b ) ; } } }
Submit a request to a batch . If the batch maxSize is hit trigger the batch immediately .