code: 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);
              }
          }
      }
docs: Submit a request to a batch. If the batch maxSize is hit trigger the batch immediately.
  
  @param arg argument to a {@link RequestCollapser}
  @return Observable<ResponseType>
  @throws IllegalStateException
  if submitting after shutdown