//
// 获取下一个可用的Worker
//
func (pq *PriorityQueue) NextWorker() *Worker {
	now := time.Now()
	for pq.Len() > 0 {
		result := (*pq)[0]

		if result.index != INVALID_INDEX && result.Expire.After(now) {
			// 只要活着，就留在优先级队列中，等待分配任务
			//			log.Println("Find Valid Worker...")

			result.priority -= 1

			// 调整Worker的优先级
			heap.Fix(pq, result.index)

			return result
		} else {
			if result.index != INVALID_INDEX {
				log.Errorf("Invalid Item index in PriorityQueue#NextWorker")
			} else {
				log.Println("Worker Expired")
				// 只有过期的元素才删除
				heap.Remove(pq, result.index)
			}
		}
	}

	log.Println("Has Not Worker...")
	return nil

}

// 创建一个BackService
func NewBackService(serviceName string, poller *zmq.Poller, topo *zk.Topology) *BackService {

	backSockets := NewBackSockets(poller)

	service := &BackService{
		ServiceName: serviceName,
		backend:     backSockets,
		poller:      poller,
		topo:        topo,
	}

	var evtbus chan interface{} = make(chan interface{}, 2)
	servicePath := topo.ProductServicePath(serviceName)
	endpoints, err := topo.WatchChildren(servicePath, evtbus)
	if err != nil {
		log.Println("Error: ", err)
		panic("Reading Service List Failed: ")
	}

	go func() {
		for true {
			// 如何监听endpoints的变化呢?
			addrSet := make(map[string]bool)
			nowStr := time.Now().Format("@2006-01-02 15:04:05")
			for _, endpoint := range endpoints {
				// 这些endpoint变化该如何处理呢?
				log.Println(utils.Green("---->Find Endpoint: "), endpoint, "For Service: ", serviceName)
				endpointInfo, _ := topo.GetServiceEndPoint(serviceName, endpoint)

				addr, ok := endpointInfo["frontend"]
				if ok {
					addrStr := addr.(string)
					log.Println(utils.Green("---->Add endpoint to backend: "), addrStr, nowStr, "For Service: ", serviceName)
					addrSet[addrStr] = true
				}
			}

			service.backend.UpdateEndpointAddrs(addrSet)

			// 等待事件
			<-evtbus
			// 读取数据，继续监听
			endpoints, err = topo.WatchChildren(servicePath, evtbus)
		}
	}()

	ticker := time.NewTicker(time.Millisecond * 1000)
	go func() {
		for _ = range ticker.C {
			service.backend.PurgeEndpoints()
		}
	}()

	return service

}

//
// 删除Service Endpoint
//
func (s *ServiceEndpoint) DeleteServiceEndpoint(top *zk.Topology) {
	path := top.ProductServiceEndPointPath(s.Service, s.ServiceId)
	if ok, _ := top.Exist(path); ok {
		zkhelper.DeleteRecursive(top.ZkConn, path, -1)
		log.Println(Red("DeleteServiceEndpoint"), "Path: ", path)
	}
}

func NewBackServices(poller *zmq.Poller, productName string, topo *zk.Topology) *BackServices {

	// 创建BackServices
	result := &BackServices{
		Services:        make(map[string]*BackService),
		OfflineServices: make(map[string]*BackService),
		poller:          poller,
		topo:            topo,
	}

	var evtbus chan interface{} = make(chan interface{}, 2)
	servicesPath := topo.ProductServicesPath()
	path, e1 := topo.CreateDir(servicesPath) // 保证Service目录存在，否则会报错
	fmt.Println("Path: ", path, "error: ", e1)
	services, err := topo.WatchChildren(servicesPath, evtbus)
	if err != nil {
		log.Println("Error: ", err)
		// TODO: 这个地方需要优化
		panic("Reading Service List Failed")
	}

	go func() {
		for true {

			result.Lock()
			for _, service := range services {
				log.Println("Service: ", service)
				if _, ok := result.Services[service]; !ok {
					result.addBackService(service)
				}
			}
			result.Unlock()

			// 等待事件
			<-evtbus
			// 读取数据，继续监听(连接过期了就过期了，再次Watch即可)
			services, err = topo.WatchChildren(servicesPath, evtbus)
		}
	}()

	// 读取zk, 等待
	log.Println("ProductName: ", result.topo.ProductName)

	return result
}

// 创建指定的Path
func (top *Topology) CreateDir(path string) (string, error) {
	dir := top.FullPath(path)
	if ok, _ := top.Exist(dir); ok {
		log.Println("Path Exists")
		return dir, nil
	} else {
		return zkhelper.CreateRecursive(top.ZkConn, dir, "", 0, zkhelper.DefaultDirACLs())
	}
}

//
// 将消息发送到Backend上去
//
func (s *BackService) HandleRequest(client_id string, msgs []string) (total int, err error, msg *[]byte) {

	backSocket := s.backend.NextSocket()
	if backSocket == nil {
		// 没有后端服务

		if config.VERBOSE {
			log.Println(utils.Red("No BackSocket Found for service:"), s.ServiceName)
		}
		errMsg := GetWorkerNotFoundData(s.ServiceName, 0)
		return 0, nil, &errMsg
	} else {
		if config.VERBOSE {
			log.Println("SendMessage With: ", backSocket.Addr, "For Service: ", s.ServiceName)
		}
		total, err = backSocket.SendMessage("", client_id, "", msgs)
		return total, err, nil
	}
}

// 在第一次使用时再连接
func (p *BackSocket) connect() error {
	var err error
	p.Socket, err = zmq.NewSocket(zmq.DEALER)
	if err == nil {
		// 这个Id存在问题:
		socketSeq += 1
		p.Socket.SetIdentity(fmt.Sprintf("proxy-%d-%d", os.Getpid(), socketSeq))

		p.Socket.Connect(p.Addr)
		// 都只看数据的输入
		// 数据的输出经过异步处理，不用考虑时间的问题
		p.poller.Add(p.Socket, zmq.POLLIN)
		log.Println("Socket Create Succeed")
		return nil
	} else {
		log.Println("Socket Create Failed: ", err)
		return err
	}
}

func (p *BackSocket) SendMessage(parts ...interface{}) (total int, err error) {
	if p.Socket == nil {
		err := p.connect()
		if err != nil {
			log.Println("Socket Connect Failed")
			return 0, err
		}
	}

	return p.Socket.SendMessage(parts...)
}

//
// 设置RPC Proxy的数据:
//     绑定的前端的ip/port, 例如: {"rpc_front": "tcp://127.0.0.1:5550"}
//
func (top *Topology) SetRpcProxyData(proxyInfo map[string]interface{}) error {
	path := top.FullPath("/rpc_proxy")
	data, err := json.Marshal(proxyInfo)
	if err != nil {
		return err
	}

	// topo.FlagEphemeral 这里的ProxyInfo是手动配置的，需要持久化
	path, err = CreateOrUpdate(top.ZkConn, path, string(data), 0, zkhelper.DefaultDirACLs(), true)
	log.Println(green("SetRpcProxyData"), "Path: ", path, ", Error: ", err, ", Data: ", string(data))
	return err
}

func (pq *PPQueue) PurgeExpired() {
	now := time.Now()
	expiredWokers := make([]*Worker, 0)
	// 给workerQueue中的所有的worker发送心跳消息
	for _, worker := range pq.WorkerQueue {
		if worker.Expire.Before(now) {
			fmt.Println("Purge Worker: ", worker.Identity, ", At Index: ", worker.index)
			expiredWokers = append(expiredWokers, worker)
		}
	}

	log.Println("expiredWokers: ", len(expiredWokers))

	// 删除过期的Worker
	for _, worker := range expiredWokers {
		log.Println("Purge Worker: ", worker.Identity, ", At Index: ", worker.index)
		heap.Remove(&(pq.WorkerQueue), worker.index)
		delete(pq.id2item, worker.Identity)
	}

	log.Println("Available Workers: ", green(fmt.Sprintf("%d", len(pq.WorkerQueue))))
}

//
// 将消息发送到Backend上去
//
func (s *BackService) HandleRequest(req *Request) (err error) {
	backendConn := s.NextBackendConn()

	s.lastRequestTime.Set(time.Now().Unix())

	if backendConn == nil {
		// 没有后端服务
		if s.verbose {
			log.Println(Red("No BackSocket Found for service:"), s.serviceName)
		}
		// 从errMsg来构建异常
		errMsg := GetWorkerNotFoundData(req, "BackService")
		req.Response.Data = errMsg

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

func (conf *Config) getFrontendAddr() string {
	var frontendAddr = ""
	// 如果没有指定FrontHost, 则自动根据 IpPrefix来进行筛选，
	// 例如: IpPrefix: 10., 那么最终内网IP： 10.4.10.2之类的被选中
	if conf.FrontHost == "" {
		log.Println("FrontHost: ", conf.FrontHost, ", Prefix: ", conf.IpPrefix)
		if conf.IpPrefix != "" {
			conf.FrontHost = GetIpWithPrefix(conf.IpPrefix)
		}
	}
	if conf.FrontPort != "" && conf.FrontHost != "" {
		frontendAddr = fmt.Sprintf("tcp://%s:%s", conf.FrontHost, conf.FrontPort)
	}
	return frontendAddr
}

//
// 读取RPC Proxy的数据:
//     绑定的前端的ip/port, 例如: {"rpc_front": "tcp://127.0.0.1:5550"}
//
func (top *Topology) GetRpcProxyData() (proxyInfo map[string]interface{}, e error) {
	path := top.FullPath("/rpc_proxy")
	data, _, err := top.ZkConn.Get(path)

	log.Println("Data: ", data, ", err: ", err)
	if err != nil {
		return nil, err
	}
	proxyInfo = make(map[string]interface{})
	err = json.Unmarshal(data, &proxyInfo)
	if err != nil {
		return nil, err
	} else {
		return proxyInfo, nil
	}
}

//
// 将不在: addrSet中的endPoint标记为下线
//
func (p *BackSockets) UpdateEndpointAddrs(addrSet map[string]bool) {
	p.Lock()
	defer p.Unlock()

	var addr string
	for addr, _ = range addrSet {
		p.addEndpoint(addr)
	}

	now := time.Now().Format("@2006-01-02 15:04:05")
	for i := 0; i < p.Active; i++ {
		if _, ok := addrSet[p.Sockets[i].Addr]; !ok {
			log.Println(utils.Red("MarkEndpointsOffline#Mark Backend Offline: "), p.Sockets[i].Addr, now)

			p.markOffline(p.Sockets[i])
			i--
		}
	}
}

//
// 通过参数依赖，保证getFrontendAddr的调用位置（必须等待Host, IpPrefix, Port读取完毕之后)
//
func (conf *Config) getFrontendAddr(frontHost, ipPrefix, frontPort string) string {
	if conf.FrontSock != "" {
		return conf.FrontSock
	}

	var frontendAddr = ""
	// 如果没有指定FrontHost, 则自动根据 IpPrefix来进行筛选，
	// 例如: IpPrefix: 10., 那么最终内网IP： 10.4.10.2之类的被选中
	if frontHost == "" {
		log.Println("FrontHost: ", frontHost, ", Prefix: ", ipPrefix)
		if ipPrefix != "" {
			frontHost = GetIpWithPrefix(ipPrefix)
		}
	}
	if frontPort != "" && frontHost != "" {
		frontendAddr = fmt.Sprintf("%s:%s", frontHost, frontPort)
	}
	return frontendAddr
}

//
// 注册一个服务的Endpoints
//
func (s *ServiceEndpoint) AddServiceEndpoint(topo *zk.Topology) error {
	path := topo.ProductServiceEndPointPath(s.Service, s.ServiceId)
	data, err := json.Marshal(s)
	if err != nil {
		return err
	}

	// 创建Service(XXX: Service本身不包含数据)
	zk.CreateRecursive(topo.ZkConn, os_path.Dir(path), "", 0, zkhelper.DefaultDirACLs())

	// 当前的Session挂了，服务就下线
	// topo.FlagEphemeral

	// 参考： https://www.box.com/blog/a-gotcha-when-using-zookeeper-ephemeral-nodes/
	// 如果之前的Session信息还存在，则先删除；然后再添加
	topo.ZkConn.Delete(path, -1)
	var pathCreated string
	pathCreated, err = topo.ZkConn.Create(path, []byte(data), int32(zookeeper.FlagEphemeral), zkhelper.DefaultFileACLs())

	log.Println(Green("AddServiceEndpoint"), "Path: ", pathCreated, ", Error: ", err)
	return err
}

//
// 注册一个服务的Endpoints
//
func (top *Topology) AddServiceEndPoint(service string, endpoint string, endpointInfo map[string]interface{}) error {
	path := top.ProductServiceEndPointPath(service, endpoint)
	data, err := json.Marshal(endpointInfo)
	if err != nil {
		return err
	}

	// 创建Service(XXX: Service本身不包含数据)
	CreateRecursive(top.zkConn, os_path.Dir(path), "", 0, zkhelper.DefaultDirACLs())

	// 当前的Session挂了，服务就下线
	// topo.FlagEphemeral

	// 参考： https://www.box.com/blog/a-gotcha-when-using-zookeeper-ephemeral-nodes/
	// 如果之前的Session信息还存在，则先删除；然后再添加
	top.zkConn.Delete(path, -1)
	var pathCreated string
	pathCreated, err = top.zkConn.Create(path, []byte(data), int32(topo.FlagEphemeral), zkhelper.DefaultFileACLs())

	log.Println(green("SetRpcProxyData"), "Path: ", pathCreated, ", Error: ", err)
	return err
}

//
// 删除过期的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-- // 保持原位
		}

	}
}

//
// 两参数是必须的:  ProductName, zkAddress, frontAddr可以用来测试
//
func mainBody(productName string, frontAddr string, zkAdresses string) {
	// 1. 创建到zk的连接
	var topo *zk.Topology
	topo = zk.NewTopology(productName, zkAdresses)

	// 3. 读取后端服务的配置
	poller := zmq.NewPoller()
	backServices := proxy.NewBackServices(poller, productName, topo)

	// 4. 创建前端服务
	frontend, _ := zmq.NewSocket(zmq.ROUTER)
	defer frontend.Close()

	// ROUTER/ROUTER绑定到指定的端口
	log.Println("---->Bind: ", magenta(frontAddr))
	frontend.Bind(frontAddr) //  For clients

	// 开始监听前端服务
	poller.Add(frontend, zmq.POLLIN)

	for {
		var sockets []zmq.Polled
		var err error

		sockets, err = poller.Poll(HEARTBEAT_INTERVAL)

		if err != nil {
			log.Println("Encounter Errors, Services Stoped: ", err)
			continue
		}

		for _, socket := range sockets {
			switch socket.Socket {

			case frontend:
				if config.VERBOSE {
					log.Println("----->Message from front: ")
				}
				msgs, err := frontend.RecvMessage(0)

				if err != nil {
					continue //  Interrupted
				}
				var service string
				var client_id string

				utils.PrintZeromqMsgs(msgs, "ProxyFrontEnd")

				// msg格式: <client_id, '', service,  '', other_msgs>
				client_id, msgs = utils.Unwrap(msgs)
				service, msgs = utils.Unwrap(msgs)

				//				log.Println("Client_id: ", client_id, ", Service: ", service)

				backService := backServices.GetBackService(service)

				if backService == nil {
					log.Println("BackService Not Found...")
					// 最后一个msg为Thrift编码后的消息
					thriftMsg := msgs[len(msgs)-1]
					// XXX: seqId如果不需要，也可以使用固定的数字
					_, _, seqId, _ := proxy.ParseThriftMsgBegin([]byte(thriftMsg))
					errMsg := proxy.GetServiceNotFoundData(service, seqId)

					// <client_id, "", errMsg>
					if len(msgs) > 1 {
						frontend.SendMessage(client_id, "", msgs[0:len(msgs)-1], errMsg)
					} else {
						frontend.SendMessage(client_id, "", errMsg)
					}

				} else {
					// <"", client_id, "", msgs>
					if config.PROFILE {
						lastMsg := msgs[len(msgs)-1]
						msgs = msgs[0 : len(msgs)-1]
						msgs = append(msgs, fmt.Sprintf("%.4f", float64(time.Now().UnixNano())*1e-9), "", lastMsg)
						if config.VERBOSE {
							log.Println(printList(msgs))
						}
					}
					total, err, errMsg := backService.HandleRequest(client_id, msgs)
					if errMsg != nil {
						if config.VERBOSE {
							log.Println("backService Error for service: ", service)
						}
						if len(msgs) > 1 {
							frontend.SendMessage(client_id, "", msgs[0:len(msgs)-1], *errMsg)
						} else {
							frontend.SendMessage(client_id, "", *errMsg)
						}
					} else if err != nil {
						log.Println(utils.Red("backService.HandleRequest Error: "), err, ", Total: ", total)
					}
				}
			default:
				// 除了来自前端的数据，其他的都来自后端
//.........这里部分代码省略.........

//
// Load Balance如何运维呢?
// 1. 在服务提供方，会会启动Load Balance, 它只负责本机器的某个指定服务的lb
// 2. 正常情况下，不能被轻易杀死
// 3. 需要考虑 graceful stop, 在死之前告知所有的proxy，如何告知呢? TODO
//
//
func main() {
	args, err := docopt.Parse(usage, nil, true, "Chunyu RPC Load Balance v0.1", true)
	if err != nil {
		log.Println(err)
		os.Exit(1)
	}
	var maxFileFrag = 2
	var maxFragSize int64 = bytesize.GB * 1
	if s, ok := args["--log-filesize"].(string); ok && s != "" {
		v, err := bytesize.Parse(s)
		if err != nil {
			log.PanicErrorf(err, "invalid max log file size = %s", s)
		}
		maxFragSize = v
	}

	// set output log file
	if s, ok := args["-L"].(string); ok && s != "" {
		f, err := log.NewRollingFile(s, maxFileFrag, maxFragSize)
		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)
	}
	var backendAddr, frontendAddr, zkAddr, productName, serviceName string

	// set config file
	if args["-c"] != nil {
		configFile := args["-c"].(string)
		conf, err := utils.LoadConf(configFile)
		if err != nil {
			log.PanicErrorf(err, "load config failed")
		}
		productName = conf.ProductName

		if conf.FrontHost == "" {
			fmt.Println("FrontHost: ", conf.FrontHost, ", Prefix: ", conf.IpPrefix)
			if conf.IpPrefix != "" {
				conf.FrontHost = utils.GetIpWithPrefix(conf.IpPrefix)
			}
		}
		if conf.FrontPort != "" && conf.FrontHost != "" {
			frontendAddr = fmt.Sprintf("tcp://%s:%s", conf.FrontHost, conf.FrontPort)
		}

		backendAddr = conf.BackAddr
		serviceName = conf.Service

		zkAddr = conf.ZkAddr
		config.VERBOSE = conf.Verbose

	} else {
		productName = ""
		zkAddr = ""
	}

	if s, ok := args["--product"].(string); ok && s != "" {
		productName = s
	} else if productName == "" {
		// 既没有config指定，也没有命令行指定，则报错
		log.PanicErrorf(err, "Invalid ProductName: %s", s)
	}

	if s, ok := args["--zk"].(string); ok && s != "" {
		zkAddr = s
	} else if zkAddr == "" {
		log.PanicErrorf(err, "Invalid zookeeper address: %s", s)
	}

	if s, ok := args["--service"].(string); ok && s != "" {
		serviceName = s
	} else if serviceName == "" {
		log.PanicErrorf(err, "Invalid ServiceName: %s", s)
	}

	if s, ok := args["--baddr"].(string); ok && s != "" {
		backendAddr = s
	} else if backendAddr == "" {
		log.PanicErrorf(err, "Invalid backend address: %s", s)
	}
	if s, ok := args["--faddr"].(string); ok && s != "" {
		frontendAddr = s
	} else if frontendAddr == "" {
		//
//.........这里部分代码省略.........