### Question:
ScanRunnable implements Callback<Integer> { @Override public void onMessage(Integer numberOfMessagesToSend) { if (this.close) { return; } long numberOfMsgsLeft = numberOfMessagesToSend; List<Result> scanResults = new ArrayList<>(); List<Integer> cellsPerResult = new ArrayList<>(); ByteBuffer previousRow = null; while (!this.close && numberOfMsgsLeft > 0) { int rowsToSend = 0; boolean moreResults; do { List<Cell> rawCells = new ArrayList<>(); try { moreResults = scanner.nextRaw(rawCells); if (!moreResults) { this.scanner.close(); this.close = true; } } catch (IOException e) { e.printStackTrace(); return; } List<c5db.client.generated.Cell> cells = new ArrayList<>(); for (Cell cell : rawCells) { ByteBuffer cellBufferRow = ByteBuffer.wrap(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()); cells.add(ReverseProtobufUtil.toCell(cell)); if (!(previousRow == null || previousRow.compareTo(cellBufferRow) == 0)) { cellsPerResult.add(cells.size()); scanResults.add(new Result(cells, cells.size(), cells.size() > 0)); cells = new ArrayList<>(); } previousRow = ByteBuffer.wrap(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()); } if (cells.size() > 0) { cellsPerResult.add(cells.size()); scanResults.add(new Result(cells, cells.size(), cells.size() > 0)); } rowsToSend++; } while (moreResults && rowsToSend < 100 && numberOfMessagesToSend - rowsToSend > 0); ScanResponse scanResponse = new ScanResponse(cellsPerResult, scannerId, moreResults, 0, scanResults); Response response = new Response(Response.Command.SCAN, call.getCommandId(), null, null, scanResponse, null); ctx.writeAndFlush(response); numberOfMsgsLeft -= rowsToSend; } } ScanRunnable(final ChannelHandlerContext ctx,
                      final Call call,
                      final long scannerId,
                      final Region region); @Override void onMessage(Integer numberOfMessagesToSend);  }### Answer:
@Test public void scannerCanDeliverASingleMessageOnlyOnce() throws InterruptedException, IOException { byte[] row = Bytes.toBytes("row"); byte[] cf = Bytes.toBytes("cf"); byte[] cq = Bytes.toBytes("cq"); byte[] value = Bytes.toBytes("value"); KeyValue keyValue = new KeyValue(row, cf, cq, value); context.checking(new Expectations() { { oneOf(regionScanner).nextRaw(with(any(List.class))); will(addElements(keyValue)); oneOf(ctx).writeAndFlush(with(any(Response.class))); } }); scanRunnable.onMessage(1); }
@Test public void scannerCanDeliverWithMultipleOnMessageInvocation() throws InterruptedException, IOException { ArrayList<KeyValue> kvs = new ArrayList<>(); for (int i = 0; i != 10000; i++) { byte[] row = Bytes.toBytes(i); byte[] cf = Bytes.toBytes(i); byte[] cq = Bytes.toBytes(i); byte[] value = Bytes.toBytes(i); KeyValue keyValue = new KeyValue(row, cf, cq, value); kvs.add(keyValue); } context.checking(new Expectations() { { exactly(3).of(regionScanner).nextRaw(with(any(List.class))); will(addElements(kvs.toArray())); oneOf(ctx).writeAndFlush(with(any(Response.class))); } }); scanRunnable.onMessage(3); }