@Test public void readEmptyFile() throws Exception { try (DataReader reader = create(0, 10)) { setReadResponses(mClient, 0, 0, 0); assertEquals(null, reader.readChunk()); } validateReadRequestSent(mClient, 0, 10, true, CHUNK_SIZE); }
@Override public DataBuffer readChunk() throws IOException { Preconditions.checkState(!mClient.get().isShutdown(), "Data reader is closed while reading data chunks."); DataBuffer buffer = null; ReadResponse response = null; if (mStream instanceof GrpcDataMessageBlockingStream) { DataMessage<ReadResponse, DataBuffer> message = ((GrpcDataMessageBlockingStream<ReadRequest, ReadResponse>) mStream) .receiveDataMessage(mDataTimeoutMs); if (message != null) { response = message.getMessage(); buffer = message.getBuffer(); if (buffer == null && response.hasChunk() && response.getChunk().hasData()) { ByteBuffer byteBuffer = response.getChunk().getData().asReadOnlyByteBuffer(); buffer = new NioDataBuffer(byteBuffer, byteBuffer.remaining()); } Preconditions.checkState(buffer != null, "response should always contain chunk"); } } else { response = mStream.receive(mDataTimeoutMs); if (response != null) { Preconditions.checkState(response.hasChunk() && response.getChunk().hasData(), "response should always contain chunk"); ByteBuffer byteBuffer = response.getChunk().getData().asReadOnlyByteBuffer(); buffer = new NioDataBuffer(byteBuffer, byteBuffer.remaining()); } } if (response == null) { return null; } mPosToRead += buffer.readableBytes(); try { mStream.send(mReadRequest.toBuilder().setOffsetReceived(mPosToRead).build()); } catch (Exception e) { LOG.debug("Failed to send receipt of data to worker {} for request {}: {}.", mAddress, mReadRequest, e.getMessage()); } Preconditions.checkState(mPosToRead - mReadRequest.getOffset() <= mReadRequest.getLength()); return buffer; }
GrpcDataReader implements DataReader { @Override public DataBuffer readChunk() throws IOException { Preconditions.checkState(!mClient.get().isShutdown(), "Data reader is closed while reading data chunks."); DataBuffer buffer = null; ReadResponse response = null; if (mStream instanceof GrpcDataMessageBlockingStream) { DataMessage<ReadResponse, DataBuffer> message = ((GrpcDataMessageBlockingStream<ReadRequest, ReadResponse>) mStream) .receiveDataMessage(mDataTimeoutMs); if (message != null) { response = message.getMessage(); buffer = message.getBuffer(); if (buffer == null && response.hasChunk() && response.getChunk().hasData()) { ByteBuffer byteBuffer = response.getChunk().getData().asReadOnlyByteBuffer(); buffer = new NioDataBuffer(byteBuffer, byteBuffer.remaining()); } Preconditions.checkState(buffer != null, "response should always contain chunk"); } } else { response = mStream.receive(mDataTimeoutMs); if (response != null) { Preconditions.checkState(response.hasChunk() && response.getChunk().hasData(), "response should always contain chunk"); ByteBuffer byteBuffer = response.getChunk().getData().asReadOnlyByteBuffer(); buffer = new NioDataBuffer(byteBuffer, byteBuffer.remaining()); } } if (response == null) { return null; } mPosToRead += buffer.readableBytes(); try { mStream.send(mReadRequest.toBuilder().setOffsetReceived(mPosToRead).build()); } catch (Exception e) { LOG.debug("Failed to send receipt of data to worker {} for request {}: {}.", mAddress, mReadRequest, e.getMessage()); } Preconditions.checkState(mPosToRead - mReadRequest.getOffset() <= mReadRequest.getLength()); return buffer; } }
GrpcDataReader implements DataReader { @Override public DataBuffer readChunk() throws IOException { Preconditions.checkState(!mClient.get().isShutdown(), "Data reader is closed while reading data chunks."); DataBuffer buffer = null; ReadResponse response = null; if (mStream instanceof GrpcDataMessageBlockingStream) { DataMessage<ReadResponse, DataBuffer> message = ((GrpcDataMessageBlockingStream<ReadRequest, ReadResponse>) mStream) .receiveDataMessage(mDataTimeoutMs); if (message != null) { response = message.getMessage(); buffer = message.getBuffer(); if (buffer == null && response.hasChunk() && response.getChunk().hasData()) { ByteBuffer byteBuffer = response.getChunk().getData().asReadOnlyByteBuffer(); buffer = new NioDataBuffer(byteBuffer, byteBuffer.remaining()); } Preconditions.checkState(buffer != null, "response should always contain chunk"); } } else { response = mStream.receive(mDataTimeoutMs); if (response != null) { Preconditions.checkState(response.hasChunk() && response.getChunk().hasData(), "response should always contain chunk"); ByteBuffer byteBuffer = response.getChunk().getData().asReadOnlyByteBuffer(); buffer = new NioDataBuffer(byteBuffer, byteBuffer.remaining()); } } if (response == null) { return null; } mPosToRead += buffer.readableBytes(); try { mStream.send(mReadRequest.toBuilder().setOffsetReceived(mPosToRead).build()); } catch (Exception e) { LOG.debug("Failed to send receipt of data to worker {} for request {}: {}.", mAddress, mReadRequest, e.getMessage()); } Preconditions.checkState(mPosToRead - mReadRequest.getOffset() <= mReadRequest.getLength()); return buffer; } private  GrpcDataReader(FileSystemContext context, WorkerNetAddress address,
      ReadRequest readRequest); }
GrpcDataReader implements DataReader { @Override public DataBuffer readChunk() throws IOException { Preconditions.checkState(!mClient.get().isShutdown(), "Data reader is closed while reading data chunks."); DataBuffer buffer = null; ReadResponse response = null; if (mStream instanceof GrpcDataMessageBlockingStream) { DataMessage<ReadResponse, DataBuffer> message = ((GrpcDataMessageBlockingStream<ReadRequest, ReadResponse>) mStream) .receiveDataMessage(mDataTimeoutMs); if (message != null) { response = message.getMessage(); buffer = message.getBuffer(); if (buffer == null && response.hasChunk() && response.getChunk().hasData()) { ByteBuffer byteBuffer = response.getChunk().getData().asReadOnlyByteBuffer(); buffer = new NioDataBuffer(byteBuffer, byteBuffer.remaining()); } Preconditions.checkState(buffer != null, "response should always contain chunk"); } } else { response = mStream.receive(mDataTimeoutMs); if (response != null) { Preconditions.checkState(response.hasChunk() && response.getChunk().hasData(), "response should always contain chunk"); ByteBuffer byteBuffer = response.getChunk().getData().asReadOnlyByteBuffer(); buffer = new NioDataBuffer(byteBuffer, byteBuffer.remaining()); } } if (response == null) { return null; } mPosToRead += buffer.readableBytes(); try { mStream.send(mReadRequest.toBuilder().setOffsetReceived(mPosToRead).build()); } catch (Exception e) { LOG.debug("Failed to send receipt of data to worker {} for request {}: {}.", mAddress, mReadRequest, e.getMessage()); } Preconditions.checkState(mPosToRead - mReadRequest.getOffset() <= mReadRequest.getLength()); return buffer; } private  GrpcDataReader(FileSystemContext context, WorkerNetAddress address,
      ReadRequest readRequest); @Override long pos(); @Override DataBuffer readChunk(); @Override void close(); }
GrpcDataReader implements DataReader { @Override public DataBuffer readChunk() throws IOException { Preconditions.checkState(!mClient.get().isShutdown(), "Data reader is closed while reading data chunks."); DataBuffer buffer = null; ReadResponse response = null; if (mStream instanceof GrpcDataMessageBlockingStream) { DataMessage<ReadResponse, DataBuffer> message = ((GrpcDataMessageBlockingStream<ReadRequest, ReadResponse>) mStream) .receiveDataMessage(mDataTimeoutMs); if (message != null) { response = message.getMessage(); buffer = message.getBuffer(); if (buffer == null && response.hasChunk() && response.getChunk().hasData()) { ByteBuffer byteBuffer = response.getChunk().getData().asReadOnlyByteBuffer(); buffer = new NioDataBuffer(byteBuffer, byteBuffer.remaining()); } Preconditions.checkState(buffer != null, "response should always contain chunk"); } } else { response = mStream.receive(mDataTimeoutMs); if (response != null) { Preconditions.checkState(response.hasChunk() && response.getChunk().hasData(), "response should always contain chunk"); ByteBuffer byteBuffer = response.getChunk().getData().asReadOnlyByteBuffer(); buffer = new NioDataBuffer(byteBuffer, byteBuffer.remaining()); } } if (response == null) { return null; } mPosToRead += buffer.readableBytes(); try { mStream.send(mReadRequest.toBuilder().setOffsetReceived(mPosToRead).build()); } catch (Exception e) { LOG.debug("Failed to send receipt of data to worker {} for request {}: {}.", mAddress, mReadRequest, e.getMessage()); } Preconditions.checkState(mPosToRead - mReadRequest.getOffset() <= mReadRequest.getLength()); return buffer; } private  GrpcDataReader(FileSystemContext context, WorkerNetAddress address,
      ReadRequest readRequest); @Override long pos(); @Override DataBuffer readChunk(); @Override void close();  }