//Run the app.
func Run() (err error) {
	var wg sync.WaitGroup
	log.Info("Starting gomasticate")
	conf, err := NewConf("conf.yaml")
	if nil != err {
		log.Error(err)
		return
	}
	log.Info(conf)
	chewChan := make(chan *messaging.Food, 2000)
	swallowChan := make(chan *messaging.Food, 4000)

	done := make(chan interface{})

	wg.Add(2)
	go lips.OpenWide(chewChan, done, &wg, conf.LipsPort())
	go chew.Chew(chewChan, swallowChan, &wg)

	sw := swallow.NewSwallow(conf.EsHost(), swallowChan, 10)

	//handle signals
	c := make(chan os.Signal, 1)
	s := make(chan int, 1)
	signal.Notify(c)
	go Death(c, s)
	death := <-s //time for shutdown
	log.Debug("Death return code: ", death)
	close(done)
	sw.Close()
	log.Info("Waiting for goroutines to finish...")
	wg.Wait()
	log.Info("Exiting")
	return
}

func CheckIn(path string) {
	log.Info("Starting db checking background script: " + path)
	ticker := time.NewTicker(10 * time.Second)
	quit := make(chan struct{})
	go func() {
		for {
			select {
			case <-ticker.C:
				nodeinfo.UpdateConnection()
				log.Info("Checking all changed stuff in db for: " + path)
				listener := utils.GetListenerFromDir(path)
				items, err := datastore.CheckIn(listener)
				if err != nil {
					log.Infof("Error occurred getting data for %s (%s): %+v", listener, err.Error(), err)
				}
				cfg := utils.GetConfig()
				handleDataChanges(items, cfg.Listeners[listener], listener)
			// @TODO: check that db knows Im alive.
			case <-quit:
				ticker.Stop()
				return
			}
		}
	}()
}

func loadNotifiers(app *ApplicationContext) error {
	// Set up the Emailer, if configured
	if len(app.Config.Email) > 0 {
		log.Info("Configuring Email notifier")
		emailer, err := NewEmailer(app)
		if err != nil {
			log.Criticalf("Cannot configure email notifier: %v", err)
			return err
		}
		app.Emailer = emailer
	}

	// Set up the HTTP Notifier, if configured
	if app.Config.Httpnotifier.Url != "" {
		log.Info("Configuring HTTP notifier")
		httpnotifier, err := NewHttpNotifier(app)
		if err != nil {
			log.Criticalf("Cannot configure HTTP notifier: %v", err)
			return err
		}
		app.HttpNotifier = httpnotifier
	}

	return nil
}

func (m *Management) Shutdown() {
	log.Info("Management: shutdown")
	close(m.shutdown)

	m.waitGroup.Wait()
	log.Info("Management: shutdown done")
}

func main() {
	f, err := os.Open("ok.txt")
	if err != nil {
		log.Info(err)
		return
	}
	defer f.Close()
	stat, err := f.Stat()
	if err != nil {
		log.Error("stat err")
	}
	data := make([]byte, stat.Size())
	_, err = f.Read(data)
	if err != nil {
		log.Error("read err")
	}
	dataStr := string(data)

	log.Info(dataStr)

	dirInfo()

	demoList()

}

//Simple pusher for testing
func PusherProto(count int, finished chan int, msg *messaging.Food, port int) {
	log.Info("Starting pusher")
	socket, err := nano.NewPushSocket()
	if nil != err {
		log.Error(err)
	}
	defer socket.Close()
	socket.SetSendTimeout(500 * time.Millisecond)
	sport := strconv.Itoa(port)
	_, err = socket.Connect("tcp://localhost:" + sport)
	if nil != err {
		log.Error(err)
		return
	}
	log.Info("Connected and ready to send data")
	tot := 0
	for {
		bytes, _ := msg.Marshal()
		_, err := socket.Send(bytes, 0) //blocking
		if nil != err {
			log.Error(err)
			continue
		} else {
			tot++
		}
		if tot >= count {
			break
		}
	}
	log.Info("Finished sending data exiting")
	finished <- tot
}

func InitialSync() {
	cfg := utils.GetConfig()
	log.Info("Verifying DB Tables")
	datastore.CreateDB()
	log.Info("Initial sync starting...")

	for key, listener := range cfg.Listeners {

		// First check to see if the table is empty and do a full import false == not empty
		if datastore.CheckEmpty(key) == false {
			// Database is not empty so pull the updates and match locally
			items := datastore.FetchAll(key)
			handleDataChanges(items, listener, key)
		} else {
			// Database is empty so lets import
			fsItems := utils.ListFilesInDir(listener.Directory)
			for _, item := range fsItems {
				success := datastore.Insert(key, item)
				if success != true {
					log.Infof("An error occurred inserting %x to database", item)
				}
				if !item.IsDir {
					storage.PutFile(item.Filename, key)
				}

			}
		}

	}

	log.Info("Initial sync completed...")
}

func (ss *Session) heartBeat() chan bool {
	done := make(chan bool)
	ss.heartTicker = done
	go func() {
		t := time.Duration(ss.server.Config.HeartbeatInterval) * time.Second
		ticker := time.NewTicker(t)
		loop := true
		for loop {
			select {
			case <-ticker.C:
				err := ss.defaultNS.sendPacket(new(heartbeatPacket))
				if err != nil {

				}
				log.Info("sent heart beat missed = ", ss.missedHeartbeats)
				ss.missedHeartbeats += 1
				// TODO: Configurable
				if ss.missedHeartbeats > 2 {
					log.Info("heartBeat missedHeartbeats ", ss.SessionId)
					ss.Close("")
					loop = false
				}
			case <-done:
				log.Infof("[%s] stop heartBeat", ss.SessionId)
				ticker.Stop()
				//ss.heartTicker = nil
				return
			}
		}
	}()
	return done
}

func (ss *Session) RawMessage(msg []byte) error {
	log.Trace("RawMessage ", string(msg))
	packet, err := decodePacket(msg)
	if err != nil {
		log.Info("decodePacket error ", err, string(msg))
		return nil
	}
	if packet == nil {
		log.Info("packet == nil ")
		return nil

	}

	if packet.EndPoint() == "" {
		if err := ss.onPacket(packet); err != nil {
			log.Error(err)
			return nil
		}
	}

	ns := ss.Of(packet.EndPoint())
	if ns == nil {
		return nil
	}
	ns.onPacket(packet)
	return nil
}

func calcDailySummary(now time.Time, config StartupConfig, runningConfig RunningConfig) {
	log.Infof("lastSummaryTime is %v", runningConfig.LastSummaryTime)
	if runningConfig.LastSummaryTime.Day() != now.Day() {
		startTime := now.Truncate(24 * time.Hour).Add(-24 * time.Hour)
		endTime := startTime.Add(24 * time.Hour)
		log.Info("Summarizing from ", startTime, " (", startTime.Unix(), ") to ", endTime, " (", endTime.Unix(), ")")

		// influx connection
		influxClient, err := influxConnect(config, runningConfig)
		if err != nil {
			log.Error("Could not connect to InfluxDb to get daily summary stats!!")
			errHndlr(err, ERROR)
			return
		}

		bp, _ := influx.NewBatchPoints(influx.BatchPointsConfig{
			Database:        "daily_stats",
			Precision:       "s",
			RetentionPolicy: config.DailySummaryRetentionPolicy,
		})

		calcDailyMaxGbps(influxClient, bp, startTime, endTime, config)
		calcDailyBytesServed(influxClient, bp, startTime, endTime, config)
		log.Info("Collected daily stats @ ", now)
	}
}

/**
 * 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 Chew(chewChan <-chan *messaging.Food, swallowChan chan *messaging.Food, wg *sync.WaitGroup) {
	log.Info("Let the chewing begin!")
	defer close(swallowChan)
	r := rep.NewReporter()
	r.RegisterStatWIndex("chew", "good")
	for msg := range chewChan {
		if nil != msg {
			//parsing work here probably change what our message type looks like when swallowed

			date := time.Unix(0, msg.GetTimeNano()).UTC()
			fmtDate := date.Format("2006-01-02")
			indexType := "all"
			customerId := "id" //should exist eventually
			index := "documents-" + customerId + "-" + fmtDate
			msg.Index = &index
			msg.IndexType = &indexType
			r.AddStatWIndex("chew", 1, "good")
			swallowChan <- msg
		}
	}
	log.Info("Done chewing")
	log.Flush()
	wg.Done()

}

func RunReceiverRPC() {
	log.Info("Starting Receiver RPC\n")
	ln, err := net.ListenUnix("unix", appAddr)
	if err != nil {
		log.Errorf("Failed to start RPC server: %v", err)
		return
	}

	Receiver = receiverrpc.Receiver{}
	rpc.Register(&Receiver)
	log.Infof("RPC server is listening on %s\n", appAddr.String())

	defer func() {
		// close the listener sock
		log.Info("Closing listener socket.\n")
		ln.Close()
	}()

	for {
		ln.SetDeadline(time.Now().Add(time.Second))
		conn, err := ln.AcceptUnix()
		if err != nil {
			if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
				// just accept timeout, not an error
				continue
			}
			log.Error("Failed to accept: %s", err.Error())
			continue
		}
		rpc.ServeConn(conn)
	}
}

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 main() {
	log.Info("Initialize!")

	var proxy = service.Proxy{}
	var refresh = service.Refresh{}

	log.Info("Run!")

	refresh.Run(&proxy)
}

func CheckEmpty(table string) bool {
	setdbstoreEngine()
	empty := dbstore.CheckEmpty(table)
	if empty {
		log.Info("Database is EMPTY, starting creation")
	} else {
		log.Info("Using existing table: " + table)
	}
	return empty
}

func libWithSealogMain() {
	defer library.FlushLog()
	defer log.Flush()
	loadAppConfig()
	log.Info("App started")
	log.Info("Config loaded")

	// Disable library log
	log.Info("* Disabled library log test")
	library.DisableLog()
	calcF2()
	log.Info("* Disabled library log tested")

	// Use a special logger for library
	log.Info("* Special output test")
	specialOutputConfig()
	calcF2()
	log.Info("* Special output tested")

	// Use the same logger for both app and library
	log.Info("* Same output test")
	sameOutputConfig()
	calcF2()
	log.Info("* Same output tested")

	log.Info("App finished")
}

func (fw *TrecFileWriter) WriteAllTokens() {
	log.Debugf("Monitoring the writer channel")
	for t := range fw.StringChan {
		log.Debugf("Received %s. Writing it out to disk.", *t)
		fw.file.WriteString(*t + "\n")
	}
	fw.file.Close()
	log.Info("Sending exit signal")
	fw.Done <- true
	log.Info("Writer, out!")
}

func (c *Conn) writeLoop() {
	for {
		select {
		case <-c.exitChan:
			clog.Info("breaking out of writeLoop")
			// Indicate drainReady because we will not pull any more off msgResponseChan
			close(c.drainReady)
			goto exit
		case cmd := <-c.cmdChan:
			err := c.WriteCommand(cmd)
			if err != nil {
				clog.Errorf("error sending command %s - %s", cmd, err)
				c.close()
				continue
			}
		case resp := <-c.msgResponseChan:
			// Decrement this here so it is correct even if we can't respond to nsqd
			msgsInFlight := atomic.AddInt64(&c.messagesInFlight, -1)

			if resp.success {
				clog.Tracef("FIN %s", resp.msg.ID)
				c.delegate.OnMessageFinished(c, resp.msg)
				if resp.backoff {
					c.delegate.OnResume(c)
				}
			} else {
				clog.Tracef("REQ %s", resp.msg.ID)
				c.delegate.OnMessageRequeued(c, resp.msg)
				if resp.backoff {
					c.delegate.OnBackoff(c)
				}
			}

			err := c.WriteCommand(resp.cmd)
			if err != nil {
				clog.Errorf("error sending command %s - %s", resp.cmd, err)
				c.close()
				continue
			}

			if msgsInFlight == 0 &&
				atomic.LoadInt32(&c.closeFlag) == 1 {
				c.close()
				continue
			}
		}
	}

exit:
	c.wg.Done()
	clog.Info("writeLoop exiting")
}

// run starts the asynchronous run-loop connecting to RabbitMQ
func (t *rabbitTransport) run() {
	initConn := func() *RabbitConnection {
		conn := NewRabbitConnection()
		t.connM.Lock()
		defer t.connM.Unlock()
		t.conn = conn
		select {
		case <-t.connReady:
			// Only swap connReady if it's already closed
			t.connReady = make(chan struct{})
		default:
		}
		return conn
	}
	conn := initConn()

	t.tomb.Go(func() error {
		defer func() {
			t.killListeners()
			conn.Close()
			log.Info("[Typhon:RabbitTransport] Dead; connection closed")
		}()

	runLoop:
		for {
			log.Info("[Typhon:RabbitTransport] Run loop connecting…")
			select {
			case <-t.tomb.Dying():
				return nil

			case <-conn.Init():
				log.Info("[Typhon:RabbitTransport] Run loop connected")
				t.listenReplies()

				select {
				case <-t.tomb.Dying():
					// Do not loop again
					return nil
				default:
					conn.Close()
					conn = initConn()
					continue runLoop
				}

			case <-time.After(connectTimeout):
				log.Criticalf("[Typhon:RabbitTransport] Run loop timed out after %s waiting to connect",
					connectTimeout.String())
				return ErrCouldntConnect
			}
		}
	})
}