//设置package包名称以及导入依赖的类
package akka.remote.testconductor

import org.jboss.netty.channel.{ Channel, ChannelPipeline, ChannelPipelineFactory, ChannelUpstreamHandler, SimpleChannelUpstreamHandler, DefaultChannelPipeline }
import org.jboss.netty.channel.socket.nio.{ NioClientSocketChannelFactory, NioServerSocketChannelFactory }
import org.jboss.netty.bootstrap.{ ClientBootstrap, ServerBootstrap }
import org.jboss.netty.handler.codec.frame.{ LengthFieldBasedFrameDecoder, LengthFieldPrepender }
import org.jboss.netty.handler.codec.compression.{ ZlibDecoder, ZlibEncoder }
import org.jboss.netty.handler.codec.protobuf.{ ProtobufDecoder, ProtobufEncoder }
import org.jboss.netty.handler.timeout.{ ReadTimeoutHandler, ReadTimeoutException }
import java.net.InetSocketAddress
import java.util.concurrent.Executors
import akka.event.Logging
import akka.util.Helpers


private[akka] object RemoteConnection {
  def apply(role: Role, sockaddr: InetSocketAddress, poolSize: Int, handler: ChannelUpstreamHandler): Channel = {
    role match {
      case Client ?
        val socketfactory = new NioClientSocketChannelFactory(Executors.newCachedThreadPool, Executors.newCachedThreadPool,
          poolSize)
        val bootstrap = new ClientBootstrap(socketfactory)
        bootstrap.setPipelineFactory(new TestConductorPipelineFactory(handler))
        bootstrap.setOption("tcpNoDelay", true)
        bootstrap.connect(sockaddr).getChannel
      case Server ?
        val socketfactory = new NioServerSocketChannelFactory(Executors.newCachedThreadPool, Executors.newCachedThreadPool,
          poolSize)
        val bootstrap = new ServerBootstrap(socketfactory)
        bootstrap.setPipelineFactory(new TestConductorPipelineFactory(handler))
        bootstrap.setOption("reuseAddress", !Helpers.isWindows)
        bootstrap.setOption("child.tcpNoDelay", true)
        bootstrap.bind(sockaddr)
    }
  }

  def getAddrString(channel: Channel) = channel.getRemoteAddress match {
    case i: InetSocketAddress ? i.toString
    case _                    ? "[unknown]"
  }

  def shutdown(channel: Channel) =
    try channel.close() finally try channel.getFactory.shutdown() finally channel.getFactory.releaseExternalResources()
} 

//设置package包名称以及导入依赖的类
package com.twitter.finagle.thrift.transport.netty3

import org.jboss.netty.channel.{ChannelPipeline, ChannelPipelineFactory, Channels}


private[finagle]
object ThriftClientFramedPipelineFactory extends ChannelPipelineFactory {
  def getPipeline(): ChannelPipeline = {
    val pipeline = Channels.pipeline()
    pipeline.addLast("thriftFrameCodec", new ThriftFrameCodec)
    pipeline.addLast("byteEncoder",      new ThriftClientChannelBufferEncoder)
    pipeline.addLast("byteDecoder",      new ThriftChannelBufferDecoder)
    pipeline
  }
} 

//设置package包名称以及导入依赖的类
package com.twitter.finagle.thrift.transport.netty3

import com.twitter.finagle.thrift.{Protocols, ThriftServerFramedCodec}
import com.twitter.finagle.{CodecFactory, ServerCodecConfig}
import org.apache.thrift.protocol.TProtocolFactory
import org.jboss.netty.channel.{ChannelPipeline, ChannelPipelineFactory}

private[netty3]
case class ThriftServerBufferedPipelineFactory(protocolFactory: TProtocolFactory)
    extends ChannelPipelineFactory {

  def getPipeline(): ChannelPipeline = {
    val pipeline = ThriftServerFramedPipelineFactory.getPipeline()
    pipeline.replace(
      "thriftFrameCodec", "thriftBufferDecoder",
      new ThriftBufferedTransportDecoder(protocolFactory))
    pipeline
  }
}


  def apply(config: ServerCodecConfig): ThriftServerBufferedCodec = {
    new ThriftServerBufferedCodec(protocolFactory, config)
  }
}

class ThriftServerBufferedCodec(protocolFactory: TProtocolFactory, config: ServerCodecConfig)
    extends ThriftServerFramedCodec(config) {
  override def pipelineFactory = ThriftServerBufferedPipelineFactory(protocolFactory)
} 

//设置package包名称以及导入依赖的类
package com.twitter.finagle.thrift.transport.netty3

import org.apache.thrift.protocol.TProtocolFactory
import org.jboss.netty.channel.{ChannelPipeline, ChannelPipelineFactory}

private[netty3] case class ThriftClientBufferedPipelineFactory(
    protocolFactory: TProtocolFactory)
  extends ChannelPipelineFactory {
  def getPipeline(): ChannelPipeline = {
    val pipeline = ThriftClientFramedPipelineFactory.getPipeline()
    pipeline.replace(
      "thriftFrameCodec", "thriftBufferDecoder",
      new ThriftBufferedTransportDecoder(protocolFactory))
    pipeline
  }
}