// getAddress gets the localhosts IPv4 address.
func GetAddress() (string, error) {
	name, err := os.Hostname()
	if err != nil {
		log.Print("Error Resolving Hostname:", err)
		return "", err
	}

	if ipv4host == "NONE" {
		as, err := net.LookupHost(name)
		if err != nil {
			return "", err
		}

		addr := ""

		for _, a := range as {
			log.Printf("a = %+v", a)
			if ipv4Reg.MatchString(a) {
				log.Print("matches")
				addr = a
			}
		}

		if addr == "" {
			err = errors.New("No IPv4 Address for Hostname")
		}
		return addr, err
	}
	return ipv4host, nil
}

// GenerateAndSave - generates cert and key and saves them on your disk
func GenerateAndSave(name, organization string, validity time.Duration) (tlsc *tls.Certificate, err error) {
	x509c, priv, err := NewCertificatePair(name, organization, validity)
	if err != nil {
		log.Fatalf("Failed to generate certificate and key pair, got error: %s", err.Error())
	}

	certOut, err := os.Create("cert.pem")
	if err != nil {
		log.Errorf("failed to open cert.pem for writing: %s", err.Error())
		return
	}
	pem.Encode(certOut, &pem.Block{Type: "CERTIFICATE", Bytes: x509c.Raw})
	certOut.Close()
	log.Print("cert.pem created\n")

	keyOut, err := os.OpenFile("key.pem", os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0600)
	if err != nil {
		log.Errorf("failed to open key.pem for writing: %s", err.Error())
		return
	}
	pem.Encode(keyOut, PemBlockForKey(priv))
	keyOut.Close()
	log.Print("key.pem created.\n")

	tlsc, err = GetTLSCertificate(x509c, priv, "hoverfly.proxy", validity)
	if err != nil {
		log.Errorf("failed to get tls certificate: %s", err.Error())
	}
	return
}

func spHandlerOpen(portname string, baud int, buftype string, isSecondary bool) {

	log.Print("Inside spHandler")

	var out bytes.Buffer

	out.WriteString("Opening serial port ")
	out.WriteString(portname)
	out.WriteString(" at ")
	out.WriteString(strconv.Itoa(baud))
	out.WriteString(" baud")
	log.Print(out.String())

	//h.broadcast <- []byte("Opened a serial port ")
	//h.broadcastSys <- out.Bytes()

	isPrimary := true
	if isSecondary {
		isPrimary = false
	}

	conf := &SerialConfig{Name: portname, Baud: baud, RtsOn: true}

	mode := &serial.Mode{
		BaudRate: baud,
		Vmin:     0,
		Vtimeout: 1,
	}

	sp, err := serial.OpenPort(portname, mode)
	log.Print("Just tried to open port")
	if err != nil {
		//log.Fatal(err)
		log.Print("Error opening port " + err.Error())
		//h.broadcastSys <- []byte("Error opening port. " + err.Error())
		h.broadcastSys <- []byte("{\"Cmd\":\"OpenFail\",\"Desc\":\"Error opening port. " + err.Error() + "\",\"Port\":\"" + conf.Name + "\",\"Baud\":" + strconv.Itoa(conf.Baud) + "}")

		return
	}
	log.Print("Opened port successfully")
	//p := &serport{send: make(chan []byte, 256), portConf: conf, portIo: sp}
	// we can go up to 256,000 lines of gcode in the buffer
	p := &serport{sendBuffered: make(chan Cmd, 256000), sendNoBuf: make(chan Cmd), portConf: conf, portIo: sp, BufferType: buftype, IsPrimary: isPrimary, IsSecondary: isSecondary}

	bw := &BufferflowDefault{}
	bw.Init()
	bw.Port = portname
	p.bufferwatcher = bw

	sh.register <- p
	defer func() { sh.unregister <- p }()
	// this is internally buffered thread to not send to serial port if blocked
	go p.writerBuffered()
	// this is thread to send to serial port regardless of block
	go p.writerNoBuf()
	p.reader()
	//go p.reader()
	//p.done = make(chan bool)
	//<-p.done
}

func (sh *serialhub) run() {

	log.Print("Inside run of serialhub")
	//cmdIdCtr := 0

	//s := ser.open()
	//ser.s := s
	//ser.write(s, []byte("hello serial data"))
	for {
		select {
		case p := <-sh.register:
			log.Print("Registering a port: ", p.portConf.Name)
			isPrimary := "false"
			if p.IsPrimary {
				isPrimary = "true"
			}
			h.broadcastSys <- []byte("{\"Cmd\":\"Open\",\"Desc\":\"Got register/open on port.\",\"Port\":\"" + p.portConf.Name + "\",\"IsPrimary\":" + isPrimary + ",\"Baud\":" + strconv.Itoa(p.portConf.Baud) + ",\"BufferType\":\"" + p.BufferType + "\"}")
			//log.Print(p.portConf.Name)
			sh.ports[p] = true
		case p := <-sh.unregister:
			log.Print("Unregistering a port: ", p.portConf.Name)
			h.broadcastSys <- []byte("{\"Cmd\":\"Close\",\"Desc\":\"Got unregister/close on port.\",\"Port\":\"" + p.portConf.Name + "\",\"Baud\":" + strconv.Itoa(p.portConf.Baud) + "}")
			delete(sh.ports, p)
			close(p.sendBuffered)
			close(p.sendNoBuf)
		case wrj := <-sh.writeJson:
			// if the user sent in the commands as json
			writeJson(wrj)

		case wr := <-sh.write:
			// if user sent in the commands as one text mode line
			write(wr, "")
		}
	}
}

func init() {
	zookeeperHost := os.Getenv("ZOOKEEPER_TEST")
	if zookeeperHost == "" {
		log.Print("Please set up ZOOKEEPER_TEST variable")
		panic("No kafka host specify")
	}
	kafkaHost := os.Getenv("KAFKA_TEST")
	if kafkaHost == "" {
		log.Print("Please set up KAFKA_TEST variable")
		panic("No kafka host specify")
	}
	brokerList = append(brokerList, kafkaHost)
	topicsInit2 = append(topicsInit2, "test2")
	producer.NewProducer(brokerList, topicsInit2, config)
	configConsumer = consumergroup.NewConfig()
	consumerBrokerList = append(consumerBrokerList, zookeeperHost)
	consumer, _ = RegisterConsumer(&Consumer{
		Topic:        "test2",
		Group:        "test2",
		BrokerList:   consumerBrokerList,
		Config:       configConsumer,
		GetModel:     testModel,
		GetEventType: testEventName,
	})
	consumer.RegisterEvent("testEvent", &ProcessEvent{testProcess})
}

// HandleError parses an AWS error and exits the program
func HandleError(err error) {
	if err == nil {
		return
	}

	if awsErr, ok := err.(awserr.Error); ok {
		log.Print("Code: " + awsErr.Code())
		log.Print("Message: " + awsErr.Message())

		if awsErr.OrigErr() != nil {
			log.Printf("Orginal Error: %v", awsErr.OrigErr())
		}

		if reqErr, ok := err.(awserr.RequestFailure); ok {
			log.Printf("Status Code: %d", reqErr.StatusCode())

			if reqErr.RequestID() != "" {
				log.Print("Request ID: " + reqErr.RequestID())
			}
		}
	} else {
		log.Print(err.Error())
	}
	os.Exit(1)
}

func newKubeClient() (*kclient.Client, error) {
	var (
		config    *restclient.Config
		err       error
		masterURL string
	)

	*argKubecfgFile = os.Getenv("KUBE_CFG_FILE")

	logrus.Print("kubeconfig: ", argKubecfgFile)

	if *argKubeMasterURL != "" {
		masterURL, err = expandKubeMasterURL()

		if err != nil {
			return nil, err
		}
	}

	if masterURL != "" && *argKubecfgFile == "" {
		config = &restclient.Config{
			Host: masterURL,
		}
	} else {
		overrides := &kclientcmd.ConfigOverrides{}
		overrides.ClusterInfo.Server = masterURL
		rules := &kclientcmd.ClientConfigLoadingRules{ExplicitPath: *argKubecfgFile}
		if config, err = kclientcmd.NewNonInteractiveDeferredLoadingClientConfig(rules, overrides).ClientConfig(); err != nil {
			return nil, err
		}
	}

	logrus.Print("Using ", config.Host, " for kubernetes master")
	return kclient.New(config)
}

// listContainers handles and reply to http requests having the path "/containers"
func listContainers(w http.ResponseWriter, r *http.Request) {

	// answer right away to avoid dead locks in LUA
	io.WriteString(w, "OK")

	go func() {
		pods, err := client.Pods("default").List(kapi.ListOptions{})

		logrus.Print("made pods request")

		if err != nil {
			logrus.Println(err.Error())
			return
		}

		//images, err := DOCKER_CLIENT.ListImages(true)

		// if err != nil {
		// 	logrus.Println(err.Error())
		// 	return
		// }

		for i := 0; i < len(pods.Items); i++ {

			logrus.Print("got pod:", pods.Items[i].ObjectMeta.Name)

			id := pods.Items[i].ObjectMeta.Name
			//info := "" //, _ := DOCKER_CLIENT.InspectContainer(id)
			name := pods.Items[i].ObjectMeta.Name
			imageRepo := ""
			imageTag := ""

			// for _, image := range images {
			// 	if image.Id == info.Image {
			// 		if len(image.RepoTags) > 0 {
			// 			imageRepo, imageTag = splitRepoAndTag(image.RepoTags[0])
			// 		}
			// 		break
			// 	}
			// }

			data := url.Values{
				"action":    {"containerInfos"},
				"id":        {id},
				"name":      {name},
				"imageRepo": {imageRepo},
				"imageTag":  {imageTag},
				"running":   {"true"},
			}

			CuberiteServerRequest(data)

			// if info.State.Running {
			// 	// Monitor stats
			// 	DOCKER_CLIENT.StartMonitorStats(id, statCallback, nil)
			// }
		}
	}()
}

func (c *Client) processReceive() {
	for {
		var data = make([]byte, 1024)
		size, err := c.conn.Read(data)
		if err != nil {
			log.Print(err.Error())
			return
		}
		if len(data) == 0 {
			log.Print("An error has occured")
			return
		}
		switch data[0] {
		case Input:
			if !c.api.PermitWrite {
				break
			}
			_, err := c.pty.Write(data[1:size])
			if err != nil {
				return
			}
		case Ping:
			_, err := c.conn.Write([]byte{Pong})
			if err != nil {
				log.Print(err.Error())
				return
			}
		case ResizeTerminal:
			var args argResizeTerminal
			err = json.Unmarshal(data[1:size], &args)
			if err != nil {
				log.Print("Malformed remote command %s %s", err, data[:size])
				return
			}
			window := struct {
				row uint16
				col uint16
				x   uint16
				y   uint16
			}{
				uint16(args.Rows),
				uint16(args.Columns),
				0,
				0,
			}
			syscall.Syscall(
				syscall.SYS_IOCTL,
				c.pty.Fd(),
				syscall.TIOCSWINSZ,
				uintptr(unsafe.Pointer(&window)),
			)
		default:
			log.Print("Unknown message type")
			return
		}
	}
}

func run(cmd *cobra.Command, args []string) {
	log.Print("Reading config.toml file")
	err := viper.ReadInConfig()
	if err != nil {
		log.Fatal("Error reading config file: ", err)
	}

	var config config.Config
	err = viper.Unmarshal(&config)

	err = config.Validate()
	if err != nil {
		log.Fatal(err.Error())
		return
	}

	if migrateFlag {
		migrate(config)
		return
	}

	app, err = gateway.NewApp(config)

	if err != nil {
		log.Fatal(err.Error())
		return
	}

	app.Serve()
}

// ProcessKafka consume th einformation from a topic kafka
func (objectConsumer *Consumer) ProcessKafka() {
	c := make(chan os.Signal, 1)
	signal.Notify(c, os.Interrupt)
	go func() {
		<-c
		if err := objectConsumer.Consumer.Close(); err != nil {
			sarama.Logger.Println("Error closing the consumer", err)
		}
	}()
	go func() {
		for err := range objectConsumer.Consumer.Errors() {
			log.Fatal(err)
		}
	}()
	for message := range objectConsumer.Consumer.Messages() {
		log.Print("Processing message with topic: ", message.Topic, " with partition: ", message.Partition, " and offset: ", message.Offset)
		model := objectConsumer.GetModel(message.Value)
		processEvent, err := objectConsumer.eventMap().Process(objectConsumer.GetEventType(model))
		if err != nil {
			log.Warn("unknow event found")
			continue
		}
		processEvent.ProcessData(model)
		time.Sleep(20 * time.Millisecond)
		objectConsumer.Consumer.CommitUpto(message)
	}
}

func (i *Filter) capturePackets() {

	tchan := make(chan TimedRawPacket, 0)
	// XXX does this need a shutdown code path?
	go func() {
		for {
			rawPacket, captureInfo, err := i.packetDataSource.ReadPacketData()
			if err == io.EOF {
				log.Print("ReadPacketData got EOF\n")
				i.Stop()
				close(tchan)
				i.supervisor.Stopped()
				return
			}
			if err != nil {
				continue
			}
			tchan <- TimedRawPacket{
				Timestamp: captureInfo.Timestamp,
				RawPacket: rawPacket,
			}
		}
	}()

	for {
		select {
		case <-i.stopCaptureChan:
			return
		case t := <-tchan:
			i.decodePacketChan <- t
		}
	}
}

func showUsage() {
	var usage string
	usage = `
Usage:
  main

Application Options:
    -r, --repo=         (required) target repository url - example format: https://github.com/nerdalert/plugin-watch.git
    -t, --time=         (requiredl) time in seconds between Git repository update checks.
    -c, --config-path=  (recommended: default: [ ./tmp/conf/ ]) path to config files.
    -b, --backup-path=  (recommended: default: [ data/snapshots ]) path to the backup endpoint config files.
    -s, --server=       (optional: default: [ data/endpoints ]) path to config files.
    -d, --daemon=       (optional:default [ true ]) run as a daemon. Alternatively could be run via a cron job.
    -l, --loglevel=     (optional:default [ info ]) set the logging level. Default is 'info'. options are [debug, info, warn, error].
    -h, --help    show app help.

Example daemon mode processing flows every 2 minutes:
	git-control -r github.com/plugin-watch -t 120 -l debug -r https://github.com/nerdalert/plugin-watch.git

Example run-once export:
    TODO:

Help Options:
  -h, --help    Show this help message
  `
	log.Print(usage)
}

func ws() http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		c, err := upgrader.Upgrade(w, r, nil)

		if err != nil {
			log.Print("upgrade:", err)
			return
		}
		defer c.Close()

		for {
			// Write
			err := c.WriteMessage(websocket.TextMessage, []byte("Hello, Client!"))
			if err != nil {
				log.Fatal(err)
			}

			// Read
			_, msg, err := c.ReadMessage()
			if err != nil {
				log.Fatal(err)
			}
			fmt.Printf("%s\n", msg)
		}
	}
}

// DiscoverDirectors
func (s *State) DiscoverDirectors(localId string, events chan *DirectorEvent) {
	log.Print("Discovering other directors")

	updates := make(chan *etcd.Response)

	go func() {
		if _, err := s.etc.Watch(etcDirectorPrefix, 0, true, updates, nil); err != nil {
			log.WithError(err).Fatal("Failed to watch etc director prefix")
		}
	}()

	for res := range updates {
		node := idFromPath(res.Node.Key)

		if node == localId {
			// ignore events about ourself
			continue
		}

		events <- &DirectorEvent{
			Action: res.Action,
			Node:   node,
		}
	}
}

func (list PortMapList) Add(portmap *PortMap) {
	if portmap != nil {
		list[portmap.Port] = portmap
	} else {
		log.Print("It is not possible to add an unspecified portmap(nil).")
	}
}

func LineByLinePipe(cb func(string)) io.Writer {
	r, w, err := os.Pipe()
	if err != nil {
		panic(err)
	}
	pipeId := RandomNumericString(5)
	go func() {
		log.Printf("[pipe:%s] Pipe created", pipeId)
		reader := bufio.NewReader(r)
		msg := ""
		for {
			bytes, ispr, err := reader.ReadLine()

			if err != nil && err != io.EOF {
				panic(err)
			}

			msg += string(bytes)
			log.Print()
			if ispr {
				continue
			}
			line := msg
			msg = ""
			cb(line)

			if err == io.EOF {
				log.Printf("[pipe:%s] Pipe closed", pipeId)
				break
			}
		}
	}()
	return w
}

// BuildNew checks what packages to build based on related repo and builds
// those that have been updated.
func (b *Builder) BuildNew(pkgs []string, aur *AUR) ([]*BuiltPkg, error) {
	// initialize pacman.conf with upstream repo
	err := b.setup()
	if err != nil {
		return nil, err
	}

	// make sure environment is up to date
	err = b.update()
	if err != nil {
		return nil, err
	}

	// get packages that should be built
	srcPkgs, err := b.getBuildPkgs(pkgs, aur)
	if err != nil {
		return nil, err
	}

	if len(srcPkgs) == 0 {
		log.Print("All packages up to date, nothing to build")
		return nil, nil
	}

	buildPkgs, err := b.buildPkgs(srcPkgs)
	if err != nil {
		return nil, err
	}

	successLog(buildPkgs)
	return buildPkgs, nil
}

func main() {
	var err error

	serverAddr, err := net.ResolveUDPAddr("udp", "127.0.0.1:58672")
	if err != nil {
		logrus.Panicf("Error resolving UDP address with %v", err)
	}

	localAddr, err := net.ResolveUDPAddr("udp", "127.0.0.1:0")
	if err != nil {
		logrus.Panicf("Error resolving UDP address with %v", err)
	}

	UDPConn, err := net.DialUDP("udp", localAddr, serverAddr)
	if err != nil {
		logrus.Panicf("Error with dial UDP: %v", err)
	}

	defer UDPConn.Close()
	readBuf := make([]byte, UDPBufSize)
	for {
		read, err := UDPConn.Read(readBuf)
		if err != nil {
			logrus.Print("Read error: %v", err)
			continue
		}

		logrus.Printf("Readed bytes %d with message: %s", read, string(readBuf))
	}
}

func wait() {
	term := make(chan os.Signal)
	signal.Notify(term, os.Interrupt, syscall.SIGTERM)
	select {
	case <-term:
		log.Print("Received SIGTERM, exiting ...")
	}
}