make BatchProcessingAgent more efficient

...and fix Observable.bufferWithTimeOrCount to call it properly