MongoDBChunkReadStream implements ChunkReadStream { private void doRead() { if (!readInProgress) { readInProgress = true; Buffer buff = Buffer.buffer(readBufferSize); ByteBuffer bb = ByteBuffer.allocate(readBufferSize); doRead(buff, bb, ar -> { if (ar.succeeded()) { readInProgress = false; Buffer buffer = ar.result(); if (buffer.length() == 0) { handleEnd(); } else { handleData(buffer); if (!paused && handler != null) { doRead(); } } } else { handleException(ar.cause()); } }); } } MongoDBChunkReadStream(AsyncInputStream is, long size,
      int readBufferSize, Context context); @Override long getSize(); @Override ReadStream<Buffer> exceptionHandler(Handler<Throwable> handler); @Override ReadStream<Buffer> handler(Handler<Buffer> handler); @Override ReadStream<Buffer> pause(); @Override ReadStream<Buffer> resume(); @Override ReadStream<Buffer> fetch(long amount); @Override ReadStream<Buffer> endHandler(Handler<Void> endHandler); @Override void close(Handler<AsyncResult<Void>> handler);  }
@Test public void tiny(TestContext context) { Vertx vertx = rule.vertx(); getChunkSize(vertx, context.asyncAssertSuccess(cs -> { doRead(2, cs, vertx, context); })); }
@Test public void small(TestContext context) { Vertx vertx = rule.vertx(); getChunkSize(vertx, context.asyncAssertSuccess(cs -> { doRead(1024, cs, vertx, context); })); }
@Test public void defaultChunkSize(TestContext context) { Vertx vertx = rule.vertx(); getChunkSize(vertx, context.asyncAssertSuccess(cs -> { doRead(cs, cs, vertx, context); })); }
@Test public void medium(TestContext context) { Vertx vertx = rule.vertx(); getChunkSize(vertx, context.asyncAssertSuccess(cs -> { doRead(cs * 3 / 2, cs, vertx, context); })); }
@Test public void large(TestContext context) { Vertx vertx = rule.vertx(); getChunkSize(vertx, context.asyncAssertSuccess(cs -> { doRead(cs * 10 + 100, cs, vertx, context); })); }