/*
 * 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.Configuration;
import alluxio.PropertyKey;
import alluxio.network.protocol.RPCMessage;
import alluxio.network.protocol.RPCMessageDecoder;
import alluxio.network.protocol.RPCMessageEncoder;
import alluxio.worker.AlluxioWorkerService;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;

import javax.annotation.concurrent.ThreadSafe;

/**
 * Adds the block server's pipeline into the channel.
 */
@ThreadSafe
final class PipelineHandler extends ChannelInitializer<SocketChannel> {
  private final AlluxioWorkerService mWorker;
  private final FileTransferType mFileTransferType;

  /**
   * @param worker the Alluxio worker
   */
  public PipelineHandler(AlluxioWorkerService worker) {
    mWorker = worker;
    mFileTransferType = Configuration
        .getEnum(PropertyKey.WORKER_NETWORK_NETTY_FILE_TRANSFER_TYPE, FileTransferType.class);
  }

  @Override
  protected void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    pipeline.addLast("frameDecoder", RPCMessage.createFrameDecoder());
    pipeline.addLast("RPCMessageDecoder", new RPCMessageDecoder());
    pipeline.addLast("RPCMessageEncoder", new RPCMessageEncoder());
    pipeline.addLast("dataServerBlockReadHandler",
        new DataServerBlockReadHandler(NettyExecutors.BLOCK_READER_EXECUTOR,
            mWorker.getBlockWorker(), mFileTransferType));
    pipeline.addLast("dataServerUfsBlockReadHandler",
        new DataServerUFSBlockReadHandler(NettyExecutors.UFS_BLOCK_READER_EXECUTOR,
            mWorker.getBlockWorker()));
    pipeline.addLast("dataServerBlockWriteHandler",
        new DataServerBlockWriteHandler(NettyExecutors.BLOCK_WRITER_EXECUTOR,
            mWorker.getBlockWorker()));
    pipeline.addLast("dataServerFileWriteHandler",
        new DataServerUFSFileWriteHandler(NettyExecutors.FILE_WRITER_EXECUTOR));
  }
}

@@ -1601,16 +1601,44 @@
 eline();
+%0A%0A    // Decoders & Encoders
 %0A    pip
@@ -1828,32 +1828,55 @@
 ssageEncoder());
+%0A%0A    // Block Handlers
 %0A    pipeline.ad
@@ -1906,32 +1906,24 @@
 eadHandler%22,
-%0A       
  new DataSer
@@ -1934,32 +1934,41 @@
 lockReadHandler(
+%0A        
 NettyExecutors.B
@@ -1984,36 +1984,24 @@
 ER_EXECUTOR,
-%0A           
  mWorker.get
@@ -2072,20 +2072,18 @@
 rver
-Ufs
 Block
-Read
+Write
 Hand
@@ -2083,32 +2083,24 @@
 iteHandler%22,
-%0A       
  new DataSer
@@ -2106,28 +2106,26 @@
 rver
-UFS
 Block
-Read
+Write
 Handler(
 Nett
@@ -2112,32 +2112,41 @@
 ockWriteHandler(
+%0A        
 NettyExecutors.U
@@ -2148,22 +2148,18 @@
 ors.
-UFS_
 BLOCK_
-READ
+WRIT
 ER_E
@@ -2162,36 +2162,24 @@
 ER_EXECUTOR,
-%0A           
  mWorker.get
@@ -2186,32 +2186,53 @@
 BlockWorker()));
+%0A%0A    // UFS Handlers
 %0A    pipeline.ad
@@ -2244,34 +2244,36 @@
 (%22dataServer
+Ufs
 Block
-Write
+Read
 Handler%22,%0A  
@@ -2265,32 +2265,24 @@
 eadHandler%22,
-%0A       
  new DataSer
@@ -2284,34 +2284,36 @@
 taServer
+UFS
 Block
-Write
+Read
 Handler(
 NettyExe
@@ -2296,32 +2296,41 @@
 lockReadHandler(
+%0A        
 NettyExecutors.B
@@ -2324,34 +2324,38 @@
 tyExecutors.
+UFS_
 BLOCK_
-WRIT
+READ
 ER_EXECUTOR,
@@ -2358,20 +2358,8 @@
 TOR,
-%0A           
  mWo
@@ -2415,16 +2415,19 @@
 taServer
+Ufs
 FileWrit
@@ -2436,24 +2436,16 @@
 andler%22,
-%0A       
  new Dat
@@ -2471,16 +2471,25 @@
 Handler(
+%0A        
 NettyExe

Improve style.