package zookeeper

import (
	"encoding/json"
	"errors"
	"fmt"
	"gitee.com/h79/goutils/common/logger"
	"gitee.com/h79/goutils/common/system"
	"gitee.com/h79/goutils/discovery/config"
	"gitee.com/h79/goutils/discovery/registry"
	"gitee.com/h79/goutils/discovery/service"
	"github.com/go-zookeeper/zk"
	"sync"
	"time"
)

var _ service.Service = (*zkService)(nil)

type zkService struct {
	Base
	id     string
	conf   config.Config
	locker sync.Mutex
	stop   chan error
}

// NewService
// 注册一个服务
func NewService(cfg config.Config, points config.EndPoints, registry registry.Registry) (service.Service, error) {

	cli, err := NewClientZk(&points, cfg.Node, 30)
	if err != nil {
		return nil, err
	}
	logger.Info("Zookeeper.service: create client")

	s, err := NewServiceWith(cli, cfg)
	if err != nil {
		cli.Close()
	}
	return s, err
}

func NewServiceWith(cli *zk.Conn, cfg config.Config) (service.Service, error) {

	id := fmt.Sprintf("service: %s", cfg.Server.To())

	path := cfg.Node.NameWith("")

	exists, _, err := cli.Exists(path)
	if err != nil {
		return nil, err
	}

	if !exists { //不存在，注册一个服务
		_, err := cli.Create(path, []byte(""), 0, zk.WorldACL(zk.PermAll))
		if err != nil && err != zk.ErrNodeExists {

			return nil, err
		}
	}

	data, err := json.Marshal(&cfg.Server)
	if err != nil {
		return nil, err
	}

	path += "/n"
	_, err = cli.CreateProtectedEphemeralSequential(path, data, zk.WorldACL(zk.PermAll))
	if err != nil {
		return nil, err
	}

	ser := &zkService{Base: Base{
		conn: cli,
	}, id: id, conf: cfg, stop: make(chan error)}

	return ser, nil
}

// Start
/**
* Service interface
* 使用 goroutine  go Start
 */
func (s *zkService) Start() error {
	system.ChildRunning(s.keepAlive)
	system.ChildRunning(s.close)
	return nil
}

// Stop interface
func (s *zkService) Stop() {
	s.stop <- errors.New("zk.service: Stop")
}

func (s *zkService) close() {
	sys := system.Exit()
	for {
		select {
		case err := <-s.stop:
			s.revoke()
			logger.Error("Zookeeper.service: close, err= %v", err)
			return

		case <-sys.Done():
			s.stop <- nil
			s.revoke()
			return

		case <-system.Closed():
			s.stop <- nil
			s.revoke()
			return
		}
	}
}

func (s *zkService) keepAlive() {
	if !s.conf.Check.Ttl {
		return
	}
	ticker := time.NewTicker(time.Second * 8)
	for {
		select {
		case err := <-s.stop:
			logger.Error("Zookeeper.service: keepAlive stop, err= %v", err)
			return

		case <-ticker.C:
			var path = s.conf.Node.NameWith("")
			if _, err := s.conn.CreateTTL(path, nil, zk.FlagTTL, zk.WorldACL(zk.PermAll), 10000); err != nil {
				logger.Error("Zookeeper.service: update ttl err= %v", err)
			}

		case <-system.Closed():
			return
		}
	}
}

func (s *zkService) revoke() {
	s.conn.Close()
}