func (p *fakeServer) Dispatch(r *Request) error {
	log.Printf("Request SeqId: %d, MethodName: %s\n", r.Request.SeqId, r.Request.Name)
	r.Wait.Add(1)
	go func() {
		time.Sleep(time.Millisecond)
		r.Response.Data = []byte(string(r.Request.Data))

		typeId, _, seqId, _ := DecodeThriftTypIdSeqId(r.Response.Data)
		log.Printf(Green("TypeId: %d, SeqId: %d\n"), typeId, seqId)
		r.Wait.Done()
	}()
	//	r.RestoreSeqId()
	//	r.Wait.Done()
	return nil
}

//
// 确保Socket成功连接到后端服务器
//
func (bc *BackendConn) ensureConn() (transport thrift.TTransport, err error) {
	// 1. 创建连接(只要IP没有问题， err一般就是空)
	timeout := time.Second * 5
	if strings.Contains(bc.addr, ":") {
		transport, err = thrift.NewTSocketTimeout(bc.addr, timeout)
	} else {
		transport, err = rpc_utils.NewTUnixDomainTimeout(bc.addr, timeout)
	}
	log.Printf(Cyan("[%s]Create Socket To: %s"), bc.service, bc.addr)

	if err != nil {
		log.ErrorErrorf(err, "[%s]Create Socket Failed: %v, Addr: %s", err, bc.service, bc.addr)
		// 连接不上，失败
		return nil, err
	}

	// 2. 只要服务存在，一般不会出现err
	sleepInterval := 1
	err = transport.Open()
	for err != nil && !bc.IsMarkOffline.Get() {
		log.ErrorErrorf(err, "[%s]Socket Open Failed: %v, Addr: %s", bc.service, err, bc.addr)

		// Sleep: 1, 2, 4这几个间隔
		time.Sleep(time.Duration(sleepInterval) * time.Second)

		if sleepInterval < 4 {
			sleepInterval *= 2
		}
		err = transport.Open()
	}
	return transport, err
}

// 配对 Request, resp, err
// PARAM: resp []byte 为一帧完整的thrift数据包
func (bc *BackendConnLB) setResponse(r *Request, data []byte, err error) error {
	//	log.Printf("#setResponse:  data: %v", data)
	// 表示出现错误了
	if data == nil {
		log.Printf("No Data From Server, error: %v\n", err)
		r.Response.Err = err
	} else {
		// 从resp中读取基本的信息
		typeId, method, seqId, err := DecodeThriftTypIdSeqId(data)

		// 解码错误，直接报错
		if err != nil {
			log.ErrorErrorf(err, "Decode SeqId Error: %v", err)
			return err
		}

		if typeId == MESSAGE_TYPE_STOP {
			// 不再接受新的输入
			// 直接来自后端的服务(不遵循: Request/Reply模型)
			bc.MarkConnActiveFalse()
			return nil
		}

		// 找到对应的Request

		req := bc.seqNumRequestMap.Pop(seqId)

		// 如果是心跳，则OK
		if typeId == MESSAGE_TYPE_HEART_BEAT {
			bc.hbLastTime.Set(time.Now().Unix())
			return nil
		}

		if req == nil {
			log.Errorf("#setResponse not found, seqId: %d", seqId)
			return nil
		} else {

			if req.Response.SeqId != seqId {
				log.Errorf("Data From Server, SeqId not match, Ex: %d, Ret: %d", req.Request.SeqId, seqId)
			}
			r = req
			r.Response.TypeId = typeId
			if req.Request.Name != method {
				data = nil
				err = req.NewInvalidResponseError(method, "conn_lb")
			}
		}
	}

	r.Response.Data, r.Response.Err = data, err
	// 还原SeqId
	if data != nil {
		r.RestoreSeqId()
	}

	r.Wait.Done()
	return err
}

func (s *BackService) StateChanged(conn *BackendConn) {
	//	log.Printf(Cyan("[%s]StateChanged: %s, Index: %d, Count: %d, IsConnActive: %t"),
	//		s.serviceName, conn.addr, conn.Index, len(s.activeConns),
	//		conn.IsConnActive.Get())

	s.activeConnsLock.Lock()
	defer s.activeConnsLock.Unlock()

	if conn.IsConnActive.Get() {
		// 上线: BackendConn
		log.Printf(Cyan("[%s]MarkConnActiveOK: %s, Index: %d, Count: %d"),
			s.serviceName, conn.addr, conn.Index, len(s.activeConns))

		if conn.Index == INVALID_ARRAY_INDEX {
			conn.Index = len(s.activeConns)
			s.activeConns = append(s.activeConns, conn)

			log.Printf(Green("[%s]Add BackendConn to activeConns: %s, Total Actives: %d"),
				s.serviceName, conn.Addr(), len(s.activeConns))
		}
	} else {
		// 下线BackendConn(急速执行
		connIndex := conn.Index
		if conn.Index != INVALID_ARRAY_INDEX {
			lastIndex := len(s.activeConns) - 1

			// 将最后一个元素和当前的元素交换位置
			if lastIndex != conn.Index {

				lastConn := s.activeConns[lastIndex]
				s.activeConns[conn.Index] = lastConn
				lastConn.Index = conn.Index
			}

			s.activeConns[lastIndex] = nil
			conn.Index = INVALID_ARRAY_INDEX

			// slice
			s.activeConns = s.activeConns[0:lastIndex]
			log.Printf(Red("[%s]Remove BackendConn From activeConns: %s, Remains: %d"),
				s.serviceName, conn.Addr(), len(s.activeConns))
		}
		log.Printf(Red("[%s]Remove BackendConn From activeConns: %s, Index: %d"),
			s.serviceName, conn.Addr(), connIndex)
	}
}

//
// 后端如何处理一个Request
//
func (s *Router) Dispatch(r *Request) error {
	backService := s.GetBackService(r.Service)
	if backService == nil {
		log.Printf(Cyan("Service Not Found for: %s.%s\n"), r.Service, r.Request.Name)
		r.Response.Data = GetServiceNotFoundData(r)
		return nil
	} else {
		return backService.HandleRequest(r)
	}
}

//
// MarkOffline发生场景:
// 1. 后端服务即将下线，预先通知
// 2. 后端服务已经挂了，zk检测到
//
// BackendConn 在这里暂时理解关闭conn, 而是从 backend_service_proxy中下线当前的conn,
// 然后conn的关闭根据 心跳&Conn的读写异常来判断; 因此 IsConnActive = false 情况下，心跳不能关闭
//
func (bc *BackendConn) MarkOffline() {
	if !bc.IsMarkOffline.Get() {
		log.Printf(Magenta("[%s]BackendConn: %s MarkOffline"), bc.service, bc.addr)
		bc.IsMarkOffline.Set(true)

		// 不再接受(来自backend_service_proxy的)新的输入
		bc.MarkConnActiveFalse()

		close(bc.input)
	}
}

func (bc *BackendConn) MarkConnActiveFalse() {
	if bc.IsConnActive.Get() {
		// 从Active切换到非正常状态
		bc.IsConnActive.Set(false)

		if bc.delegate != nil {
			bc.delegate.StateChanged(bc) // 通知其他人状态出现问题
		}

		// 日志延后, 控制信息尽快生效
		log.Printf(Red("[%s]MarkConnActiveFalse: %s, %p"), bc.service, bc.addr, bc.delegate)
	}
}

// run之间 transport刚刚建立，因此服务的可靠性比较高
func (bc *BackendConnLB) Run() {
	log.Printf(Green("[%s]Add New BackendConnLB: %s"), bc.serviceName, bc.address)

	// 1. 首先BackendConn将当前 input中的数据写到后端服务中
	err := bc.loopWriter()

	// 2. 从Active切换到非正常状态, 同时不再从backend_service_lb接受新的任务
	//    可能出现异常，也可能正常退出(反正不干活了)
	bc.MarkConnActiveFalse()

	log.Printf(Red("[%s]Remove Faild BackendConnLB: %s"), bc.serviceName, bc.address)

	if err == nil {
		// bc.input被关闭了，应该就没有 Request 了
	} else {
		// 如果出现err, 则将bc.input中现有的数据都flush回去（直接报错)
		for i := len(bc.input); i != 0; i-- {
			r := <-bc.input
			bc.setResponse(r, nil, err)
		}
	}

}

func NewThriftRpcServer(config *Config, processor thrift.TProcessor) *ThriftRpcServer {
	log.Printf("FrontAddr: %s\n", Magenta(config.FrontendAddr))

	return &ThriftRpcServer{
		config:       config,
		ZkAddr:       config.ZkAddr,
		ProductName:  config.ProductName,
		ServiceName:  config.Service,
		FrontendAddr: config.FrontendAddr,
		Verbose:      config.Verbose,
		Processor:    processor,
	}

}

func NewNonBlockSessionSize(c thrift.TTransport, address string, verbose bool,
	lastRequestTime *atomic2.Int64, bufsize int, timeout int) *NonBlockSession {
	s := &NonBlockSession{
		RemoteAddress:            address,
		lastRequestTime:          lastRequestTime,
		verbose:                  verbose,
		TBufferedFramedTransport: NewTBufferedFramedTransport(c, time.Microsecond*100, 20),
	}

	// 还是基于c net.Conn进行读写，只是采用Redis协议进行编码解码
	// Reader 处理Client发送过来的消息
	// Writer 将后端服务的数据返回给Client
	log.Printf(Green("Session From Proxy [%s] created"), address)
	return s
}

//
// 先写入数据，然后再Flush Transport
//
func (p *TBufferedFramedTransport) FlushBuffer(force bool) error {
	size := p.Buffer.Len()

	// 没有有效的数据，直接返回
	if size == 0 {
		return nil
	}
	// TODO: 待优化
	force = true

	// 1. 将p.buf的大小以BigEndian模式写入: buf中
	buf := p.LenghW[:4]
	binary.BigEndian.PutUint32(buf, uint32(size))

	//	log.Printf("----> Frame Size: %d, %v\n", size, buf)
	// 然后transport中先写入: 长度信息
	_, err := p.Writer.Write(buf)
	if err != nil {
		return thrift.NewTTransportExceptionFromError(err)
	}

	// 2. 然后继续写入p.buf中的数据
	if size > 0 {
		var (
			n   int64
			err error
		)
		// 如果 err == io.ErrShortWrite， p.Writer中也有buffer, 因此可以不用考虑异常
		if n, err = p.Buffer.WriteTo(p.Writer); err != nil {
			log.ErrorErrorf(err, "Error Flushing Expect Write: %d, but %d\n",
				size, n)
			return thrift.NewTTransportExceptionFromError(err)
		}
		if n < int64(size) {
			log.Printf(Red("Buffer Write Not Finished"))
		}
	}

	p.nbuffered++

	// Buffer重新开始处理数据
	p.Buffer.Reset()

	// Flush Buffer
	return p.flushTransport(force)
}

// 处理所有的等待中的请求
func (bc *BackendConnLB) flushRequests(err error) {
	// 告诉BackendService, 不再接受新的请求
	bc.MarkConnActiveFalse()

	seqRequest := bc.seqNumRequestMap.Purge()

	for _, request := range seqRequest {
		if request.Request.TypeId == MESSAGE_TYPE_HEART_BEAT {
			// 心跳出错了，则直接直接跳过
		} else {
			log.Printf(Red("Handle Failed Request: %s.%s"), request.Service, request.Request.Name)
			request.Response.Err = err
			request.Wait.Done()
		}
	}

	// 关闭输入
	close(bc.input)

}

//
// 不断建立到后端的逻辑，负责: BackendConn#input到redis的数据的输入和返回
//
func (bc *BackendConn) Run() {

	for k := 0; !bc.IsMarkOffline.Get(); k++ {

		// 1. 首先BackendConn将当前 input中的数据写到后端服务中
		transport, err := bc.ensureConn()
		if err != nil {
			log.ErrorErrorf(err, "[%s]BackendConn#ensureConn error: %v", bc.service, err)
			return
		}

		connOver := &sync.WaitGroup{}
		c := NewTBufferedFramedTransport(transport, 100*time.Microsecond, 20)

		bc.MarkConnActiveOK() // 准备接受数据
		connOver.Add(1)
		bc.loopReader(c, connOver) // 异步(读取来自后端服务器的返回数据)
		// 2. 将 bc.input 中的请求写入 后端的Rpc Server
		err = bc.loopWriter(c) // 同步

		// 3. 停止接受Request
		bc.MarkConnActiveFalse()

		// 等待Conn正式关闭
		connOver.Wait()

		// 4. 将bc.input中剩余的 Request直接出错处理
		if err == nil {
			log.Printf(Red("[%s]BackendConn#loopWriter normal Exit..."), bc.service)
			break
		} else {
			// 对于尚未处理的Request, 直接报错
			for i := len(bc.input); i != 0; i-- {
				r := <-bc.input
				bc.setResponse(r, nil, err)
			}
		}
	}
}

// 处理来自Client的请求
func (s *Session) handleRequest(request []byte, d Dispatcher) (*Request, error) {
	// 构建Request
	if s.verbose {
		log.Printf("HandleRequest: %s", string(request))
	}
	r, err := NewRequest(request, true)
	if err != nil {
		return r, err
	}

	// 增加统计
	s.LastOpUnix = time.Now().Unix()
	s.Ops++
	if r.Request.TypeId == MESSAGE_TYPE_HEART_BEAT {
		HandleProxyPingRequest(r) // 直接返回数据
		return r, nil
	}

	// 交给Dispatch
	// Router
	return r, d.Dispatch(r)
}

func (s *BackService) Stop() {
	// 标志停止
	s.stop.Set(true)
	// 触发一个事件（之后ServiceNodes也不再监控)
	s.evtbus <- true
	go func() {
		// TODO:
		for true {
			now := time.Now().Unix()
			if now-s.lastRequestTime.Get() > 10 {
				break
			} else {
				time.Sleep(time.Second)
			}
		}
		for len(s.activeConns) > 0 {
			s.activeConns[0].MarkOffline()
		}

		log.Printf(Red("Mark All Connections Off: %s"), s.serviceName)

	}()
}

func NewThriftLoadBalanceServer(config *Config) *ThriftLoadBalanceServer {
	log.Printf("FrontAddr: %s\n", Magenta(config.FrontendAddr))

	// 前端对接rpc_proxy
	p := &ThriftLoadBalanceServer{
		config:       config,
		zkAddr:       config.ZkAddr,
		productName:  config.ProductName,
		serviceName:  config.Service,
		frontendAddr: config.FrontendAddr,
		backendAddr:  config.BackAddr,
		verbose:      config.Verbose,
		exitEvt:      make(chan bool),
	}

	p.topo = NewTopology(p.productName, p.zkAddr)
	p.lbServiceName = GetServiceIdentity(p.frontendAddr)

	// 后端对接: 各种python的rpc server
	p.backendService = NewBackServiceLB(p.serviceName, p.backendAddr, p.verbose,
		p.config.FalconClient, p.exitEvt)
	return p

}

// 创建一个BackService
func NewBackService(productName string, serviceName string, topo *Topology, verbose bool) *BackService {

	service := &BackService{
		productName: productName,
		serviceName: serviceName,
		activeConns: make([]*BackendConn, 0, 10),
		addr2Conn:   make(map[string]*BackendConn),
		topo:        topo,
		verbose:     verbose,
	}

	service.WatchBackServiceNodes()

	go func() {
		for !service.stop.Get() {
			log.Printf(Blue("[Report]: %s --> %d backservice, coroutine: %d"),
				service.serviceName, service.Active(), runtime.NumGoroutine())
			time.Sleep(time.Second * 10)
		}
	}()

	return service

}

func RpcMain(binaryName string, serviceDesc string, configCheck ConfigCheck,
	serverFactory ServerFactorory, buildDate string, gitVersion string) {

	// 1. 准备解析参数
	usage = fmt.Sprintf(usage, binaryName, binaryName)

	version := fmt.Sprintf("Version: %s\nBuildDate: %s\nDesc: %s\nAuthor: [email protected]", gitVersion, buildDate, serviceDesc)
	args, err := docopt.Parse(usage, nil, true, version, true)
	if err != nil {
		fmt.Println(err)
		os.Exit(1)
	}

	if s, ok := args["-V"].(bool); ok && s {
		fmt.Println(Green(version))
		os.Exit(1)
	}

	// 这就是为什么 Codis 傻乎乎起一个 http server的目的
	if s, ok := args["--profile-addr"].(string); ok && len(s) > 0 {
		go func() {
			log.Printf(Red("Profile Address: %s"), s)
			log.Println(http.ListenAndServe(s, nil))
		}()
	}

	// 2. 解析Log相关的配置
	log.SetLevel(log.LEVEL_INFO)

	var maxKeepDays int = 3
	if s, ok := args["--log-keep-days"].(string); ok && s != "" {
		v, err := strconv.ParseInt(s, 10, 32)
		if err != nil {
			log.PanicErrorf(err, "invalid max log file keep days = %s", s)
		}
		maxKeepDays = int(v)
	}

	// set output log file
	if s, ok := args["-L"].(string); ok && s != "" {
		f, err := log.NewRollingFile(s, maxKeepDays)
		if err != nil {
			log.PanicErrorf(err, "open rolling log file failed: %s", s)
		} else {
			defer f.Close()
			log.StdLog = log.New(f, "")
		}
	}
	log.SetLevel(log.LEVEL_INFO)
	log.SetFlags(log.Flags() | log.Lshortfile)

	// set log level
	if s, ok := args["--log-level"].(string); ok && s != "" {
		SetLogLevel(s)
	}

	// 没有就没有
	workDir, _ := args["--work-dir"].(string)
	codeUrlVersion, _ := args["--code-url-version"].(string)
	if len(workDir) == 0 {
		workDir, _ = os.Getwd()
	}

	log.Printf("WorkDir: %s, CodeUrl: %s, Wd: %s", workDir, codeUrlVersion)

	// 3. 解析Config
	configFile := args["-c"].(string)
	conf, err := LoadConf(configFile)
	if err != nil {
		log.PanicErrorf(err, "load config failed")
	}

	// 额外的配置信息
	conf.WorkDir = workDir
	conf.CodeUrlVersion = codeUrlVersion

	if configCheck != nil {
		configCheck(conf)
	} else {
		log.Panic("No Config Check Given")
	}
	// 每次启动的时候都打印版本信息
	log.Infof(Green("-----------------\n%s\n--------------------------------------------------------------------"), version)

	// 启动服务
	server := serverFactory(conf)
	server.Run()
}

//.........这里部分代码省略.........
	if err != nil {
		log.ErrorErrorf(err, "Server Socket Create Failed: %v", err)
		panic(fmt.Sprintf("Binding Error FrontendAddress: %s", p.frontendAddr))
	}

	ch := make(chan thrift.TTransport, 4096)
	defer close(ch)

	// 等待后端服务起来
	waitTicker := time.NewTicker(time.Second)

	// 等待上线采用的策略:
	// 1. 检测到有效的Worker注册之后，再等5s即可像zk注册; 避免了Worker没有连接上来，就有请求过来
	// 2. 一旦注册之后，就不再使用该策略；避免服务故障时，lb频繁更新zk, 导致proxy等频繁读取zk
START_WAIT:
	for true {
		select {
		case <-waitTicker.C:
			if p.backendService.Active() <= 0 {
				log.Infof("Sleep Waiting for back Service to Start")
				time.Sleep(time.Second)
			} else {
				break START_WAIT
			}
		case <-exitSignal:
			// 直接退出
			transport.Interrupt()
			transport.Close()
			return
		}
	}

	log.Infof("Stop Waiting")
	// 停止: waitTicker, 再等等就继续了
	waitTicker.Stop()
	time.Sleep(time.Second * 5)

	log.Infof("Begin to Reg To Zk...")
	state.Set(true)
	stateChan <- true

	// 强制退出? TODO: Graceful退出
	go func() {
		<-exitSignal

		// 通知RegisterService终止循环
		evtExit <- true
		log.Info(Green("Receive Exit Signals...."))
		serviceEndpoint.DeleteServiceEndpoint(p.topo)

		start := time.Now().Unix()
		for true {
			// 如果5s内没有接受到新的请求了，则退出
			now := time.Now().Unix()
			if now-p.lastRequestTime.Get() > 5 {
				log.Printf(Red("[%s]Graceful Exit..."), p.serviceName)
				break
			} else {
				log.Printf(Cyan("[%s]Sleeping %d seconds before Exit...\n"),
					p.serviceName, now-start)
				time.Sleep(time.Second)
			}
		}

		transport.Interrupt()
		transport.Close()
	}()

	go func() {
		var address string
		for c := range ch {
			// 为每个Connection建立一个Session
			socket, ok := c.(rpc_utils.SocketAddr)

			if ok {
				if isUnixDomain {
					address = p.frontendAddr
				} else {
					address = socket.Addr().String()
				}
			} else {
				address = "unknow"
			}
			x := NewNonBlockSession(c, address, p.verbose, &p.lastRequestTime)
			// Session独立处理自己的请求
			go x.Serve(p.backendService, 1000)
		}
	}()

	// Accept什么时候出错，出错之后如何处理呢?
	for {
		c, err := transport.Accept()
		if err != nil {
			close(ch)
			break
		} else {
			ch <- c
		}
	}
}

//
// go test proxy -v -run "TestSession"
//
func TestSession(t *testing.T) {
	// 作为一个Server
	transport, err := thrift.NewTServerSocket("127.0.0.1:0")
	err = transport.Open() // 打开Transport
	defer transport.Close()

	err = transport.Listen() // 开始监听
	assert.NoError(t, err)

	addr := transport.Addr().String()
	fmt.Println("Addr: ", addr)

	// 1. Fake Requests
	var requestNum int32 = 10
	requests := make([]*Request, 0, requestNum)
	var i int32
	for i = 0; i < requestNum; i++ {
		buf := make([]byte, 100, 100)
		l := fakeData("Hello", thrift.CALL, i+1, buf[0:0])
		buf = buf[0:l]

		req, _ := NewRequest(buf, true)

		req.Wait.Add(1) // 因为go routine可能还没有执行，代码就跑到最后面进行校验了

		assert.Equal(t, i+1, req.Request.SeqId, "Request SeqId是否靠谱")

		requests = append(requests, req)
	}

	// 2. 将请求交给BackendConn
	go func() {
		// 模拟请求:
		// 客户端代码
		bc := NewBackendConn(addr, nil, "test", true)
		bc.currentSeqId = 10

		// 上线 BackendConn
		bc.IsConnActive.Set(true)

		// 准备发送数据
		var i int32
		for i = 0; i < requestNum; i++ {
			fmt.Println("Sending Request to Backend Conn", i)
			bc.PushBack(requests[i])

			requests[i].Wait.Done()
		}

		// 需要等待数据返回?
		time.Sleep(time.Second * 2)
	}()

	server := &fakeServer{}
	go func() {
		// 服务器端代码
		tran, err := transport.Accept()
		defer tran.Close()
		if err != nil {
			log.ErrorErrorf(err, "Error: %v\n", err)
		}
		assert.NoError(t, err)

		// 建立一个长连接, 同上面的: NewBackendConn通信
		session := NewSession(tran, "", true)
		session.Serve(server, 6)

		time.Sleep(time.Second * 2)
	}()

	for i = 0; i < requestNum; i++ {
		fmt.Println("===== Before Wait")
		requests[i].Wait.Wait()
		fmt.Println("===== Before After Wait")

		log.Printf("Request: %d, .....", i)
		assert.Equal(t, len(requests[i].Request.Data), len(requests[i].Response.Data))
	}
}