//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
}

//Simple pusher for testing
func Pusher(count int, finished chan int, port int) {
	socket, err := nano.NewPushSocket()
	if nil != err {
		log.Error(err)
	}
	defer socket.Close()
	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
	bytes := []byte{'h', 'e', 'l', 'l', '\xc3', '\xb8', 'x', 'f', 'c', 'x', 'f'}
	for {
		_, err := socket.Send(bytes, 0) //blocking
		if nil != err {
			log.Error(err)
		} else {
			tot++
		}
		if tot >= count {
			break
		}
	}
	finished <- tot
}

func (s *ESAPIV0) Bulk(data *bytes.Buffer) {
	if data == nil || data.Len() == 0 {
		return
	}
	data.WriteRune('\n')
	url := fmt.Sprintf("%s/_bulk", s.Host)

	client := &http.Client{}
	reqest, _ := http.NewRequest("POST", url, data)
	if s.Auth != nil {
		reqest.SetBasicAuth(s.Auth.User, s.Auth.Pass)
	}
	resp, errs := client.Do(reqest)
	if errs != nil {
		log.Error(errs)
		return
	}

	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		log.Error(err)
		return
	}
	log.Trace(url, string(body))
	defer resp.Body.Close()
	defer data.Reset()
	if resp.StatusCode != 200 {
		log.Errorf("bad bulk response: %s %s", body, resp.StatusCode)
		return
	}
}

func setInitLogging(logLevel string) {

	logLevel = strings.ToLower(logLevel)

	testConfig := `
	<seelog  type="sync" minlevel="`
	testConfig = testConfig + logLevel
	testConfig = testConfig + `">
		<outputs formatid="main">
			<filter levels="error">
				<file path="./log/gopa.log"/>
			</filter>
			<console formatid="main" />
		</outputs>
		<formats>
			<format id="main" format="[%Date(01-02) %Time] [%LEV] [%File:%Line,%FuncShort] %Msg%n"/>
		</formats>
	</seelog>`

	logger, err := log.LoggerFromConfigAsString(testConfig)
	if err != nil {
		log.Error("init config error,", err)
	}
	err = log.ReplaceLogger(logger)
	if err != nil {
		log.Error("init config error,", err)
	}
}

func (this *RESTTransceiver) routine() {
	errors := 0
	onHTTPError := func(err error) {
		log.Error(err)
		errors += 1
		time.Sleep(time.Duration(errors) * time.Second)
	}
	onActionError := func(err error) {
		log.Error(err)
	}
	for {
		action, err := getAction(this.Client, this.OrchestratorURL, this.EntityID)
		if err != nil {
			onHTTPError(err)
			continue
		}
		err = deleteAction(this.Client, this.OrchestratorURL, action)
		if err != nil {
			onHTTPError(err)
			continue
		}
		err = this.onAction(action)
		if err != nil {
			onActionError(err)
			continue
		}
		errors = 0
	}
}

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()

}

func initLogger() {
	var err error

	// 初始化raven
	if cfg.Sentry != "" {
		Raven, err = raven.NewClient(cfg.Sentry, nil)
		if err != nil {
			log.Error("Init Sentry Error:", err)
		}
	}

	// 初始化logger
	if cfg.Logger != "" {
		// 自定义一个seelog raven receiver
		receiver := &RavenReciver{Client: Raven}
		parseParams := &log.CfgParseParams{
			CustomReceiverProducers: map[string]log.CustomReceiverProducer{
				"sentry": func(log.CustomReceiverInitArgs) (log.CustomReceiver, error) {
					return receiver, nil
				},
			},
		}
		if logger, err := log.LoggerFromParamConfigAsFile(cfg.Logger, parseParams); err == nil {
			log.ReplaceLogger(logger)
		} else {
			log.Error("Parse Logger Error: ", err)
		}

	}

}

func (m *Migrator) NewFileReadWorker(pb *pb.ProgressBar, wg *sync.WaitGroup) {
	log.Debug("start reading file")
	f, err := os.Open(m.Config.DumpInputFile)
	if err != nil {
		log.Error(err)
		return
	}

	defer f.Close()
	r := bufio.NewReader(f)
	lineCount := 0
	for {
		line, err := r.ReadString('\n')
		if io.EOF == err || nil != err {
			break
		}
		lineCount += 1
		js := map[string]interface{}{}

		//log.Trace("reading file,",lineCount,",", line)
		err = json.Unmarshal([]byte(line), &js)
		if err != nil {
			log.Error(err)
			continue
		}
		m.DocChan <- js
		pb.Increment()
	}

	defer f.Close()
	log.Debug("end reading file")
	close(m.DocChan)
	wg.Done()
}

func GetConfig() (currConfig config, err error) {

	//Open the config file. Defer to closing it when the function goes out of scope
	configFile, err := os.Open("config/env.json")
	defer configFile.Close()

	//If we had an error log it and return our error
	if err != nil {
		log.Error("Problem grabbing configuration", err)
		return
	}

	// Create a json parser for this file
	jsonParser := json.NewDecoder(configFile)

	//This is a good golang pattern to use:
	// grab the error from the Decode call and if the err is not null log
	if err = jsonParser.Decode(&currConfig); err != nil {
		log.Error("Problem parsing configuration", err)
		return
	}

	return currConfig, err

}

func processFile(req uploadRequest, db *database.DB, store *storage.Store) {
	defer req.file.Close()

	epub, err := openMultipartEpub(req.file)
	if err != nil {
		log.Warn("Not valid epub uploaded file ", req.filename, ": ", err)
		return
	}
	defer epub.Close()

	book, id := parseFile(epub, store)
	req.file.Seek(0, 0)
	size, err := store.Store(id, req.file, EPUB_FILE)
	if err != nil {
		log.Error("Error storing book (", id, "): ", err)
		return
	}

	book["filesize"] = size
	err = db.AddBook(book)
	if err != nil {
		log.Error("Error storing metadata (", id, "): ", err)
		return
	}
	log.Info("File uploaded: ", req.filename)
}

func (s *ESAPIV0) NextScroll(scrollTime string, scrollId string) (*Scroll, error) {
	//  curl -XGET 'http://es-0.9:9200/_search/scroll?scroll=5m'
	id := bytes.NewBufferString(scrollId)
	url := fmt.Sprintf("%s/_search/scroll?scroll=%s&scroll_id=%s", s.Host, scrollTime, id)
	resp, body, errs := Get(url, s.Auth, s.HttpProxy)
	if errs != nil {
		log.Error(errs)
		return nil, errs[0]
	}

	if resp.StatusCode != 200 {
		return nil, errors.New(body)
	}

	defer resp.Body.Close()

	log.Trace("next scroll,", url, body)

	// decode elasticsearch scroll response
	scroll := &Scroll{}
	err := json.Unmarshal([]byte(body), &scroll)
	if err != nil {
		log.Error(body)
		log.Error(err)
		return nil, err
	}

	return scroll, nil
}

func (emailer *EmailNotifier) sendConsumerGroupStatusNotify() error {
	var bytesToSend bytes.Buffer
	log.Debug("send email")

	msgs := make([]Message, len(emailer.Groups))
	i := 0
	for group, msg := range emailer.groupMsgs {
		msgs[i] = msg
		delete(emailer.groupMsgs, group)
		i++
	}

	err := emailer.template.Execute(&bytesToSend, struct {
		From    string
		To      string
		Results []Message
	}{
		From:    emailer.From,
		To:      emailer.To,
		Results: msgs,
	})
	if err != nil {
		log.Error("Failed to assemble email:", err)
		return err
	}

	err = smtp.SendMail(fmt.Sprintf("%s:%v", emailer.Server, emailer.Port),
		emailer.auth, emailer.From, []string{emailer.To}, bytesToSend.Bytes())
	if err != nil {
		log.Error("Failed to send email message:", err)
		return err
	}
	return nil
}

func (fs *GDriveFileSystem) Get(p string) (webdav.StatusCode, io.ReadCloser, int64) {
	pFile := fs.getFile(p, false)
	if pFile == nil {
		return webdav.StatusCode(404), nil, -1
	}

	f := pFile.file
	downloadUrl := f.DownloadUrl
	log.Debug("downloadUrl=", downloadUrl)
	if downloadUrl == "" {
		log.Error("No download url: ", f)
		return webdav.StatusCode(500), nil, -1
	}

	req, err := http.NewRequest("GET", downloadUrl, nil)
	if err != nil {
		log.Error("NewRequest ", err)
		return webdav.StatusCode(500), nil, -1
	}

	resp, err := fs.transport.RoundTrip(req)
	if err != nil {
		log.Error("RoundTrip ", err)
		return webdav.StatusCode(500), nil, -1
	}

	return webdav.StatusCode(200), resp.Body, f.FileSize
}

func dial() *net.UDPConn {
	err, host := config.GetStringMapString("udp", "host")
	if err != nil {
		log.Error("can't find udp host")
		os.Exit(0)
	}
	err, port := config.GetStringMapString("udp", "port")
	if err != nil {
		log.Error("can't find udp port")
		os.Exit(0)
	}
	raddr, err := net.ResolveUDPAddr("udp", host+":"+port)
	if err != nil {
		log.Error("connect udp %s:%s error\n", host, port)
		os.Exit(0)
	}
	conn, err := net.DialUDP("udp", nil, raddr)
	if err != nil {
		log.Error("connect udp %s:%s error\n", host, port)
		os.Exit(0)
	}
	// bump up the packet size for large log lines
	err = conn.SetWriteBuffer(writeBuffer)
	if err != nil {
		log.Error("connect udp %s:%s error\n", host, port)
		os.Exit(0)
	}
	return conn
}

func (h *handler) handleCopy(r *http.Request) StatusCode {
	var err error

	p := url2path(r.URL)

	strDepth := r.Header.Get("Depth")
	depth := 9999999
	if strDepth != "" && strDepth != "infinity" {
		depth, err = strconv.Atoi(strDepth)
		if err != nil {
			log.Error("Can't parse depth ", err)
			return StatusCode(500)
		}
	}

	dest, err := urlstring2path(r.Header.Get("Destination"))
	if err != nil {
		log.Error("Can't parse dest ", err)
		return StatusCode(500)
	}

	overwrite := true

	if r.Header.Get("Overwrite") == "F" {
		overwrite = false
	}

	log.Debug("Copy from ", p, " to ", dest, " depth=", depth, " overwrite=", overwrite)

	return StatusCode(h.fs.Copy(p, dest, depth, overwrite))
}

func InitTokens() (err error) {
	signBytes, err := ioutil.ReadFile("config/rsaKey")
	if err != nil {
		log.Error("Error reading private key from file: ", err)
		return
	}

	signKey, err = jwt.ParseRSAPrivateKeyFromPEM(signBytes)
	if err != nil {
		log.Error("Error parsing private key from file: ", err)
		return
	}

	verifyBytes, err := ioutil.ReadFile("config/pubKey")
	if err != nil {
		log.Error("Error reading public key from file: ", err)
		return
	}

	verifyKey, err = jwt.ParseRSAPublicKeyFromPEM(verifyBytes)
	if err != nil {
		log.Error("Error parsing public key from file: ", err)
		return
	}
	return
}

func (s *ESAPIV5) NewScroll(indexNames string, scrollTime string, docBufferCount int, query string, slicedId, maxSlicedCount int) (scroll *Scroll, err error) {
	url := fmt.Sprintf("%s/%s/_search?scroll=%s&size=%d", s.Host, indexNames, scrollTime, docBufferCount)

	jsonBody := ""
	if len(query) > 0 || maxSlicedCount > 0 {
		queryBody := map[string]interface{}{}

		if len(query) > 0 {
			queryBody["query"] = map[string]interface{}{}
			queryBody["query"].(map[string]interface{})["query_string"] = map[string]interface{}{}
			queryBody["query"].(map[string]interface{})["query_string"].(map[string]interface{})["query"] = query
		}

		if maxSlicedCount > 1 {
			log.Tracef("sliced scroll, %d of %d", slicedId, maxSlicedCount)
			queryBody["slice"] = map[string]interface{}{}
			queryBody["slice"].(map[string]interface{})["id"] = slicedId
			queryBody["slice"].(map[string]interface{})["max"] = maxSlicedCount
		}

		jsonArray, err := json.Marshal(queryBody)
		if err != nil {
			log.Error(err)

		} else {
			jsonBody = string(jsonArray)
		}
	}

	resp, body, errs := Post(url, s.Auth, jsonBody, s.HttpProxy)

	if errs != nil {
		log.Error(errs)
		return nil, errs[0]
	}
	defer resp.Body.Close()

	if resp.StatusCode != 200 {
		return nil, errors.New(body)
	}

	log.Trace("new scroll,", body)

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

	scroll = &Scroll{}
	err = json.Unmarshal([]byte(body), scroll)
	if err != nil {
		log.Error(err)
		return nil, err
	}

	return scroll, err
}

func errHndlr(err error, severity int) {
	if err != nil {
		switch {
		case severity == ERROR:
			log.Error(err)
		case severity == FATAL:
			log.Error(err)
			panic(err)
		}
	}
}

func writeSummaryStats(config StartupConfig, statsSummary traffic_ops.StatsSummary) {
	to, err := traffic_ops.Login(config.ToURL, config.ToUser, config.ToPasswd, true)
	if err != nil {
		newErr := fmt.Errorf("Could not store summary stats! Error logging in to %v: %v", config.ToURL, err)
		log.Error(newErr)
		return
	}
	err = to.AddSummaryStats(statsSummary)
	if err != nil {
		log.Error(err)
	}
}

func handleCredentials(apiVersion, subpath string, c *ContainerService, w http.ResponseWriter, r *http.Request) {
	resp, err := instanceServiceClient.RoundTrip(NewGET(baseUrl + "/" + apiVersion + "/meta-data/iam/security-credentials/"))

	if err != nil {
		log.Error("Error requesting creds path for API version ", apiVersion, ": ", err)
		w.WriteHeader(http.StatusInternalServerError)
		return
	}

	resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		w.WriteHeader(resp.StatusCode)
		return
	}

	clientIP := remoteIP(r.RemoteAddr)
	role, err := c.RoleForIP(clientIP)

	if err != nil {
		log.Error(clientIP, " ", err)
		http.Error(w, "An unexpected error getting container role", http.StatusInternalServerError)
		return
	}

	roleName := role.Arn.RoleName()

	if len(subpath) == 0 {
		w.Write([]byte(roleName))
	} else if !strings.HasPrefix(subpath, roleName) || (len(subpath) > len(roleName) && subpath[len(roleName)-1] != '/') {
		// An idiosyncrasy of the standard EC2 metadata service:
		// Subpaths of the role name are ignored. So long as the correct role name is provided,
		// it can be followed by a slash and anything after the slash is ignored.
		w.WriteHeader(http.StatusNotFound)
	} else {
		creds, err := json.Marshal(&MetadataCredentials{
			Code:            "Success",
			LastUpdated:     role.LastUpdated,
			Type:            "AWS-HMAC",
			AccessKeyId:     role.Credentials.AccessKey,
			SecretAccessKey: role.Credentials.SecretKey,
			Token:           role.Credentials.Token,
			Expiration:      role.Credentials.Expiration,
		})

		if err != nil {
			log.Error("Error marshaling credentials: ", err)
			w.WriteHeader(http.StatusInternalServerError)
		} else {
			w.Write(creds)
		}
	}
}