model

var ASYNC_EXPIRE int = 1000

var CONSUMERGROUP string = "default"

var GetAsyncBatchConsumer = func() (*consumergroup.ConsumerGroup, error) {
	return NewAsyncBatchConsumer(ZOOKEEPER,
		CONSUMERGROUP,
		TOPIC,
		HEAD_OFFSETS,
		RESET_OFFSETS)
}

var GetAsyncBatchProducer = func() (sarama.SyncProducer, error) {
	return NewAsyncBatchProducer(ZOOKEEPER)
}

var GetAsyncJobRedis = func() *redis.Client {
	return NewRedisClient(REDIS_HOST,
		REDIS_PORT,
		REDIS_PW,
		strconv.Itoa(REDIS_DB))
}

var GetRequestClient = func() BatchClient {
	return &http.Client{}
}

var HEAD_OFFSETS int64 = 0

var HostMap map[string]string

var NewAsyncBatchProducer = func(zookeeperConn string) (sarama.SyncProducer, error) {
	zookeeper, err := kazoo.NewKazooFromConnectionString(zookeeperConn, kazoo.NewConfig())
	if err != nil {
		log.Fatalln("An error occurred connecting to zookeeper", err)
		return nil, err
	}

	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll
	config.Producer.Retry.Max = 3
	config.Producer.Retry.Backoff = 100 * time.Millisecond

	brokerList, err := zookeeper.BrokerList()
	if err != nil {
		log.Fatalln("An error occurred getting broker list from zookeeper", err)
		return nil, err
	}

	producer, err := sarama.NewSyncProducer(brokerList, config)
	if err != nil {
		log.Fatalln("Failed to start Sarama producer:", err)
		return nil, err
	}

	return producer, nil
}

var NewRedisClient = func(host, port, password, db string) *redis.Client {
	return redis.NewTCPClient(&redis.Options{
		Addr:     fmt.Sprintf("%s:%s", REDIS_HOST, REDIS_PORT),
		Password: REDIS_PW,
		DB:       int64(REDIS_DB),
	})
}

var REDIS_DB int = 0

var REDIS_HOST string = "default"

var REDIS_PORT string = "default"

var REDIS_PW string = "default"

var RESET_OFFSETS bool = false

var TOPIC string = "default"

var WORKER_SLEEP int = 100

var ZOOKEEPER string = "default"