//设置package包名称以及导入依赖的类
package co.teapot.thriftbase

import org.apache.thrift.TServiceClient
import org.apache.thrift.protocol.{TBinaryProtocol, TProtocol}
import org.apache.thrift.transport.TSocket

trait TeapotThriftClient[ClientClass <:  TServiceClient] {

  val server: String
  val port: Int
  val executor = initExecutor(initProtocol())

  
  def timeoutInMillisOption: Option[Int] = None

  def initProtocol(): TProtocol = {
    val transport = timeoutInMillisOption match {
      case Some(timeout) => new TSocket(server, port, timeout)
      case None => new TSocket(server, port)
    }
    transport.open()
    new TBinaryProtocol(transport)
  }

  def initExecutor(protocol: TProtocol): ClientClass

  def getExecutor = executor

  def close(): Unit = {
    getExecutor.getInputProtocol().getTransport().close()
  }
} 

//设置package包名称以及导入依赖的类
package com.squarecode.axia

import java.net.InetSocketAddress
import com.squarecode.axia.repository.user.UserRepository
import com.twitter.finagle.axia.user.UserService
import com.squarecode.axia.controller.UserControllerImpl
import com.twitter.finagle.builder.ServerBuilder
import com.twitter.finagle.thrift.ThriftServerFramedCodec
import org.apache.thrift.protocol.TBinaryProtocol
import com.typesafe.config.ConfigFactory
import com.typesafe.scalalogging.Logger
import org.slf4j.LoggerFactory


object Server extends App {
  val logger = Logger(LoggerFactory.getLogger(this.getClass))

  val config = ConfigFactory.load("application.conf").getConfig("app")
  val host = config.getString("host")
  val port = config.getInt("port")

  logger.info(s"Start server user host:$host port:$port")
  serve(new InetSocketAddress(host, port))
  logger.info("Server user running...")

  def serve(address: InetSocketAddress) = {
    UserRepository.createTable()

    val service = new UserService.FinagledService(new UserControllerImpl, new TBinaryProtocol.Factory())

    ServerBuilder()
      .keepAlive(true)
      .logChannelActivity(true)
      .codec(ThriftServerFramedCodec())
      .bindTo(address)
      .name("User")
      .build(service)
  }
} 

//设置package包名称以及导入依赖的类
package com.squarecode.axia

import java.net.InetSocketAddress
import java.util.logging.Logger

import com.twitter.finagle.axia.user.UserService$FinagleClient
import com.twitter.finagle.builder.ClientBuilder
import com.twitter.finagle.thrift.ThriftClientFramedCodec
import org.apache.thrift.protocol.TBinaryProtocol


object ClientUtilServer {
  def server(address: InetSocketAddress) = {

    val serve = ClientBuilder()
      .hosts(address)
      .codec(ThriftClientFramedCodec())
      .hostConnectionLimit(1)
      .retries(3)
      .logger(Logger.getLogger("http"))
      .build()
    new UserService$FinagleClient(serve, new TBinaryProtocol.Factory())
  }
} 

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

import com.twitter.finagle.Service
import com.twitter.finagle.tracing._
import com.twitter.finagle.util.ByteArrays
import com.twitter.util.Future
import org.apache.thrift.protocol.{TBinaryProtocol, TMessage, TMessageType}
import org.junit.runner.RunWith
import org.mockito.Matchers
import org.mockito.Mockito.when
import org.scalatest.FunSuite
import org.scalatest.junit.JUnitRunner
import org.scalatest.mock.MockitoSugar

@RunWith(classOf[JUnitRunner])
class ThriftServerFramedCodecTest extends FunSuite with MockitoSugar {
  val protocolFactory = new TBinaryProtocol.Factory()

  test("ThriftServerTracingFilter read header correctly") {
    val traceId = TraceId(Some(SpanId(1L)), None, SpanId(2L), Some(true), Flags().setDebug)
    val bufferingTracer = new BufferingTracer
    Trace.letTracer(bufferingTracer) {
      val filter = new TTwitterServerFilter("service", protocolFactory)
  
      val upgradeMsg = new OutputBuffer(protocolFactory)
      upgradeMsg().writeMessageBegin(new TMessage(ThriftTracing.CanTraceMethodName, TMessageType.CALL, 0))
      val options = new thrift.ConnectionOptions
      options.write(upgradeMsg())
      upgradeMsg().writeMessageEnd()
  
      val service = mock[Service[Array[Byte], Array[Byte]]]
      when(service(Matchers.any[Array[Byte]])).thenReturn(Future(Array[Byte]()))
  
      // now finagle knows we can handle the headers
      filter(upgradeMsg.toArray, service)
  
      // let's create a header
      val header = new thrift.RequestHeader
      header.setSpan_id(2L)
      header.setTrace_id(1L)
      header.setSampled(true)
      header.setFlags(1L)
  
      val ignoreMsg = new OutputBuffer(protocolFactory)
      ignoreMsg().writeMessageBegin(new TMessage("ignoreme", TMessageType.CALL, 0))
      new thrift.ConnectionOptions().write(ignoreMsg())
      ignoreMsg().writeMessageEnd()
  
      filter(ByteArrays.concat(OutputBuffer.messageToArray(header, protocolFactory), ignoreMsg.toArray), service)
  
      bufferingTracer.iterator foreach { record =>
        assert(record.traceId == traceId)
        assert(record.traceId.flags == traceId.flags)
      }
    }
  }
} 

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

import com.twitter.finagle.thrift.ThriftServerFramedCodec
import org.apache.thrift.protocol.TBinaryProtocol
import com.twitter.finagle.builder.{ServerBuilder, Server}
import java.net.InetSocketAddress
import com.twitter.util.Future

object ThriftServer {
  def main(args: Array[String]) {
    // Implement the Thrift Interface
    val processor = new Hello.FutureIface {
      def hi() = Future.value("hi")
    }

    // Convert the Thrift Processor to a Finagle Service
    val service = new Hello.FinagledService(processor, new TBinaryProtocol.Factory())

    val server: Server = ServerBuilder()
      .bindTo(new InetSocketAddress(8080))
      .codec(ThriftServerFramedCodec())
      .name("thriftserver")
      .build(service)
  }
} 

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

import com.twitter.finagle.builder.ClientBuilder
import com.twitter.finagle.Service
import java.net.InetSocketAddress
import org.apache.thrift.protocol.TBinaryProtocol
import com.twitter.finagle.thrift.{ThriftClientFramedCodec, ThriftClientRequest}

object ThriftClient {
  def main(args: Array[String]) {
    // Create a raw Thrift client service. This implements the
    // ThriftClientRequest => Future[Array[Byte]] interface.
    val service: Service[ThriftClientRequest, Array[Byte]] = ClientBuilder()
      .hosts(new InetSocketAddress(8080))
      .codec(ThriftClientFramedCodec())
      .hostConnectionLimit(1)
      .build()

    // Wrap the raw Thrift service in a Client decorator. The client
    // provides a convenient procedural interface for accessing the Thrift
    // server.
    val client = new Hello.FinagledClient(service, new TBinaryProtocol.Factory())

    client.hi() onSuccess { response =>
      println("Received response: " + response)
    } ensure {
      service.close()
    }
  }
} 

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

import com.twitter.finagle.thrift.ThriftClientRequest
import java.net.SocketAddress
import org.apache.thrift.protocol.TBinaryProtocol


object ThriftMux 
  extends Client[ThriftClientRequest, Array[Byte]] with ThriftMuxRichClient
  with Server[Array[Byte], Array[Byte]] with ThriftMuxRichServer
{
  protected val protocolFactory = new TBinaryProtocol.Factory()

  def newClient(group: Group[SocketAddress]): ServiceFactory[ThriftClientRequest, Array[Byte]] =
    ThriftMuxClient.newClient(group)

  def serve(addr: SocketAddress, service: ServiceFactory[Array[Byte], Array[Byte]]) =
    ThriftMuxServer.serve(addr, service)
} 

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

import org.jboss.netty.channel.ChannelPipelineFactory
import com.twitter.finagle.{CodecFactory, ClientCodecConfig}
import org.apache.thrift.protocol.{TBinaryProtocol, TProtocolFactory}


  def apply(config: ClientCodecConfig) = {
    new ThriftClientBufferedCodec(new TBinaryProtocol.Factory(), config)
  }
}

class ThriftClientBufferedCodec(protocolFactory: TProtocolFactory, config: ClientCodecConfig)
  extends ThriftClientFramedCodec(protocolFactory, config)
{
  override def pipelineFactory = {
    val framedPipelineFactory = super.pipelineFactory

    new ChannelPipelineFactory {
      def getPipeline() = {
        val pipeline = framedPipelineFactory.getPipeline
        pipeline.replace(
          "thriftFrameCodec", "thriftBufferDecoder",
          new ThriftBufferDecoder(protocolFactory))
        pipeline
      }
    }
  }
} 

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

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


  def apply(config: ServerCodecConfig) = {
    new ThriftServerBufferedCodec(new TBinaryProtocol.Factory(), config)
  }
}

class ThriftServerBufferedCodec(
  protocolFactory: TProtocolFactory,
  config: ServerCodecConfig
)
  extends ThriftServerFramedCodec(config)
{
  override def pipelineFactory = {
    val framedPipelineFactory = super.pipelineFactory

    new ChannelPipelineFactory {
      def getPipeline() = {
        val pipeline = framedPipelineFactory.getPipeline
        pipeline.replace(
          "thriftFrameCodec", "thriftBufferDecoder",
          new ThriftBufferDecoder(protocolFactory))
        pipeline
      }
    }
  }
} 

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

import com.twitter.finagle.Service
import com.twitter.util.Future
import org.specs.SpecificationWithJUnit
import org.specs.mock.Mockito
import org.apache.thrift.protocol.{TMessageType, TMessage, TBinaryProtocol}
import com.twitter.finagle.tracing._
import java.net.InetSocketAddress
import com.twitter.finagle.util.ByteArrays

class ThriftServerFramedCodecSpec extends SpecificationWithJUnit with Mockito {
  val protocolFactory = new TBinaryProtocol.Factory()

  "ThriftServerTracingFilter" should {
    "read header correctly" in {
      val traceId = TraceId(Some(SpanId(1L)), None, SpanId(2L), Some(true), Flags().setDebug)
      val bufferingTracer = new BufferingTracer
      Trace.pushTracer(bufferingTracer)

      val filter = new ThriftServerTracingFilter("service", new InetSocketAddress(0), protocolFactory)

      val upgradeMsg = new OutputBuffer(protocolFactory)
      upgradeMsg().writeMessageBegin(new TMessage(ThriftTracing.CanTraceMethodName, TMessageType.CALL, 0))
      val options = new thrift.ConnectionOptions
      options.write(upgradeMsg())
      upgradeMsg().writeMessageEnd()

      val service = mock[Service[Array[Byte], Array[Byte]]]
      service(any[Array[Byte]]) returns Future(Array[Byte]())

      // now finagle knows we can handle the headers
      filter(upgradeMsg.toArray, service)

      // let's create a header
      val header = new thrift.RequestHeader
      header.setSpan_id(2L)
      header.setTrace_id(1L)
      header.setSampled(true)
      header.setFlags(1L)

      val ignoreMsg = new OutputBuffer(protocolFactory)
      ignoreMsg().writeMessageBegin(new TMessage("ignoreme", TMessageType.CALL, 0))
      new thrift.ConnectionOptions().write(ignoreMsg())
      ignoreMsg().writeMessageEnd()

      filter(ByteArrays.concat(OutputBuffer.messageToArray(header, protocolFactory), ignoreMsg.toArray), service)

      bufferingTracer.iterator foreach { record =>
        record.traceId mustEqual traceId
        record.traceId.flags mustEqual traceId.flags
      }
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.twitter.util

import org.apache.thrift.TBase
import org.apache.thrift.protocol.{TBinaryProtocol, TCompactProtocol, TProtocolFactory}

object ThriftCodec {
  def apply[T <: TBase[_, _]: Manifest, P <: TProtocolFactory: Manifest]: ThriftCodec[T, P] =
    new ThriftCodec[T, P]
}

class ThriftCodec[T <: TBase[_, _]: Manifest, P <: TProtocolFactory: Manifest]
  extends Codec[T, Array[Byte]]
  with ThriftSerializer {

  protected lazy val prototype: T =
    manifest[T].runtimeClass.asInstanceOf[Class[T]].newInstance

  override lazy val protocolFactory: TProtocolFactory =
    manifest[P].runtimeClass.asInstanceOf[Class[P]].newInstance

  override def encode(item: T): Array[Byte] = toBytes(item)

  override def decode(bytes: Array[Byte]): T = {
    val obj = prototype.deepCopy
    fromBytes(obj, bytes)
    obj.asInstanceOf[T]
  }
}

class BinaryThriftCodec[T <: TBase[_, _]: Manifest]
  extends ThriftCodec[T, TBinaryProtocol.Factory]

class CompactThriftCodec[T <: TBase[_, _]: Manifest]
  extends ThriftCodec[T, TCompactProtocol.Factory] 

//设置package包名称以及导入依赖的类
package com.cj.serialization

import com.twitter.scrooge.{ThriftStruct, ThriftStructCodec}
import org.apache.thrift.protocol.TBinaryProtocol
import org.apache.thrift.transport.TMemoryBuffer

package object thrift {

  
  class DeserializeThriftStruct[T <: ThriftStruct](codec : ThriftStructCodec[T])
    extends Deserialize[T] {

    def deserialize(bytes: Array[Byte]): Option[T] = {
      safely({
        codec.decode({
          val output = new TMemoryBuffer(32)
          output.write(bytes)
          new TBinaryProtocol(output)
        })
      })
    }
  }
} 

//设置package包名称以及导入依赖的类
package hello_thrift

import java.net.InetSocketAddress
import com.twitter.finagle.builder.ServerBuilder
import com.twitter.finagle.thrift.ThriftServerFramedCodec
import org.apache.thrift.protocol.TBinaryProtocol

import hello_thrift.recordTag._

object RecordTagServer {
  val handler = new RecordTagHandler

  def buildServer(sockAddr: Int) = {
    val protocol = new TBinaryProtocol.Factory()
    val service = new TagService.FinagledService(new RecordTagHandler, protocol)
    ServerBuilder()
      .codec(ThriftServerFramedCodec())
      .name("hello_thrift")
      .bindTo(new InetSocketAddress(sockAddr))
      .build(service)
  }
}