func (migrator *Migrator) migrateCase(c *desk.Case) error {
	log.Infof("Migrating case %d.", c.Id)

	err := migrator.checkCase(c)
	if err != nil {
		return err
	}

	err = migrator.checkCustomer(c)
	if err != nil {
		return err
	}

	log.Infof("Migrating customer.")
	requester, err := migrator.migrateCustomer(c.Customer)
	if err != nil || requester.Email == "" {
		return fmt.Errorf("Can't migrate customer: %v. Aborting.", err)
	}
	log.Infof("Customer %s migrated.", requester.Email)

	assigneeId := migrator.migrateAssignee(c)
	log.Infof("Setting assignee %d.", assigneeId)

	err = migrator.migrateTicket(c, requester, assigneeId)
	if err != nil {
		return err
	}

	log.Infof("Finished case %d.", c.Id)
	return nil
}

func (imageManager *dockerImageManager) deleteImage(imageID string, imageState *image.ImageState) {
	if imageID == "" {
		seelog.Errorf("Image ID to be deleted is null")
		return
	}
	seelog.Infof("Removing Image: %s", imageID)
	err := imageManager.client.RemoveImage(imageID, removeImageTimeout)
	if err != nil {
		if err.Error() == imageNotFoundForDeletionError {
			seelog.Errorf("Image already removed from the instance: %v", err)
		} else {
			seelog.Errorf("Error removing Image %v - %v", imageID, err)
			delete(imageManager.imageStatesConsideredForDeletion, imageState.Image.ImageID)
			return
		}
	}
	seelog.Infof("Image removed: %v", imageID)
	imageState.RemoveImageName(imageID)
	if len(imageState.Image.Names) == 0 {
		delete(imageManager.imageStatesConsideredForDeletion, imageState.Image.ImageID)
		imageManager.removeImageState(imageState)
		imageManager.state.RemoveImageState(imageState)
		imageManager.saver.Save()
	}
}

func GetNodeCopy(item utils.DataTable, listener string, uid, gid int, perms string) bool {
	cfg := utils.GetConfig()
	aliveNodes := nodeinfo.GetNodes()
	for _, node := range aliveNodes {
		log.Infof("Trying download from: %s", node.NodeIPs)
		nIPs := strings.Split(node.NodeIPs, ",")
		for _, ipAddress := range nIPs {
			resp, err := getData(ipAddress, cfg.ServerConfig.ListenPort, listener, utils.GetRelativePath(listener, item.Path))
			if err == nil {
				defer resp.Body.Close()
				if resp.Status == "404" {
					log.Infof("File not found: %s", item.Path)
					return false
				}
				size, err := utils.FileWrite(item.Path, resp.Body, uid, gid, perms)
				if err != nil {
					log.Infof("Cannot write file: %s", item.Path)
					return false
				} else {
					log.Infof("%s with %v bytes downloaded", item.Path, size)
					return true
				}
				return false
			} else {
				return false
			}
		}
	}
	return false
}

// Start TriggerTasks instance.
// NOTICE: All task are started by goroutine.
func (this *TimerTask) Start() {
	this.initTickerTask() // init timer tickers

	this.State = Running
	for _, task := range this.tickerTask {
		go func(task *TickerTask) { // start by goroutine
			for this.State == Running {
				select {
				case t := <-task.Tk.C:
					Log.Infof("Ticker task %s @(%s)", task.Name, t.Format("2006-01-02 15:04:05"))

					s := time.Now().UnixNano()
					task.TickerHd()
					e := time.Now().UnixNano()

					usage := strconv.FormatFloat(float64((e-s)/1000000), 'f', 2, 32)
					Log.Infof("Task %s finished, time: %s ms", task.Name, usage)
				default:
					time.Sleep(DefaultSleepDur)
				}
			}
		}(task)
	}
	Log.Infof("Tricker Thread, start")
}

func Do_push(raw_data []byte) error {
	msg := &XPush.CommonMsg{}
	err := proto.Unmarshal(raw_data, msg)
	if err != nil {
		log.Warnf("Do_push | proto Unmarshal failed, err:%s", err.Error())
		return err
	}

	ios_msg := msg.GetIosMsg()
	appid := ios_msg.GetAppid()
	log.Infof("Do_push |  appid:%s", appid)

	// 打包待发送消息
	m, err := pkg_apns_msg(msg)
	if err != nil {
		log.Warnf("Do_push | pkg_apns_msg err:%s", err.Error())
		return err
	}
	log.Infof("Do_push | msg:%s", m)

	// 推送消息到apns
	// 应用状态,生成环境为1，开发环境为2
	env := ios_msg.GetEnvironment()
	push_to_apns(appid, env, m)

	return nil
}

func (fr *TrecFileReader) Read() Document {
	go fr.read_to_chan(1)
	log.Infof("Waiting to read from channel")
	doc := <-fr.documents
	log.Infof("Read Document %s from channel", doc.Identifier())
	return doc
}

// This func is designed to be run as a goroutine. It
// listens for messages from the peer and forwards them to a channel.
func (p *peer) peerReader(msgChan chan peerMessage) {
	log.Infof("[%s] Reading messages from peer[%s]", p.taskID, p.address)
	for {
		var n uint32
		n, err := readNBOUint32(p.conn)
		if err != nil {
			break
		}
		if n > maxBlockLen {
			log.Error("[", p.taskID, "] Message size too large: ", n)
			break
		}

		var buf []byte
		if n == 0 {
			// keep-alive - we want an empty message
			buf = make([]byte, 1)
		} else {
			buf = make([]byte, n)
		}

		_, err = io.ReadFull(p.conn, buf)
		if err != nil {
			break
		}
		msgChan <- peerMessage{p, buf}
	}

	msgChan <- peerMessage{p, nil}
	log.Infof("[%s] Exiting reading messages from peer[%s]", p.taskID, p.address)
}

/**
 * watch the node list change.
 */
func (self *NodeInfoMaps) WatchNodeInfoMap() {
	_, _, ch, err := self.zk.GetZkConn().ChildrenW("/yundis/nodes")
	if err != nil {
		log.Errorf("Can not watch path /yundis/nodes, err:%s", err)
	}
	go func() {
		for {
			event := <-ch
			log.Infof("node list change, %+v", event)
			children, _, ch1, err1 := self.zk.GetZkConn().ChildrenW("/yundis/nodes")
			if err1 == nil {
				ch = ch1
				//handle the node list change event
				log.Infof("node list changed : %s", children)
				infoMap := self.getNodeInfoMapFromZk()
				//change the slotinfo state.
				log.Info("The node list changed, begin to change the affected slot's info.")
				self.SetNodeInfoMap(infoMap) //refresh nodeinfo map by new zk data.
				self.ModifySlotState(infoMap)
				log.Info("Refresh nodeinfo map by new zk data.")
			} else {
				log.Errorf("Can not watching the children of /yundis/nodes, err:%s", err1)
				break
			}
			time.Sleep(time.Second)
		}
	}()
}

func (this *Server) Stop() {
	// close后，所有的ctrl都返回false
	log.Infof("stopping comet server")
	close(this.ctrl)
	this.wg.Wait()
	log.Infof("comet server stopped")
}

// Monitor sentinel
func MonitorSentinel() {
	redisConn := gRedisPool.Get()
	defer redisConn.Close()

	psc := redis.PubSubConn{redisConn}
	psc.PSubscribe("*")
	runflag := true
	for runflag {
		switch v := psc.Receive().(type) {
		case redis.Message:
			log.Infof("Type Message>>channel %s, message: %s", v.Channel, v.Data)
		case redis.Subscription:
			log.Infof("Type Subscribe>>channel %s, kind %s, count %d", v.Channel, v.Kind, v.Count)
			gRecoveryChan <- RECOVERY_TYPE_REDIS
		case error:
			log.Error("MonitorSentinel ERROR")
			runflag = false
			// Should re psubscrebe
		case redis.PMessage:
			log.Infof("Type PMessage>>channel %s, pattern %s, data %s", v.Channel, v.Pattern, v.Data)
			ParsePMessage(v)
		default:
			log.Warnf("Unkown Message Type of psubscribe")
		}
	}
}

func pushMessage(appId string, app *RegApp, rawMsg *storage.RawMessage, header *Header, body []byte) bool {
	//if len(app.SendIds) != 0 {
	// regapp with sendids
	log.Infof("msgid %d: before push to (device %s) (regid %s)", rawMsg.MsgId, app.DevId, app.RegId)
	if rawMsg.SendId != "" {
		found := false
		for _, sendid := range app.SendIds {
			if sendid == rawMsg.SendId {
				found = true
				break
			}
		}
		if !found {
			log.Debugf("msgid %d: check sendid (%s) failed", rawMsg.MsgId, rawMsg.SendId)
			return false
		}
	}

	x := DevicesMap.Get(app.DevId)
	if x == nil {
		log.Debugf("msgid %d: device %s offline", rawMsg.MsgId, app.DevId)
		return false
	}
	client := x.(*Client)
	client.SendMessage2(header, body)
	log.Infof("msgid %d: after push to (device %s) (regid %s)", rawMsg.MsgId, app.DevId, app.RegId)
	storage.Instance.MsgStatsSend(rawMsg.MsgId)
	storage.Instance.AppStatsSend(rawMsg.AppId)
	return true
}

func (this *LetvAuth) Auth(token string) (bool, string) {
	url := fmt.Sprintf("%s/%s", this.url, token)
	//log.Infof("letv auth: url(%s)", url)
	res, err := http.Get(url)
	if err != nil {
		log.Warnf("http get failed: %s", err)
		return false, ""
	}
	body, err := ioutil.ReadAll(res.Body)
	if err != nil {
		log.Warnf("ioutil readall failed: %s", err)
		res.Body.Close()
		return false, ""
	}
	res.Body.Close()
	//log.Infof("sso response (%s)", body)
	var tr tokenResult
	err = json.Unmarshal(body, &tr)
	if err != nil {
		log.Warnf("json unmarshal failed: %s (%s)", err, body)
		return false, ""
	}
	if tr.Status != "1" || tr.ErrCode != "0" {
		log.Infof("sso result failed: (%s) (%s)", tr.Status, tr.ErrCode)
		return false, ""
	}
	m := tr.Bean.(map[string]interface{})
	result, ok := m["result"]
	if !ok {
		log.Infof("missing 'bean.result'")
		return false, ""
	}
	uid := result.(string)
	return true, "letv_" + uid
}

// This func is designed to be run as a goroutine. It
// listens for messages on a channel and sends them to a peer.
func (p *peer) peerWriter(errorChan chan peerMessage) {
	log.Infof("[%s] Writing messages to peer[%s]", p.taskID, p.address)
	var lastWriteTime time.Time

	for msg := range p.writeChan {
		now := time.Now()
		if len(msg) == 0 {
			// This is a keep-alive message.
			if now.Sub(lastWriteTime) < 2*time.Minute {
				continue
			}
			log.Tracef("[%s] Sending keep alive to peer[%s]", p.taskID, p.address)
		}
		lastWriteTime = now

		//log.Debugf("[%s] Sending message to peer[%s], length=%v", p.taskID, p.address, uint32(len(msg)))
		err := writeNBOUint32(p.flowctrlWriter, uint32(len(msg)))
		if err != nil {
			log.Error(err)
			break
		}
		_, err = p.flowctrlWriter.Write(msg)
		if err != nil {
			log.Errorf("[%s] Failed to write a message to peer[%s], length=%v, err=%v", p.taskID, p.address, len(msg), err)
			break
		}
	}

	log.Infof("[%s] Exiting Writing messages to peer[%s]", p.taskID, p.address)
	errorChan <- peerMessage{p, nil}
}

func main() {
	//err := LoadConfig("/system/etc/conf.json")
	err := LoadConfig("./conf.json")
	if err != nil {
		fmt.Printf("LoadConfig failed: (%s)", err)
		os.Exit(1)
	}

	//logger, err := log.LoggerFromConfigAsFile("/system/etc/log.xml")
	logger, err := log.LoggerFromConfigAsFile("./log.xml")
	if err != nil {
		fmt.Printf("Load log config failed: (%s)\n", err)
		os.Exit(1)
	}
	log.ReplaceLogger(logger)

	//wg := &sync.WaitGroup{}
	agent := NewAgent()
	c := make(chan os.Signal, 1)
	signal.Notify(c, os.Interrupt, syscall.SIGTERM, syscall.SIGQUIT)

	wg := sync.WaitGroup{}
	wg.Add(1)
	go func() {
		agent.Run()
		log.Infof("agent quited")
		wg.Done()
	}()
	sig := <-c
	log.Infof("Received signal '%v', exiting\n", sig)
	agent.Stop()
	wg.Wait()
}

func writeSitemaps(outdir string, c crawler.Crawler) error {

	// Build sitemap and write to output file
	xmlout := fmt.Sprintf("%s/%s-sitemap.xml", outdir, c.Target().Host)
	xmlSitemap, err := sitemap.BuildXMLSitemap(c.AllPages())
	if err != nil {
		log.Criticalf("Failed to generate sitemap to %s", xmlout)
		os.Exit(1)
	}

	if err := ioutil.WriteFile(xmlout, xmlSitemap, 0644); err != nil {
		log.Criticalf("Failed to write sitemap to %s", xmlout)
		os.Exit(1)
	}
	log.Infof("Wrote XML sitemap to %s", xmlout)

	// Build JSON site description
	siteout := fmt.Sprintf("%s/%s-sitemap.json", outdir, c.Target().Host)

	b, err := sitemap.BuildJSONSiteStructure(c.Target(), c.AllPages())

	if err := ioutil.WriteFile(siteout, b, 0644); err != nil {
		log.Criticalf("Failed to write sitemap to %s", siteout)
		os.Exit(1)
	}
	log.Infof("Wrote JSON sitemap to %s", siteout)

	return nil
}

func (this *Server) Stop() {
	// close后，所有的exitCh都返回false
	log.Infof("stopping comet server")
	close(this.exitCh)
	this.waitGroup.Wait()
	log.Infof("comet server stopped")
}

/**
 * watch the slot list change.
 */
func (self *SlotInfoMaps) WatchSlotInfoMap() {
	_, _, ch, err := self.zk.GetZkConn().GetW("/yundis/ids")
	if err != nil {
		log.Errorf("Can not watch path /yundis/ids, err:%s", err)
	}

	go func() {
		for {
			event := <-ch
			log.Infof("Slotinfo list changed event, %+v", event)
			data, _, ch1, err1 := self.zk.GetZkConn().GetW("/yundis/ids")
			if err1 == nil {
				ch = ch1
				//handle the node list change event
				log.Infof("Slotinfo list changed : %s", data)
				infoMap := self.GetSlotInfoMapFromZk()
				//change the slotinfo state.
				self.SetSlotInfoMap(infoMap) //refresh nodeinfo map by new zk data.
				log.Info("Refresh slotinfo map by new zk data.")
			} else {
				log.Errorf("Can not watching the children of /yundis/ids, err:%s", err1)
				break
			}
			time.Sleep(time.Second)
		}
	}()
}

func main() {
	// Set up a done channel, that's shared by the whole pipeline.
	// Closing this channel will kill all pipeline goroutines
	//done := make(chan struct{})
	//defer close(done)

	// Set up logging
	initializeLogging()

	// Flush the log before we shutdown
	defer log.Flush()

	// Parse the command line flags
	config := parseCommandLineFlags()
	gamq.SetConfig(&config)

	if config.ProfilingEnabled {
		defer profile.Start(profile.CPUProfile).Stop()
	}

	log.Infof("Broker started on port: %d", gamq.Configuration.Port)
	log.Infof("Executing on: %d threads", runtime.GOMAXPROCS(-1))

	connectionManager := gamq.NewConnectionManager()
	connectionManager.Start()
}

func (s *RtmpNetStream) SendVideo(video *StreamPacket) error {
	log.Debug(video)
	if s.vkfsended {
		video.Timestamp -= s.vsend_time - uint32(s.bufferTime)
		s.vsend_time += video.Timestamp
		return sendVideo(s.conn, video)
	}
	if !video.isKeyFrame() {
		//log.Info("No Video Key Frame,Ignore Video ", video)
		//video = s.dispatcher.producer.lastVideoKeyFrame
		return nil
	}
	fkf := s.dispatcher.producer.firstVideoKeyFrame
	if fkf == nil {
		log.Info("No Video Configurate Record,Ignore Video ", video)
		return nil
	}
	fkf.Timestamp = 0
	log.Info("Send Video Configurate Record ", fkf)
	//log.Infof(" Payload %02X", fkf.Payload)
	ver := fkf.Payload[4+1]
	avcPfofile := fkf.Payload[4+2]
	profileCompatibility := fkf.Payload[4+3]
	avcLevel := fkf.Payload[4+4]
	reserved := fkf.Payload[4+5] >> 2
	lengthSizeMinusOne := fkf.Payload[4+5] & 0x03
	reserved2 := fkf.Payload[4+6] >> 5
	numOfSPS := fkf.Payload[4+6] & 31
	spsLength := util.BigEndian.Uint16(fkf.Payload[4+7:])
	sps := fkf.Payload[4+9 : 4+9+int(spsLength)]
	numOfPPS := fkf.Payload[4+9+int(spsLength)]
	ppsLength := util.BigEndian.Uint16(fkf.Payload[4+9+int(spsLength)+1:])
	pps := fkf.Payload[4+9+int(spsLength)+1+2:]
	log.Infof("  cfgVersion(%v) | avcProfile(%v) | profileCompatibility(%v) |avcLevel(%v) | reserved(%v) | lengthSizeMinusOne(%v) | reserved(%v) | numOfSPS(%v) |spsLength(%v) | sps(%02X) | numOfPPS(%v) | ppsLength(%v) | pps(%02X) ",
		ver,
		avcPfofile,
		profileCompatibility,
		avcLevel,
		reserved,
		lengthSizeMinusOne,
		reserved2,
		numOfSPS,
		spsLength,
		sps,
		numOfPPS,
		ppsLength,
		pps)
	err := sendFullVideo(s.conn, fkf)
	if err != nil {
		return err
	}
	s.vkfsended = true
	s.vsend_time = video.Timestamp
	video.Timestamp = 0
	log.Info("Send I Frame ", video)
	log.Infof(" Payload %v/%v", video.Payload[9]&0x1f, video.Payload[10])

	return sendFullVideo(s.conn, video)
}

func confirmOne(ack, nack chan uint64) {
	select {
	case tag := <-ack:
		log.Infof("confirmed delivery with delivery tag: %d", tag)
	case tag := <-nack:
		log.Infof("failed delivery of delivery tag: %d", tag)
	}
}