// 获取下一个active状态的BackendConn
func (s *BackServiceLB) nextBackendConn() *BackendConnLB {
	s.activeConnsLock.Lock()
	defer s.activeConnsLock.Unlock()

	// TODO: 暂时采用RoundRobin的方法，可以采用其他具有优先级排列的方法
	var backSocket *BackendConnLB

	if len(s.activeConns) == 0 {
		if s.verbose {
			log.Printf(Cyan("[%s]ActiveConns Len 0"), s.serviceName)
		}
		backSocket = nil
	} else {
		if s.currentConnIndex >= len(s.activeConns) {
			s.currentConnIndex = 0
		}
		backSocket = s.activeConns[s.currentConnIndex]
		s.currentConnIndex++
		if s.verbose {
			log.Printf(Cyan("[%s]ActiveConns Len %d, CurrentIndex: %d"), s.serviceName,
				len(s.activeConns), s.currentConnIndex)
		}
	}
	return backSocket
}

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

	bc.Lock()
	seqRequest := bc.seqNum2Request
	bc.seqNum2Request = make(map[int32]*Request, 4096)
	bc.Unlock()
	threshold := time.Now().Add(-time.Second * 5)
	for _, request := range seqRequest {
		if request.Start > 0 {
			t := time.Unix(request.Start, 0)
			if t.After(threshold) {
				// 似乎在笔记本上，合上显示器之后出出现网络错误
				log.Printf(Red("[%s]Handle Failed Request: %s, Started: %s"),
					request.Service, request.Request.Name, FormatYYYYmmDDHHMMSS(t))
			}
		} else {
			log.Printf(Red("[%s]Handle Failed Request: %s"), request.Service,
				request.Request.Name)
		}
		request.Response.Err = err
		if request.Wait != nil {
			request.Wait.Done()
		}
	}

}

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

		// 解码错误，直接报错
		if err != nil {
			return err
		}

		// 找到对应的Request
		bc.Lock()
		req, ok := bc.seqNum2Request[seqId]
		if ok {
			delete(bc.seqNum2Request, seqId)
		}
		bc.Unlock()

		// 如果是心跳，则OK
		if typeId == MESSAGE_TYPE_HEART_BEAT {
			//			log.Printf(Magenta("Get Ping/Pang Back"))
			bc.hbLastTime.Set(time.Now().Unix())
			return nil
		}

		if !ok {
			return errors.New("Invalid Response")
		}
		if bc.verbose {
			log.Printf("[%s]Data From Server, seqId: %d, Request: %d", req.Service, seqId, req.Request.SeqId)
		}
		r = req
		r.Response.TypeId = typeId
	}

	r.Response.Data, r.Response.Err = data, err

	// 还原SeqId
	if data != nil {
		r.RestoreSeqId()
	}

	// 设置几个控制用的channel
	if err != nil && r.Failed != nil {
		r.Failed.Set(true)
	}
	if r.Wait != nil {
		r.Wait.Done()
	}

	return err
}

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
}

//
// 不断建立到后端的逻辑，负责: 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
		}

		c := NewTBufferedFramedTransport(transport, 100*time.Microsecond, 20)

		// 2. 将 bc.input 中的请求写入 后端的Rpc Server
		err = bc.loopWriter(c) // 同步

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

		// 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)
			}
		}
	}
}

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

	bc.Lock()
	seqRequest := bc.seqNum2Request
	bc.seqNum2Request = make(map[int32]*Request)
	bc.Unlock()

	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
			if request.Wait != nil {
				request.Wait.Done()
			}
		}
	}

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

}

//
// 后端如何处理一个Request, 处理完毕之后直接返回，因为Caller已经做好异步处理了
//
func (s *BackServiceLB) Dispatch(r *Request) error {
	backendConn := s.nextBackendConn()

	r.Service = s.serviceName

	if backendConn == nil {
		// 没有后端服务
		if s.verbose {
			log.Printf(Red("[%s]No BackSocket Found: %s"),
				s.serviceName, r.Request.Name)
		}
		// 从errMsg来构建异常
		errMsg := GetWorkerNotFoundData(r, "BackServiceLB")
		//		log.Printf(Magenta("---->Convert Error Back to Exception:[%d] %s\n"), len(errMsg), string(errMsg))
		r.Response.Data = errMsg

		return nil
	} else {
		//		if s.verbose {
		//			log.Println("SendMessage With: ", backendConn.Addr4Log(), "For Service: ", s.serviceName)
		//		}
		backendConn.PushBack(r)

		r.Wait.Wait()

		return nil
	}
}

func (s *Session) loopWriter(tasks <-chan *Request) error {
	// Proxy: Session ---> Client
	for r := range tasks {
		// 1. 等待Request对应的Response
		//    出错了如何处理呢?
		s.handleResponse(r)

		// 2. 将结果写回给Client
		if s.verbose {
			log.Printf("[%s]Session#loopWriter --> client FrameSize: %d",
				r.Service, len(r.Response.Data))
		}

		// r.Response.Data ---> Client
		_, err := s.TBufferedFramedTransport.Write(r.Response.Data)
		if err != nil {
			log.ErrorErrorf(err, "Write back Data Error: %v", err)
			return err
		}

		// 3. Flush
		err = s.TBufferedFramedTransport.FlushBuffer(true) // len(tasks) == 0
		if err != nil {
			log.ErrorErrorf(err, "Write back Data Error: %v", err)
			return err
		}
		r.Recycle()
	}
	return nil
}

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

		// 解码错误，直接报错
		if err != nil {
			return err
		}

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

		// 找到对应的Request
		bc.Lock()
		req, ok := bc.seqNum2Request[seqId]
		if ok {
			delete(bc.seqNum2Request, seqId)
		}
		bc.Unlock()

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

		if !ok {
			return errors.New("Invalid Response")
		}

		//		log.Printf("Data From Server, seqId: %d, Request: %d\n", seqId, req.Request.SeqId)
		r = req
		r.Response.TypeId = typeId
	}

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

	// 设置几个控制用的channel
	if err != nil && r.Failed != nil {
		r.Failed.Set(true)
	}
	if r.Wait != nil {
		r.Wait.Done()
	}

	return err
}

//
// 确保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 = 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
}

//
// 两参数是必须的:  ProductName, zkAddress, frontAddr可以用来测试
//
func (p *ProxyServer) Run() {

	var transport thrift.TServerTransport
	var err error

	log.Printf(Magenta("Start Proxy at Address: %s"), p.proxyAddr)
	// 读取后端服务的配置
	isUnixDomain := false
	if !strings.Contains(p.proxyAddr, ":") {
		if FileExist(p.proxyAddr) {
			os.Remove(p.proxyAddr)
		}
		transport, err = NewTServerUnixDomain(p.proxyAddr)
		isUnixDomain = true
	} else {
		transport, err = thrift.NewTServerSocket(p.proxyAddr)
	}
	if err != nil {
		log.ErrorErrorf(err, "Server Socket Create Failed: %v, Front: %s", err, p.proxyAddr)
	}

	// 开始监听
	//	transport.Open()
	transport.Listen()

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

	go func() {
		var address string
		for c := range ch {
			// 为每个Connection建立一个Session
			socket, ok := c.(SocketAddr)
			if isUnixDomain {
				address = p.proxyAddr
			} else if ok {
				address = socket.Addr().String()
			} else {
				address = "unknow"
			}
			x := NewSession(c, address, p.verbose)
			// Session独立处理自己的请求
			go x.Serve(p.router, 1000)
		}
	}()

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

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() {
		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 {
		log.Printf(Red("[%s]Remove BackendConn From activeConns: %s, Index: %d"),
			s.serviceName, conn.Addr(), 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))
		}
	}
}

//
// 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() {
		log.Printf(Red("[%s]MarkConnActiveFalse: %s, %p"), bc.service, bc.addr, bc.delegate)
		// 从Active切换到非正常状态
		bc.IsConnActive.Set(false)

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

// 只有在conn出现错误时才会调用
func (s *BackServiceLB) StateChanged(conn *BackendConnLB) {
	s.activeConnsLock.Lock()
	defer s.activeConnsLock.Unlock()

	log.Printf(Green("[%s]StateChanged: %s, Index: %d, Count: %d"), conn.serviceName, conn.addr4Log, conn.Index, len(s.activeConns))
	if conn.IsConnActive.Get() {
		// BackServiceLB 只有一个状态转移: Active --> Not Active
		log.Printf(Magenta("Unexpected BackendConnLB State"))
		if s.verbose {
			panic("Unexpected BackendConnLB State")
		}
	} else {
		log.Printf(Red("Remove BackendConn From activeConns: %s, Index: %d, Count: %d"),
			conn.Addr4Log(), conn.Index, len(s.activeConns))

		// 从数组中删除一个元素(O(1)的操作)
		if conn.Index != INVALID_ARRAY_INDEX {
			// 1. 和最后一个元素进行交换
			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

			}
			log.Printf(Red("Remove BackendConn From activeConns: %s"), conn.Addr4Log())

			// 2. slice
			s.activeConns = s.activeConns[0:lastIndex]

		}
	}
}

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

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

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

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

	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)
		}
	}

}

//
//
// 等待Request请求的返回: Session最终被Block住
//
func (s *Session) handleResponse(r *Request) {
	// 等待结果的出现
	r.Wait.Wait()

	// 将Err转换成为Exception
	if r.Response.Err != nil {

		r.Response.Data = GetThriftException(r, "proxy_session")
		log.Printf(Magenta("---->Convert Error Back to Exception"))
	}

	// 如何处理Data和Err呢?
	incrOpStats(r.OpStr, microseconds()-r.Start)
}

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

	// 增加统计
	s.LastOpUnix = time.Now().Unix()
	s.Ops++

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

//
// 删除过期的Endpoints
//
func (p *BackSockets) PurgeEndpoints() {
	// 没有需要删除的对象
	if p.Active == len(p.Sockets) {
		return
	}

	log.Printf(utils.Green("PurgeEndpoints %d vs. %d"), p.Active, len(p.Sockets))

	p.Lock()
	defer p.Unlock()

	now := time.Now().Unix()
	nowStr := time.Now().Format("@2006-01-02 15:04:05")

	for i := p.Active; i < len(p.Sockets); i++ {
		// 逐步删除过期的Sockets
		current := p.Sockets[i]
		lastIndex := len(p.Sockets) - 1
		if now-current.markedOfflineTime > 5 {

			// 将i和最后一个元素交换
			p.swap(current, p.Sockets[lastIndex])

			// 关闭
			// current
			// 关闭旧的Socket
			log.Println(utils.Red("PurgeEndpoints#Purge Old Socket: "), current.Addr, nowStr)
			// 由Socket自己维护自己的状态
			// current.Socket.Close()

			p.Sockets[lastIndex] = nil
			p.Sockets = p.Sockets[0:lastIndex]

			i-- // 保持原位
		}

	}
}

func NewThriftLoadBalanceServer(config *utils.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 = zk.NewTopology(p.productName, p.zkAddr)
	p.lbServiceName = GetServiceIdentity(p.frontendAddr)

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

}