package source

import (
	"github.com/go-redis/redis"
	streams "gitee.com/qw_1215/glink"
	"gitee.com/qw_1215/glink/flow"
)

// RedisSource implements Redis Pub/Sub Source
type RedisSource struct {
	redisdb *redis.Client
	channel string
	out     chan interface{}
}

// NewRedisSource returns new RedisSource instance
func NewRedisSource(config *redis.Options, channel string) (*RedisSource, error) {
	redisdb := redis.NewClient(config)
	pubsub := redisdb.Subscribe(channel)

	// Wait for confirmation that subscription is created before publishing anything
	if _, err := pubsub.Receive(); err != nil {
		return nil, err
	}
	source := &RedisSource{
		redisdb,
		channel,
		make(chan interface{}),
	}
	go source.init(pubsub.Channel())
	return source, nil
}

func (rs *RedisSource) init(ch <-chan *redis.Message) {
	for msg := range ch {
		rs.out <- msg
	}
}

// Via streams data through given flow
func (rs *RedisSource) Via(_flow streams.Flow) streams.Flow {
	flow.DoStream(rs, _flow)
	return _flow
}

// Out returns channel for sending data
func (rs *RedisSource) Out() <-chan interface{} {
	return rs.out
}