/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.worker.netty;

import alluxio.metrics.MetricsSystem;
import alluxio.network.protocol.RPCProtoMessage;
import alluxio.proto.dataserver.Protocol;
import alluxio.underfs.UnderFileSystem;
import alluxio.worker.file.FileSystemWorker;

import com.codahale.metrics.Counter;
import io.netty.buffer.ByteBuf;

import java.io.IOException;
import java.io.OutputStream;
import java.util.concurrent.ExecutorService;

import javax.annotation.concurrent.NotThreadSafe;

/**
 * This handler handles writes to a file in the under file system. Due to the semantics enforced
 * on under file systems, the client must write all the data of a file through the same stream to
 * the under file system. This prevents us from handling the writes at a block level.
 *
 * For more information about the implementation of read/write buffering, see
 * {@link DataServerWriteHandler}.
 *
 * For more information about the implementation of the client side writer, see
 * UnderFileSystemFileOutStream.
 */
@NotThreadSafe
final class DataServerUFSFileWriteHandler extends DataServerWriteHandler {
  /** Filesystem worker which handles file level operations for the worker. */
  private final FileSystemWorker mWorker;
  private static final long UNUSED_SESSION_ID = -1;

  private class FileWriteRequestInternal extends WriteRequestInternal {
    private final UnderFileSystem mUnderFileSystem;
    OutputStream mOutputStream;

    FileWriteRequestInternal(Protocol.WriteRequest request) throws Exception {
      super(request.getId(), UNUSED_SESSION_ID);
      mUnderFileSystem = UnderFileSystem.Factory.get(request.get)
      mOutputStream = mWorker.getUfsOutputStream(mId);
    }

    @Override
    public void close() throws IOException {}

    @Override
    void cancel() throws IOException {}
  }

  /**
   * Creates an instance of {@link DataServerUFSFileWriteHandler}.
   *
   * @param executorService the executor service to run {@link PacketWriter}s
   * @param worker the file system worker
   */
  DataServerUFSFileWriteHandler(ExecutorService executorService, FileSystemWorker worker) {
    super(executorService);
    mWorker = worker;
  }

  @Override
  protected boolean acceptMessage(Object object) {
    if (!super.acceptMessage(object)) {
      return false;
    }
    Protocol.WriteRequest request = ((RPCProtoMessage) object).getMessage().getMessage();
    return request.getType() == Protocol.RequestType.UFS_FILE;
  }

  /**
   * Initializes the handler if necessary.
   *
   * @param msg the block write request
   * @throws Exception if it fails to initialize
   */
  @Override
  protected void initializeRequest(RPCProtoMessage msg) throws Exception {
    super.initializeRequest(msg);
    if (mRequest == null) {
      mRequest = new FileWriteRequestInternal(msg.getMessage().<Protocol.WriteRequest>getMessage());
    }
  }

  @Override
  protected void writeBuf(ByteBuf buf, long pos) throws Exception {
    buf.readBytes(((FileWriteRequestInternal) mRequest).mOutputStream, buf.readableBytes());
  }

  @Override
  protected void incrementMetrics(long bytesWritten) {
    Metrics.BYTES_WRITTEN_UFS.inc(bytesWritten);
  }

  /**
   * Class that contains metrics for BlockDataServerHandler.
   */
  private static final class Metrics {
    private static final Counter BYTES_WRITTEN_UFS = MetricsSystem.workerCounter("BytesWrittenUFS");

    private Metrics() {
    } // prevent instantiation
  }
}

@@ -1592,129 +1592,8 @@
 r %7B%0A
-  /** Filesystem worker which handles file level operations for the worker. */%0A  private final FileSystemWorker mWorker;%0A
   pr
@@ -1713,16 +1713,48 @@
 ernal %7B%0A
+    private final String mPath;%0A
     priv
@@ -1801,16 +1801,30 @@
 em;%0A    
+private final 
 OutputSt
@@ -1972,16 +1972,52 @@
 ON_ID);%0A
+      mPath = request.getUfsPath();%0A
       mU
@@ -2065,20 +2065,15 @@
 get(
-request.get)
+mPath);
 %0A   
@@ -2096,37 +2096,36 @@
  = m
-Worker.getUfsOutputStream(mId
+UnderFileSystem.create(mPath
 );%0A 
@@ -2188,16 +2188,50 @@
 eption %7B
+%0A      mOutputStream.close();%0A    
 %7D%0A%0A    @
@@ -2277,16 +2277,161 @@
 eption %7B
+%0A      // TODO(calvin): Consider adding cancel to the ufs stream api.%0A      mOutputStream.close();%0A      mUnderFileSystem.deleteFile(mPath);%0A    
 %7D%0A  %7D%0A%0A 
@@ -2757,30 +2757,8 @@
 e);%0A
-    mWorker = worker;%0A
   %7D%0A

Update server side ufs write to only need data path.