func (self *KiteQServer) Start() {

	self.remotingServer = server.NewRemotionServer(self.kc.server, self.kc.rc,
		func(rclient *client.RemotingClient, p *packet.Packet) {
			event := pipe.NewPacketEvent(rclient, p)
			err := self.pipeline.FireWork(event)
			if nil != err {
				log.Error("RemotingServer|onPacketRecieve|FAIL|%s|%t\n", err, p)
			} else {
				// log.Debug("RemotingServer|onPacketRecieve|SUCC|%s|%t\n", rclient.RemoteAddr(), packet)
			}
		})

	err := self.remotingServer.ListenAndServer()
	if nil != err {
		log.Crashf("KiteQServer|RemotionServer|START|FAIL|%s|%s\n", err, self.kc.server)
	} else {
		log.Info("KiteQServer|RemotionServer|START|SUCC|%s\n", self.kc.server)
	}
	//推送可发送的topic列表并且获取了对应topic下的订阅关系
	succ := self.exchanger.PushQServer(self.kc.server, self.kc.topics)
	if !succ {
		log.Crashf("KiteQServer|PushQServer|FAIL|%s|%s\n", err, self.kc.topics)
	} else {
		log.Info("KiteQServer|PushQServer|SUCC|%s\n", self.kc.topics)
	}

	//开启流量统计
	self.kc.flowstat.Start()

	//开启recover
	self.recoverManager.Start()

}

func (self *ConnPool) Shutdown() {
	self.mutex.Lock()
	defer self.mutex.Unlock()
	self.running = false

	for i := 0; i < 3; {
		//等待五秒中结束
		time.Sleep(5 * time.Second)
		if self.workPool.Len() <= 0 {
			break
		}

		log.Info("CONNECTION POOL|CLOSEING|WORK POOL SIZE|:%d", self.workPool.Len())
		i++
	}

	var idleconn *IdleConn
	//关闭掉空闲的client
	for e := self.idlePool.Front(); e != nil; e = e.Next() {
		idleconn = e.Value.(*IdleConn)
		idleconn.conn.Close()
		self.idlePool.Remove(e)
		idleconn = nil
	}

	log.Info("CONNECTION_POOL|SHUTDOWN")
}

func NewApnsHttpServer(option Option) *ApnsHttpServer {

	feedbackChan := make(chan *entry.Feedback, 1000)
	var apnsClient *apns.ApnsClient
	if option.startMode == STARTMODE_MOCK {
		//初始化mock apns
		apnsClient = apns.NewMockApnsClient(option.cert,
			option.pushAddr, chan<- *entry.Feedback(feedbackChan), option.feedbackAddr, entry.NewCycleLink(3, option.storageCapacity))
		log.Info("MOCK APNS HTTPSERVER IS STARTING ....")
	} else {
		//初始化apns
		apnsClient = apns.NewDefaultApnsClient(option.cert,
			option.pushAddr, chan<- *entry.Feedback(feedbackChan), option.feedbackAddr, entry.NewCycleLink(3, option.storageCapacity))
		log.Info("ONLINE APNS HTTPSERVER IS STARTING ....")
	}

	server := &ApnsHttpServer{feedbackChan: feedbackChan,
		apnsClient: apnsClient, expiredTime: option.expiredTime}

	//创建http
	server.httpserver = NewMomoHttpServer(option.bindAddr, nil)

	go server.dial(option.bindAddr)

	return server
}

func (self *StmtPool) Shutdown() {
	self.mutex.Lock()
	defer self.mutex.Unlock()
	self.running = false
	//等待五秒中结束
	time.Sleep(1 * time.Second)
	for i := 0; i < 3; {

		if self.numWork <= 0 {
			break
		}

		log.Info("Statment Pool|CLOSEING|WORK POOL SIZE|:%d\n", self.numWork)
		i++
	}

	var idleStmt *IdleStmt
	//关闭掉空闲的client
	for e := self.idlePool.Front(); e != nil; e = e.Next() {
		idleStmt = e.Value.(*IdleStmt)
		idleStmt.stmt.Close()
		self.idlePool.Remove(e)
		idleStmt = nil
	}

	log.Info("Statment Pool|SHUTDOWN")
}

func (self *ApnsHttpServer) dial(hp string) {

	log.Info("APNS HTTPSERVER IS STARTING ....")
	http.HandleFunc("/apns/push", self.handlePush)
	http.HandleFunc("/apns/feedback", self.handleFeedBack)

	err := self.httpserver.ListenAndServe()
	if nil != err {
		log.Error("APNSHTTPSERVER|LISTEN|FAIL|%s", err)
	} else {
		log.Info("APNSHTTPSERVER|LISTEN|SUCC|%s .....", hp)
	}

}

//启动
func (self *KiteClientManager) Start() {

	self.zkManager = binding.NewZKManager(self.zkAddr)
	//注册kiteqserver的变更
	self.zkManager.RegisteWather(PATH_KITEQ_SERVER, self)

	hostname, _ := os.Hostname()
	//推送本机到
	err := self.zkManager.PublishTopics(self.topics, self.ga.GroupId, hostname)
	if nil != err {
		log.Crashf("KiteClientManager|PublishTopics|FAIL|%s|%s\n", err, self.topics)
	} else {
		log.Info("KiteClientManager|PublishTopics|SUCC|%s\n", self.topics)
	}

outter:
	for _, b := range self.binds {
		for _, t := range self.topics {
			if t == b.Topic {
				continue outter
			}
		}
		self.topics = append(self.topics, b.Topic)
	}

	for _, topic := range self.topics {

		hosts, err := self.zkManager.GetQServerAndWatch(topic)
		if nil != err {
			log.Crashf("KiteClientManager|GetQServerAndWatch|FAIL|%s|%s\n", err, topic)
		} else {
			log.Info("KiteClientManager|GetQServerAndWatch|SUCC|%s|%s\n", topic, hosts)
		}
		self.onQServerChanged(topic, hosts)
	}

	if len(self.kiteClients) <= 0 {
		log.Crashf("KiteClientManager|Start|NO VALID KITESERVER|%s\n", self.topics)
	}

	if len(self.binds) > 0 {
		//订阅关系推送，并拉取QServer
		err = self.zkManager.PublishBindings(self.ga.GroupId, self.binds)
		if nil != err {
			log.Crashf("KiteClientManager|PublishBindings|FAIL|%s|%s\n", err, self.binds)
		}
	}

}

//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 *ClientManager) Shutdown() {
	self.reconnectManager.stop()
	for _, c := range self.allClients {
		c.Shutdown()
	}
	log.Info("ClientManager|Shutdown....")
}

func (self *ClientManager) removeClient(hostport string) {
	ga, ok := self.groupAuth[hostport]
	if ok {
		//删除分组
		delete(self.groupAuth, hostport)
		//删除group中的client
		gc, ok := self.groupClients[ga.GroupId]
		if ok {
			for i, cli := range gc {
				if cli.RemoteAddr() == hostport {
					self.groupClients[ga.GroupId] = append(gc[0:i], gc[i+1:]...)
					break
				}
			}
		}

		//删除hostport->client的对应关系
		c, ok := self.allClients[hostport]
		if ok {
			c.Shutdown()
			delete(self.allClients, hostport)
		}
	}

	log.Info("ClientManager|removeClient|%s...\n", hostport)
}

//发布topic对应的server
func (self *ZKManager) PublishQServer(hostport string, topics []string) error {

	for _, topic := range topics {
		qpath := KITEQ_SERVER + "/" + topic
		spath := KITEQ_SUB + "/" + topic
		ppath := KITEQ_PUB + "/" + topic

		//创建发送和订阅的根节点
		self.traverseCreatePath(ppath, nil, zk.CreatePersistent)
		// self.session.ExistsW(ppath)
		self.traverseCreatePath(spath, nil, zk.CreatePersistent)
		self.session.ExistsW(spath)

		//先删除当前这个临时节点再注册 避免监听不到临时节点变更的事件
		self.session.Delete(qpath+"/"+hostport, -1)

		//注册当前节点
		path, err := self.registePath(qpath, hostport, zk.CreateEphemeral, nil)
		if nil != err {
			log.Error("ZKManager|PublishQServer|FAIL|%s|%s/%s\n", err, qpath, hostport)
			return err
		}
		log.Info("ZKManager|PublishQServer|SUCC|%s\n", path)
	}

	//注册当前的kiteqserver
	self.session.Delete(KITEQ_ALIVE_SERVERS+"/"+hostport, -1)
	self.registePath(KITEQ_ALIVE_SERVERS, hostport, zk.CreateEphemeral, nil)
	self.registePath(KITEQ_ALL_SERVERS, hostport, zk.CreatePersistent, nil)
	return nil
}

//发布订阅关系
func (self *ZKManager) PublishBindings(groupId string, bindings []*Binding) error {

	//按topic分组
	groupBind := make(map[string][]*Binding, 10)
	for _, b := range bindings {
		g, ok := groupBind[b.Topic]
		if !ok {
			g = make([]*Binding, 0, 2)
		}
		b.GroupId = groupId
		g = append(g, b)
		groupBind[b.Topic] = g
	}

	for topic, binds := range groupBind {
		data, err := MarshalBinds(binds)
		if nil != err {
			log.Error("ZKManager|PublishBindings|MarshalBind|FAIL|%s|%s|%t\n", err, groupId, binds)
			return err
		}

		createType := zk.CreatePersistent

		path := KITEQ_SUB + "/" + topic
		//注册对应topic的groupId //注册订阅信息
		succpath, err := self.registePath(path, groupId+"-bind", createType, data)
		if nil != err {
			log.Error("ZKManager|PublishTopic|Bind|FAIL|%s|%s/%s\n", err, path, binds)
			return err
		} else {
			log.Info("ZKManager|PublishTopic|Bind|SUCC|%s|%s\n", succpath, binds)
		}
	}
	return nil
}

func main() {

	runtime.GOMAXPROCS(8)
	startMode := flag.Int("startMode", 1, " 0 为mock ,1 为正式")
	bindAddr := flag.String("bindAddr", ":17070", "-bindAddr=:17070")
	certPath := flag.String("certPath", "./cert.pem", "-certPath=xxxxxx/cert.pem or -certPath=http://")
	keyPath := flag.String("keyPath", "./key.pem", "-keyPath=xxxxxx/key.pem or -keyPath=http://")
	runMode := flag.Int("runMode", 0, "-runMode=1(online) ,0(sandbox)")
	storeCap := flag.Int("storeCap", 1000, "-storeCap=100000  //重发链条长度")
	logxml := flag.String("log", "log.xml", "-log=log.xml //log配置文件")
	pprofPort := flag.String("pprof", ":9090", "pprof=:9090 //端口")
	flag.Parse()

	go func() {
		if len(*pprofPort) > 0 {
			addr, _ := net.ResolveTCPAddr("tcp4", *bindAddr)
			log.Error(http.ListenAndServe(addr.IP.String()+*pprofPort, nil))
		}
	}()

	//加载log4go的配置
	log.LoadConfiguration(*logxml)

	//设置启动项
	option := server.NewOption(*startMode, *bindAddr, *certPath, *keyPath, *runMode, *storeCap)
	apnsserver := server.NewApnsHttpServer(option)
	ch := make(chan os.Signal, 1)
	signal.Notify(ch, os.Kill)
	//kill掉的server
	<-ch
	apnsserver.Shutdown()

	log.Info("APNS SERVER IS STOPPED!")

}

//发布topic对应的server
func (self *ZKManager) PublishQServer(hostport string, topics []string) error {

	for _, topic := range topics {

		qpath := KITEQ_SERVER + "/" + topic
		spath := KITEQ_SUB + "/" + topic
		ppath := KITEQ_PUB + "/" + topic

		//创建发送和订阅的根节点
		self.traverseCreatePath(ppath, nil, zk.CreatePersistent)
		// self.session.ExistsW(ppath)
		self.traverseCreatePath(spath, nil, zk.CreatePersistent)
		self.session.ExistsW(spath)

		//注册当前节点
		path, err := self.registePath(qpath, hostport, zk.CreateEphemeral, nil)
		if nil != err {
			log.Error("ZKManager|PublishQServer|FAIL|%s|%s/%s\n", err, qpath, hostport)
			return err
		}
		log.Info("ZKManager|PublishQServer|SUCC|%s\n", path)
	}

	return nil
}

//关闭掉exchanger
func (self *BindExchanger) Shutdown() {
	//删除掉当前的QServer
	self.zkmanager.UnpushlishQServer(self.kiteqserver, self.topics)
	time.Sleep(10 * time.Second)
	self.zkmanager.Close()
	log.Info("BindExchanger|Shutdown...")
}

//握手包
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() + "]")
}

func (self *FlowStat) Start() {

	go func() {
		t := time.NewTicker(1 * time.Second)
		for !self.stop {
			self.DeliverCount = self.DeliverFlow.Changes()
			line := fmt.Sprintf("%s\tdeliver:%d\tdeliver-go:%d\t", self.name, self.DeliverCount, self.DeliverPool.Count())
			log.Info(line)
			if nil != self.Kitestore {
				log.Info(self.Kitestore.Monitor())
			}
			<-t.C
		}
		t.Stop()
	}()
}

func (self *KiteMysqlStore) Start() {

	count := SHARD_SEED
	//创建Hash的channel
	batchDelChan := make([]chan string, 0, count)
	batchUpChan := make([]chan *MessageEntity, 0, count)
	batchComChan := make([]chan string, 0, count)
	for i := 0; i < count; i++ {
		batchUpChan = append(batchUpChan, make(chan *MessageEntity, self.batchUpSize*2))
		batchDelChan = append(batchDelChan, make(chan string, self.batchDelSize*2))
		batchComChan = append(batchComChan, make(chan string, self.batchUpSize*2))
	}

	//批量的channel
	self.batchUpChan = batchUpChan
	self.batchDelChan = batchDelChan
	self.batchComChan = batchComChan

	//创建每种批量的preparedstmt
	stmts := make(map[batchType][][]*StmtPool, 4)
	for k, v := range self.sqlwrapper.batchSQL {
		btype := k
		pool := make([][]*StmtPool, 0, self.dbshard.ShardNum())
		//对每个shard构建stmt的pool
		for i := 0; i < self.dbshard.ShardNum(); i++ {
			innerPool := make([]*StmtPool, 0, self.dbshard.HashNum())
			for j, s := range v {
				psql := s
				db := self.dbshard.FindShardById(i*self.dbshard.HashNum() + j).master
				err, p := NewStmtPool(5, 10, 20, 1*time.Minute, func() (error, *sql.Stmt) {
					stmt, err := db.Prepare(psql)
					if nil != err {
						log.Error("StmtPool|Create Stmt|FAIL|%s|%s\n", err, psql)
						return err, nil
					}
					return nil, stmt
				})
				if nil != err {
					log.Error("NewKiteMysql|NewStmtPool|FAIL|%s\n", err)
					panic(err)
				}
				innerPool = append(innerPool, p)
			}
			pool = append(pool, innerPool)
		}
		stmts[btype] = pool
	}

	self.stmtPools = stmts

	for i := 0; i < count; i++ {
		// log.Printf("KiteMysqlStore|start|SQL|%s\n|%s\n", sqlu, sqld)
		self.startBatch(i, self.batchUpChan[i],
			self.batchDelChan[i], self.batchComChan[i])
	}
	log.Info("KiteMysqlStore|Start...")
}

func (self *ReconnectManager) stop() {
	self.allowReconnect = false
	self.lock.Lock()
	defer self.lock.Unlock()
	for _, t := range self.timers {
		t.Stop()
	}
	log.Info("ReconnectManager|stop...")
}

func (self *KiteClientManager) remointflow() {
	go func() {
		t := time.NewTicker(1 * time.Second)
		for {
			log.Info(self.rc.FlowStat.Monitor())
			<-t.C
		}
	}()
}

func (self *KiteQServer) Shutdown() {
	//先关闭exchanger让客户端不要再输送数据
	self.exchanger.Shutdown()
	self.recoverManager.Stop()
	self.kitedb.Stop()
	self.clientManager.Shutdown()
	self.remotingServer.Shutdown()
	log.Info("KiteQServer|Shutdown...")

}