package redis_consumer

import (
	"context"
	"fmt"
	"github.com/gomodule/redigo/redis"
	"github.com/pkg/errors"
	"log"
	"time"
)

type RedisConsumeHandler func(data string)

type RedisConsumerOptions struct {
	Addr     string `json:"addr"`
	Port     int    `json:"port"`
	Password string `json:"password"`
	DB       int    `json:"db"`
}

var defaultOptions = RedisConsumerOptions{
	Addr:     "127.0.0.1",
	Port:     6379,
	Password: "",
	DB:       0,
}

type RedisConsumer struct {
	ctx      context.Context
	dbPool   *redis.Pool
	handlers map[string]RedisConsumeHandler
	client   *redis.PubSubConn
	opts     *RedisConsumerOptions
}

func NewRedisConsumer(ctx context.Context, options *RedisConsumerOptions) (*RedisConsumer, error) {
	var opts = defaultOptions
	if options != nil {
		opts = *options
	}
	c, err := redis.Dial("tcp", fmt.Sprintf("%s:%d", options.Addr, options.Port))
	if err != nil {
		return nil, err
	}
	defer c.Close()
	result := &RedisConsumer{
		ctx:      ctx,
		opts:     &opts,
		handlers: make(map[string]RedisConsumeHandler),
	}
	return result, nil
}

func (rc *RedisConsumer) Subscribe(topic string, handler RedisConsumeHandler) error {
	_, ok := rc.handlers[topic]
	if ok {
		return errors.Errorf("主题[%s]已被订阅", topic)
	}
	rc.handlers[topic] = handler
	return nil
}

func (rc *RedisConsumer) processTopicHandler(topic string, data []byte) {
	fn, ok := rc.handlers[topic]
	if !ok {
		return
	}
	defer func() {
		if err := recover(); err != nil {
			log.Printf("error:%v\n", err)
		}
	}()
	fn(string(data))
}

func (rc *RedisConsumer) initSubscribe() (redis.Conn, error) {
	c, err := redis.Dial("tcp", fmt.Sprintf("%s:%d", rc.opts.Addr, rc.opts.Port), redis.DialDatabase(rc.opts.DB), redis.DialPassword(rc.opts.Password))
	if err != nil {
		return nil, err
	}
	rc.client = &redis.PubSubConn{
		Conn: c,
	}
	for topic := range rc.handlers {
		if err := rc.client.Subscribe(topic); err != nil {
			log.Printf("订阅[%s]出错", topic)
		}
	}
	return c, nil
}

func (rc *RedisConsumer) Start() error {
	con, err := rc.initSubscribe()
	if err != nil {
		return errors.Wrap(err, "启动订阅出错")
	}
	defer func() {
		if err := con.Close(); err != nil {
			fmt.Printf("close redis con error:%s\n", err)
		}
	}()
	go func() {
		select {
		case <-rc.ctx.Done():
			if err := rc.client.Unsubscribe(); err != nil {
				log.Printf("redis un subscribe error:%s\n", err)
			}
		}
	}()
	errCount := 0
restart:
	if c, err := rc.initSubscribe(); err != nil {
		log.Printf("重新初始化订阅出错：%s\n", err)
		errCount++
		if errCount > 100 {
			goto exit
		}
	} else {
		con = c
		errCount = 0
	}
	for {
		switch res := rc.client.Receive().(type) {
		case redis.Message:
			rc.processTopicHandler(res.Channel, res.Data)
			break
		case redis.Subscription:
			if res.Kind == "subscribe" {
				fmt.Printf("订阅: channel: %s, kind: %s, count: %d\n", res.Channel, res.Kind, res.Count)
			} else if res.Kind == "unsubscribe" {
				fmt.Printf("取消订阅: channel: %s, kind: %s, count: %d\n", res.Channel, res.Kind, res.Count)
				if res.Count == 0 {
					goto exit
				}
			}
			break
		case error:
			log.Printf("redis pub subscribe error:%s\n", res)
			con.Close()
			time.Sleep(2 * time.Second)
			log.Printf("重新初始化订阅\n")
			goto restart
		default:
			log.Printf("redis 消息:%v\n", res)
		}
	}
exit:
	log.Printf("订阅结束.\n")
	return nil
}