@Test public void testUploadBatch() throws DataSinkException, IOException { TestUtils.pause(50); for(int i = 0; i < 25; i++) { sink.receive(message); } TestUtils.pause(1000); assertTrue(FakeHttp.httpRequestWasMade()); Type listType = new TypeToken<List<SimpleVehicleMessage>>() {}.getType(); ArrayList<SimpleVehicleMessage> messages = new ArrayList<>(); HttpPost request; while((request = (HttpPost) FakeHttp.getNextSentHttpRequest()) != null) { InputStream payload = request.getEntity().getContent(); int length = payload.available(); byte[] buffer = new byte[length]; payload.read(buffer); messages.addAll((List<SimpleVehicleMessage>)gson.fromJson(new String(buffer), listType)); } assertThat(messages, hasSize(25)); for(SimpleVehicleMessage deserializedMessage : messages) { assertThat(message, equalTo((VehicleMessage) deserializedMessage)); } }
@Override public void receive(VehicleMessage message) { mRecordQueue.offer(message); if(mRecordQueue.size() >= UPLOAD_BATCH_SIZE) { try { mQueueLock.lock(); mRecordsQueued.signal(); } finally { mQueueLock.unlock(); } } }
UploaderSink extends ContextualVehicleDataSink { @Override public void receive(VehicleMessage message) { mRecordQueue.offer(message); if(mRecordQueue.size() >= UPLOAD_BATCH_SIZE) { try { mQueueLock.lock(); mRecordsQueued.signal(); } finally { mQueueLock.unlock(); } } } }
UploaderSink extends ContextualVehicleDataSink { @Override public void receive(VehicleMessage message) { mRecordQueue.offer(message); if(mRecordQueue.size() >= UPLOAD_BATCH_SIZE) { try { mQueueLock.lock(); mRecordsQueued.signal(); } finally { mQueueLock.unlock(); } } } UploaderSink(Context context, URI uri);  UploaderSink(Context context, String path); }
UploaderSink extends ContextualVehicleDataSink { @Override public void receive(VehicleMessage message) { mRecordQueue.offer(message); if(mRecordQueue.size() >= UPLOAD_BATCH_SIZE) { try { mQueueLock.lock(); mRecordsQueued.signal(); } finally { mQueueLock.unlock(); } } } UploaderSink(Context context, URI uri);  UploaderSink(Context context, String path); @Override void stop(); @Override void receive(VehicleMessage message); static boolean validatePath(String path); @Override String toString(); }
UploaderSink extends ContextualVehicleDataSink { @Override public void receive(VehicleMessage message) { mRecordQueue.offer(message); if(mRecordQueue.size() >= UPLOAD_BATCH_SIZE) { try { mQueueLock.lock(); mRecordsQueued.signal(); } finally { mQueueLock.unlock(); } } } UploaderSink(Context context, URI uri);  UploaderSink(Context context, String path); @Override void stop(); @Override void receive(VehicleMessage message); static boolean validatePath(String path); @Override String toString();  }