//握手包
func handshake(ga *c.GroupAuth, remoteClient *c.RemotingClient) (bool, error) {

	for i := 0; i < 3; i++ {
		p := protocol.MarshalConnMeta(ga.GroupId, ga.SecretKey)
		rpacket := packet.NewPacket(protocol.CMD_CONN_META, p)
		resp, err := remoteClient.WriteAndGet(*rpacket, 5*time.Second)
		if nil != err {
			//两秒后重试
			time.Sleep(2 * time.Second)
			log.Warn("kiteClient|handShake|FAIL|%s|%s\n", ga.GroupId, err)
		} else {
			authAck, ok := resp.(*protocol.ConnAuthAck)
			if !ok {
				return false, errors.New("Unmatches Handshake Ack Type! ")
			} else {
				if authAck.GetStatus() {
					log.Info("kiteClient|handShake|SUCC|%s|%s\n", ga.GroupId, authAck.GetFeedback())
					return true, nil
				} else {
					log.Warn("kiteClient|handShake|FAIL|%s|%s\n", ga.GroupId, authAck.GetFeedback())
					return false, errors.New("Auth FAIL![" + authAck.GetFeedback() + "]")
				}
			}
		}
	}

	return false, errors.New("handshake fail! [" + remoteClient.RemoteAddr() + "]")
}

//remove segment
func (self *MessageStore) remove(s *Segment) {

	self.Lock()
	//remove from segments
	for i, s := range self.segments {
		if s.sid == s.sid {
			self.segments = append(self.segments[0:i], self.segments[i+1:]...)
			break
		}
	}
	//remove from cache
	for e := self.segmentCache.Front(); nil != e; e = e.Next() {
		cs := e.Value.(*Segment)
		if cs.sid == s.sid {
			self.segmentCache.Remove(e)
			break
		}
	}

	//close segment
	s.Close()

	err := os.Remove(s.path)
	if nil != err {
		log.Warn("MessageStore|Remove|Segment|FAIL|%s|%s", err, s.path)
	}
	err = os.Remove(s.slog.path)
	if nil != err {
		log.Warn("MessageStore|Remove|SegmentLog|FAIL|%s|%s", err, s.slog.path)
	}

	self.Unlock()
	log.Info("MessageStore|Remove|Segment|%s", s.path)
}

//订阅关系改变
func (self *BindExchanger) onBindChanged(topic, groupId string, newbinds []*Binding) {

	if len(groupId) <= 0 {
		delete(self.exchanger, topic)
		return
	}

	//不是当前服务可以处理的topic则直接丢地啊哦
	if sort.SearchStrings(self.topics, topic) == len(self.topics) {
		log.Warn("BindExchanger|onBindChanged|UnAccept Bindings|%s|%s|%s\n", topic, self.topics, newbinds)
		return
	}

	v, ok := self.exchanger[topic]
	if !ok {
		v = make(map[string][]*Binding, 10)
		self.exchanger[topic] = v
	}

	if len(newbinds) > 0 {
		v[groupId] = newbinds
	} else {
		delete(v, groupId)
	}
}

func (self *RemotingHandler) Process(ctx *DefaultPipelineContext, event IEvent) error {

	revent, ok := self.cast(event)
	if !ok {
		return ERROR_INVALID_EVENT_TYPE
	}

	// log.Printf("RemotingHandler|Process|%s|%t\n", self.GetName(), revent)
	var futures map[string]*turbo.Future
	if len(revent.GroupIds) <= 0 && len(revent.TargetHost) <= 0 {
		log.Warn("RemotingHandler|%s|Process|NO GROUP OR HOSTS|%s|%s\n", self.GetName(), revent)
		futures = EMPTY_FUTURE
	} else {
		//发送数据
		futures = self.invokeGroup(revent)
	}
	//写入future的响应
	revent.futures <- futures
	close(revent.futures)

	//创建创建网络写出结果
	fe := NewRemoteFutureEvent(revent, futures)
	ctx.SendForward(fe)
	return nil
}

func (self *ValidateHandler) Process(ctx *DefaultPipelineContext, event IEvent) error {

	aevent, ok := self.cast(event)
	if !ok {
		return ERROR_INVALID_EVENT_TYPE
	}

	remoteClient := aevent.getClient()
	//做权限校验.............
	isAuth := self.clientManager.Validate(remoteClient)
	// log.DebugLog("kite_handler",  "ValidateHandler|CONNETION|%s|%s\n", remoteClient.RemoteAddr(), isAuth)
	if isAuth {
		ctx.SendForward(event)
	} else {
		log.Warn("ValidateHandler|UnAuth CONNETION|%s\n", remoteClient.RemoteAddr())
		cmd := protocol.MarshalConnAuthAck(false, "未授权的访问,连接关闭!")
		//响应包
		p := packet.NewPacket(protocol.CMD_CONN_AUTH, cmd)

		//直接写出去授权失败
		remoteClient.Write(*p)
		//断开连接
		remoteClient.Shutdown()
	}

	return nil

}

func (self *AccessHandler) Process(ctx *DefaultPipelineContext, event IEvent) error {

	// log.Debug("accessEvent|Process|%s|%t\n", self.GetName(), event)

	aevent, ok := self.cast(event)
	if !ok {
		return ERROR_INVALID_EVENT_TYPE
	}

	//做权限校验.............
	if false {
		log.Warn("accessEvent|Process|INVALID AUTH|%s|%s\n", aevent.groupId, aevent.secretKey)
	}

	// 权限验证通过 保存到clientmanager
	self.clientManager.Auth(client.NewGroupAuth(aevent.groupId, aevent.secretKey), aevent.remoteClient)

	// log.Info("accessEvent|Process|NEW CONNECTION|AUTH SUCC|%s|%s|%s\n", aevent.groupId, aevent.secretKey, aevent.remoteClient.RemoteAddr())

	cmd := protocol.MarshalConnAuthAck(true, "授权成功")
	//响应包
	packet := packet.NewRespPacket(aevent.opaque, protocol.CMD_CONN_AUTH, cmd)

	//向当前连接写入当前包
	remoteEvent := NewRemotingEvent(packet, []string{aevent.remoteClient.RemoteAddr()})

	//向后走网络传输
	ctx.SendForward(remoteEvent)
	return nil

}

func (self *ConnPool) enhancedPool(size int) error {

	//初始化一下最小的Poolsize,让入到idlepool中
	for i := 0; i < size; i++ {
		j := 0
		var err error
		var conn IConn
		for ; j < 3; j++ {
			err, conn = self.dialFunc(self.id())
			if nil != err || nil == conn {
				log.Warn("POOL_FACTORY|CREATE CONNECTION|INIT|FAIL|%s", err)
				continue

			} else {
				break
			}
		}

		if j >= 3 || nil == conn {
			return errors.New("POOL_FACTORY|CREATE CONNECTION|INIT|FAIL|%s" + err.Error())
		}

		idleconn := &IdleConn{conn: conn, expiredTime: (time.Now().Add(self.idletime))}
		self.idlePool.PushFront(idleconn)
	}

	return nil
}

func (self *HeartbeatHandler) keepAlive() {

	for {
		select {
		case <-time.After(self.heartbeatPeriod):
			//心跳检测
			func() {
				id := time.Now().Unix()
				clients := self.clientMangager.ClientsClone()
				p := protocol.MarshalHeartbeatPacket(id)
				for h, c := range clients {
					i := 0
					//关闭的时候发起重连
					if c.IsClosed() {
						i = 3
					} else {
						//如果是空闲的则发起心跳
						if c.Idle() {
							for ; i < 3; i++ {
								hp := packet.NewPacket(protocol.CMD_HEARTBEAT, p)
								err := c.Ping(hp, time.Duration(self.heartbeatTimeout))
								//如果有错误则需要记录
								if nil != err {
									log.Warn("HeartbeatHandler|KeepAlive|FAIL|%s|local:%s|remote:%s|%d\n", err, c.LocalAddr(), h, id)
									continue
								} else {
									log.Info("HeartbeatHandler|KeepAlive|SUCC|local:%s|remote:%s|%d|%d ...\n", c.LocalAddr(), h, id, i)
									break
								}
							}
						}
					}
					if i >= 3 {
						//说明连接有问题需要重连
						c.Shutdown()
						self.clientMangager.SubmitReconnect(c)
						log.Warn("HeartbeatHandler|SubmitReconnect|%s\n", c.RemoteAddr())
					}
				}
			}()
		}
	}

}

//同步发起ping的命令
func (self *RemotingClient) Ping(heartbeat *packet.Packet, timeout time.Duration) error {
	pong, err := self.WriteAndGet(*heartbeat, timeout)
	if nil != err {
		return err
	}
	version, ok := pong.(int64)
	if !ok {
		log.Warn("RemotingClient|Ping|Pong|ERROR TYPE |%s\n", pong)
		return ERROR_PONG
	}
	self.updateHeartBeat(version)
	return nil
}

//traverse oplog
func (self *SegmentLog) Replay(do func(l *oplog)) {

	self.Open()
	offset := int64(0)
	tmp := make([]byte, 1024)
	//seek to head
	self.rf.Seek(0, 0)
	self.br.Reset(self.rf)

	for {
		var length int32
		err := binary.Read(self.br, binary.BigEndian, &length)
		if nil != err {
			if err == io.EOF {
				self.br.Reset(self.rf)
				break
			}
			log.Warn("SegmentLog|Replay|LEN|%s|Skip...", err)
			continue
		}

		// log.Debug("SegmentLog|Replay|LEN|%d", length)

		if int(length) > cap(tmp) {
			grow := make([]byte, int(length)-cap(tmp))
			tmp = append(tmp, grow...)
		}

		err = binary.Read(self.br, binary.BigEndian, tmp[:int(length)-4])
		if nil != err {
			self.br.Reset(self.rf)
			log.Error("SegmentLog|Replay|Data|%s", err)
			break
		}

		var ol oplog
		r := bytes.NewReader(tmp[:int(length)-4])
		deco := gob.NewDecoder(r)
		err = deco.Decode(&ol)
		if nil != err {
			log.Error("SegmentLog|Replay|unmarshal|oplog|FAIL|%s", err)
			continue
		}
		// log.Debug("SegmentLog|Replay|oplog|%s", ol)
		do(&ol)
		//line
		offset += int64(length)

	}
	self.offset = int64(offset)
}

func (self *BaseForwardHandler) HandleForward(ctx *DefaultPipelineContext, event IForwardEvent) error {

	//处理逻辑成功则向后传递
	if !self.processor.TypeAssert(event) {
		ctx.SendForward(event)
		return nil
	} else {
		now := time.Now().Unix()
		err := self.processor.Process(ctx, event)
		cost := time.Now().Unix() - now
		if cost > 100 {
			log.Warn("BaseForwardHandler|%s|cost:%d\n", self.GetName(), cost)
		}
		return err
	}
}

//监听数据变更
func (self *ZKManager) listenEvent() {
	for !self.isClose {
		//根据zk的文档 watcher机制是无法保证可靠的，其次需要在每次处理完watcher后要重新注册watcher
		change := <-self.eventChan
		path := change.Path
		switch change.Type {
		case zk.EventSession:
			if change.State == zk.StateExpired {
				log.Warn("ZKManager|OnSessionExpired!")
				//zk链接开则需要重新链接重新推送
				self.watcher.OnSessionExpired()

			}
		case zk.EventNodeDeleted:
			self.session.ExistsW(path)
			self.watcher.NodeChange(path, ZkEvent(change.Type), []string{})
			// log.Info("ZKManager|listenEvent|%s|%s\n", path, change)
		case zk.EventNodeCreated, zk.EventNodeChildrenChanged:
			childnodes, _, _, err := self.session.ChildrenW(path)
			if nil != err {
				log.Error("ZKManager|listenEvent|CD|%s|%s|%t\n", err, path, change.Type)
			} else {
				self.watcher.NodeChange(path, ZkEvent(change.Type), childnodes)
				// log.Info("ZKManager|listenEvent|%s|%s|%s\n", path, change, childnodes)
			}

		case zk.EventNodeDataChanged:
			split := strings.Split(path, "/")
			//如果不是bind级别的变更则忽略
			if len(split) < 5 || strings.LastIndex(split[4], "-bind") <= 0 {
				continue
			}
			//获取一下数据
			binds, err := self.getBindData(path)
			if nil != err {
				log.Error("ZKManager|listenEvent|Changed|Get DATA|FAIL|%s|%s\n", err, path)
				//忽略
				continue
			}
			self.watcher.DataChange(path, binds)
			// log.Info("ZKManager|listenEvent|%s|%s|%s\n", path, change, binds)

		}

	}

}

func (self *BindExchanger) DataChange(path string, binds []*Binding) {

	//订阅关系变更才处理
	if strings.HasPrefix(path, KITEQ_SUB) {

		split := strings.Split(path, "/")
		//获取topic
		topic := split[3]
		groupId := split[4]
		self.lock.Lock()
		defer self.lock.Unlock()
		//开始处理变化的订阅关系
		self.onBindChanged(topic, groupId, binds)

	} else {
		log.Warn("BindExchanger|DataChange|非SUB节点变更|%s\n", path)
	}

}

func (self *ApnsConnection) sendMessage(msg *entry.Message) error {
	//将当前的msg强制设置为当前conn的id作为标识
	msg.ProcessId = self.connectionId

	err, packet := msg.Encode()
	if nil != err {
		return err
	}

	length, sendErr := self.conn.Write(packet)
	if nil != err || length != len(packet) {
		log.Warn("CONNECTION|SEND MESSAGE|FAIL|%s", err)
	} else {
		log.Debug("CONNECTION|SEND MESSAGE|SUCC")

	}

	return sendErr
}

/*
*重连zk
 */
func (self *ZKManager) reconnect() <-chan bool {
	ch := make(chan bool, 1)
	go func() {

		reconnTimes := int64(0)
		f := func() error {
			ss, eventChan, err := zk.Connect(strings.Split(self.zkhosts, ","), 5*time.Second)
			if nil != err {
				log.Warn("连接zk失败.....%ds后重连任务重新发起...|", (reconnTimes+1)*5)
				return err
			} else {
				log.Info("重连ZK任务成功....")
				//初始化当前的状态
				self.session = ss
				self.eventChan = eventChan

				ch <- true
				close(ch)
				return nil
			}

		}
		//启动重连任务
		for !self.isClose {
			select {
			case <-time.After(time.Duration(reconnTimes * time.Second.Nanoseconds())):
				err := f()
				if nil != err {
					reconnTimes += 1
				} else {
					//重连成功则推出
					break
				}
			}
		}

		//失败
		ch <- false
		close(ch)
	}()
	return ch
}

func (self *KiteClientManager) NodeChange(path string, eventType binding.ZkEvent, children []string) {

	//如果是订阅关系变更则处理
	if strings.HasPrefix(path, binding.KITEQ_SERVER) {
		//获取topic
		split := strings.Split(path, "/")
		if len(split) < 4 {
			//不合法的订阅璐姐
			log.Warn("KiteClientManager|ChildWatcher|INVALID SERVER PATH |%s|%t\n", path, children)
			return
		}
		//获取topic
		topic := split[3]
		//search topic
		for _, t := range self.topics {
			if t == topic {
				self.onQServerChanged(topic, children)
				break
			}
		}
	}
}

func (self *TxAckHandler) Process(ctx *DefaultPipelineContext, event IEvent) error {

	// log.Debug("TxAckHandler|Process|%s|%t\n", self.GetName(), event)

	pevent, ok := self.cast(event)
	if !ok {
		return ERROR_INVALID_EVENT_TYPE
	}

	h := pevent.txPacket.GetHeader()
	//提交或者回滚
	if pevent.txPacket.GetStatus() == int32(protocol.TX_COMMIT) {

		succ := self.kitestore.Commit(h.GetMessageId())

		if succ {
			//发起投递事件
			//启动异步协程处理分发逻辑
			preevent := NewDeliverPreEvent(h.GetMessageId(), h, nil)
			ctx.SendForward(preevent)

		} else {
			//失败了等待下次recover询问
			// log.Info("TxAckHandler|%s|Process|Commit|FAIL|%s|%s\n", self.GetName(), h.GetMessageId(), succ)
		}

	} else if pevent.txPacket.GetStatus() == int32(protocol.TX_ROLLBACK) {
		succ := self.kitestore.Rollback(h.GetMessageId())
		if !succ {
			log.Warn("TxAckHandler|%s|Process|Rollback|FAIL|%s|%s|%s\n", self.GetName(), h.GetMessageId(), pevent.txPacket.GetFeedback(), succ)
		}

	} else {
		//UNKNOWN其他的不处理

	}
	ctx.SendForward(&SunkEvent{})
	return nil
}

func (self *AcceptHandler) Process(ctx *pipe.DefaultPipelineContext, event pipe.IEvent) error {
	// log.Debug("AcceptHandler|Process|%s|%t\n", self.GetName(), event)

	ae, ok := self.cast(event)
	if !ok {
		return pipe.ERROR_INVALID_EVENT_TYPE
	}
	//这里处理一下ae,做一下校验
	var msg *store.MessageEntity
	switch ae.msgType {
	case protocol.CMD_DELIVER_ACK:
		//收到投递结果直接attach响应
		ae.remoteClient.Attach(ae.opaque, ae.msg)
		return nil
	case protocol.CMD_HEARTBEAT:
		hb := ae.msg.(*protocol.HeartBeat)
		event = pipe.NewHeartbeatEvent(ae.remoteClient, ae.opaque, hb.GetVersion())
		ctx.SendForward(event)
		return nil

	case protocol.CMD_BYTES_MESSAGE:
		msg = store.NewMessageEntity(protocol.NewQMessage(ae.msg.(*protocol.BytesMessage)))
	case protocol.CMD_STRING_MESSAGE:
		msg = store.NewMessageEntity(protocol.NewQMessage(ae.msg.(*protocol.StringMessage)))
	default:
		//这只是一个bug不支持的数据类型能给你
		log.Warn("AcceptHandler|Process|%s|%t\n", INVALID_MSG_TYPE_ERROR, ae.msg)
	}

	if nil != msg {
		msg.PublishTime = time.Now().Unix()
		msg.KiteServer = self.kiteserver
		deliver := newPersistentEvent(msg, ae.remoteClient, ae.opaque)
		ctx.SendForward(deliver)
		return nil
	}
	return INVALID_MSG_TYPE_ERROR
}

func (self *ApnsConnection) dial() error {

	config := tls.Config{}
	config.Certificates = []tls.Certificate{self.cert}
	config.InsecureSkipVerify = true
	conn, err := tls.Dial("tcp", self.hostport, &config)
	if nil != err {
		//connect fail
		log.Warn("CONNECTION|%s|DIAL CONNECT|FAIL|%s|%s", self.name(), self.hostport, err.Error())
		return err
	}

	// conn.SetDeadline(0 * time.Second)
	for {
		state := conn.ConnectionState()
		if state.HandshakeComplete {
			log.Info("CONNECTION|%s|HANDSHAKE SUCC", self.name())
			break
		}
		time.Sleep(1 * time.Second)
	}
	self.conn = conn
	return nil
}

func (self *ZKManager) Start() {
	if len(self.zkhosts) <= 0 {
		log.Warn("使用默认zkhosts！|localhost:2181\n")
		self.zkhosts = "localhost:2181"
	} else {
		log.Info("使用zkhosts:[%s]！\n", self.zkhosts)
	}

	ss, eventChan, err := zk.Connect(strings.Split(self.zkhosts, ","), 5*time.Second)
	if nil != err {
		panic("连接zk失败..." + err.Error())
		return
	}

	exist, _, err := ss.Exists(KITEQ)
	if nil != err {
		ss.Close()
		panic("无法创建KITEQ " + err.Error())

	}

	if !exist {
		resp, err := ss.Create(KITEQ, nil, zk.CreatePersistent, zk.WorldACL(zk.PermAll))
		if nil != err {
			ss.Close()
			panic("NewZKManager|CREATE ROOT PATH|FAIL|" + KITEQ + "|" + err.Error())
		} else {
			log.Info("NewZKManager|CREATE ROOT PATH|SUCC|%s", resp)
		}
	}

	self.session = ss
	self.isClose = false
	self.eventChan = eventChan
	go self.listenEvent()
}