@Override protected void onMessageOnProducerThread(T msg){
synchronized (_lock) {
    if (_pending == null) {
      _pending=new ArrayList<>();
      _queue.schedule(_flushRunnable,_interval,_timeUnit);
    }
    _pending.add(msg);
  }
}

@ Override protected void onMessageOnProducerThread ( T msg ) { synchronized ( _lock ) { if ( _pending == null ) { _pending = new ArrayList < > ( ) ; _queue . schedule ( _flushRunnable , _interval , _timeUnit ) ; } _pending . add ( msg ) ; } }
receives message and batches as needed .
