@@ -0,0 +1,3297 @@
+/*%0A * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0%0A * (the %E2%80%9CLicense%E2%80%9D). You may not use this work except in compliance with the License, which is%0A * available at www.apache.org/licenses/LICENSE-2.0%0A *%0A * This software is distributed on an %22AS IS%22 basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,%0A * either express or implied, as more fully set forth in the License.%0A *%0A * See the NOTICE file distributed with this work for information regarding copyright ownership.%0A */%0A%0Apackage alluxio.client.block;%0A%0Aimport alluxio.AlluxioURI;%0Aimport alluxio.client.file.FileSystemContext;%0Aimport alluxio.client.file.FileSystemWorkerClient;%0Aimport alluxio.client.file.UnderFileSystemFileInStream;%0Aimport alluxio.client.file.options.CloseUfsFileOptions;%0Aimport alluxio.client.file.options.OpenUfsFileOptions;%0Aimport alluxio.exception.AlluxioException;%0Aimport alluxio.exception.ExceptionMessage;%0A%0Aimport java.io.IOException;%0A%0A/**%0A * Implementation of %7B@link UnderStoreBlockInStream%7D which accesses the under storage through an%0A * Alluxio worker. Note that the seek implementation of this class is not fail fast, and seeking%0A * past the end of the file will be successful until a subsequent read occurs.%0A */%0Apublic class DelegatedUnderStoreBlockInStream extends UnderStoreBlockInStream %7B%0A  /** File System Worker Client */%0A  private final FileSystemWorkerClient mClient;%0A  /** File id of the ufs file */%0A  private final long mUfsFileId;%0A%0A  /**%0A   * Constructor for a direct under store block in stream.%0A   *%0A   * @param initPos position in the file%0A   * @param length length of the block%0A   * @param fileBlockSize file block size%0A   * @param ufsPath path in the ufs%0A   * @throws IOException if an error occurs initializing the stream to the ufs file%0A   */%0A  public DelegatedUnderStoreBlockInStream(long initPos, long length, long fileBlockSize,%0A      String ufsPath) throws IOException %7B%0A    super(initPos, length, fileBlockSize, ufsPath);%0A    mClient = FileSystemContext.INSTANCE.acquireWorkerClient();%0A    try %7B%0A      mUfsFileId = mClient.openUfsFile(new AlluxioURI(ufsPath), OpenUfsFileOptions.defaults());%0A    %7D catch (AlluxioException e) %7B%0A      mClient.close();%0A      throw new IOException(e);%0A    %7D%0A    setUnderStoreStream(0);%0A  %7D%0A%0A  @Override%0A  public void close() throws IOException %7B%0A    mUnderStoreStream.close();%0A    try %7B%0A      mClient.closeUfsFile(mUfsFileId, CloseUfsFileOptions.defaults());%0A    %7D catch (AlluxioException e) %7B%0A      throw new IOException(e);%0A    %7D finally %7B%0A      mClient.close();%0A    %7D%0A  %7D%0A%0A  @Override%0A  protected void setUnderStoreStream(long pos) throws IOException %7B%0A    if (mUnderStoreStream != null) %7B%0A      mUnderStoreStream.close();%0A    %7D%0A    if (pos %3C 0 %7C%7C pos %3E mLength) %7B%0A      throw new IOException(ExceptionMessage.FAILED_SEEK.getMessage(pos));%0A    %7D%0A    mUnderStoreStream =%0A        new UnderFileSystemFileInStream(mClient.getWorkerDataServerAddress(), mUfsFileId);%0A    // The stream is at the beginning of the file, so skip to the correct absolute position.%0A    if ((mInitPos + pos) != 0 && mInitPos + pos != mUnderStoreStream.skip(mInitPos + pos)) %7B%0A      throw new IOException(ExceptionMessage.FAILED_SKIP.getMessage(pos));%0A    %7D%0A    // Set the current block position to the specified block position.%0A    mPos = pos;%0A  %7D%0A%7D%0A

Implement delegated ufs block in stream.