package xcdnload

import (
	"crypto/tls"
	"crypto/x509"
	"fmt"
	"io/ioutil"
	"log"
	"os"
	"os/signal"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/Shopify/sarama"
	"github.com/spf13/cobra"
	"google.golang.org/protobuf/proto"
)

var (
	server string
	topic  string

	// thread hold
	t1            int64
	t2            int64
	cycle         int64
	provinceCycle int64

	urlIdPrefix string

	// NAME_1 ssl configs
	enableTLS  bool
	clientCert string
	clientKey  string
	caCert     string

	CmdKafka = &cobra.Command{
		Use:   "xcdnload",
		NAME_2: "xcdn NAME_1 xcdnload",
		Long:  "xcdn NAME_1 xcdnload",
		RunE:  process,
	}
)

// init parameters
func init() {
	CmdKafka.Flags().StringVarP(&server, "server", "", "yq01-cdn-jlog-m4225-19.yq01.baidu.com:8091,yq01-cdn-jlog-m4225-20.yq01.baidu.com:8091,yq01-cdn-jlog-m4225-21.yq01.baidu.com:8091,yq01-cdn-jlog-m4225-22.yq01.baidu.com:8091,yq01-cdn-jlog-m4225-23.yq01.baidu.com:8091", "NAME_1 server format host:port")
	CmdKafka.Flags().StringVarP(&topic, "topic", "", "xcdn_test_01", "NAME_1 topic")
	CmdKafka.Flags().Int64VarP(&cycle, "cycle", "", 100, "cycleCntTh default 100")
	CmdKafka.Flags().Int64VarP(&provinceCycle, "provinceCycle", "", 100000, "provinceCycle default 100000")
	CmdKafka.Flags().Int64VarP(&t1, "t1", "", 10, "t1 default every 10 sleep 1000 ms")
	CmdKafka.Flags().Int64VarP(&t2, "t2", "", 1000, "t2 default every 10 sleep 1000 ms")
	CmdKafka.Flags().StringVarP(&urlIdPrefix, "urlIdPrefix", "", "a", "urlIdPrefix")

	CmdKafka.Flags().BoolVarP(&enableTLS, "enableTLS", "", false, "enable TLS")
	CmdKafka.Flags().StringVarP(&clientCert, "clientCert", "", "NAME_1-key/client.pem", "client certificate")
	CmdKafka.Flags().StringVarP(&clientKey, "clientKey", "", "NAME_1-key/client.key", "client key")
	CmdKafka.Flags().StringVarP(&clientCert, "ca", "", "NAME_1-key/ca.pem", "ca certificate")
}

// process producer processing
func process(*cobra.Command, []string) error {
	config := sarama.NewConfig()
	config.Producer.Return.Successes = true
	config.Producer.Partitioner = sarama.NewRoundRobinPartitioner

	if enableTLS {
		tlsConfig, err := genTLSConfig(clientCert, clientKey, caCert)
		if err != nil {
			return fmt.Errorf("get config fail: %s", err.Error())
		}
		config.Net.TLS.Enable = true
		config.Net.TLS.Config = tlsConfig
	}
	client, err := sarama.NewClient(strings.Split(server, ","), config)
	if err != nil {
		log.Fatalf("init NAME_1 client fail: %s", err.Error())
	}
	log.Printf("init NAME_1 client success, NAME_1 version: %s", client.Config().Version)

	// produce msg
	if err := produ