@Override protected void onMessageOnProducerThread(T msg){
  _lock.lock();
  try {
    if (_pending.isEmpty()) {
      _queue.schedule(_flushRunnable,_interval,_timeUnit);
    }
    _pending.add(msg);
  }
  finally {
    _lock.unlock();
  }
}

@ Override protected void onMessageOnProducerThread ( T msg ) { _lock . lock ( ) ; try { if ( _pending . isEmpty ( ) ) { _queue . schedule ( _flushRunnable , _interval , _timeUnit ) ; } _pending . add ( msg ) ; } finally { _lock . unlock ( ) ; } }
receives message and batches as needed .
