4 files changed, 2 insertions, 270 deletions
diff --git a/plugins/broadcast/redis/driver.go b/plugins/broadcast/redis/driver.go
new file mode 100644
index 00000000..65a229e1
--- /dev/null
+++ b/plugins/broadcast/redis/driver.go
@@ -0,0 +1 @@
+package redis
diff --git a/plugins/broadcast/redis/plugin.go b/plugins/broadcast/redis/plugin.go
new file mode 100644
index 00000000..65a229e1
--- /dev/null
+++ b/plugins/broadcast/redis/plugin.go
@@ -0,0 +1 @@
+package redis
diff --git a/plugins/broadcast/redis/redis.go b/plugins/broadcast/redis/redis.go
deleted file mode 100644
index 41f48658..00000000
--- a/plugins/broadcast/redis/redis.go
+++ /dev/null
@@ -1,172 +0,0 @@
-package redis
-
-import (
-	"context"
-	"errors"
-	"sync/atomic"
-
-	"github.com/go-redis/redis/v8"
-)
-
-// Redis based broadcast Router.
-type Redis struct {
-	client        redis.UniversalClient
-	psClient      redis.UniversalClient
-	router        *Router
-	messages      chan *Message
-	listen, leave chan subscriber
-	stop          chan interface{}
-	stopped       int32
-}
-
-// creates new redis broker
-func redisBroker(cfg *RedisConfig) (*Redis, error) {
-	client := cfg.redisClient()
-	if _, err := client.Ping(context.Background()).Result(); err != nil {
-		return nil, err
-	}
-
-	psClient := cfg.redisClient()
-	if _, err := psClient.Ping(context.Background()).Result(); err != nil {
-		return nil, err
-	}
-
-	return &Redis{
-		client:   client,
-		psClient: psClient,
-		router:   NewRouter(),
-		messages: make(chan *Message),
-		listen:   make(chan subscriber),
-		leave:    make(chan subscriber),
-		stop:     make(chan interface{}),
-		stopped:  0,
-	}, nil
-}
-
-// Serve serves broker.
-func (r *Redis) Serve() error {
-	pubsub := r.psClient.Subscribe(context.Background())
-	channel := pubsub.Channel()
-
-	for {
-		select {
-		case ctx := <-r.listen:
-			ctx.done <- r.handleJoin(ctx, pubsub)
-		case ctx := <-r.leave:
-			ctx.done <- r.handleLeave(ctx, pubsub)
-		case msg := <-channel:
-			r.router.Dispatch(&Message{
-				Topic:   msg.Channel,
-				Payload: []byte(msg.Payload),
-			})
-		case <-r.stop:
-			return nil
-		}
-	}
-}
-
-func (r *Redis) handleJoin(sub subscriber, pubsub *redis.PubSub) error {
-	if sub.pattern != "" {
-		newPatterns, err := r.router.SubscribePattern(sub.upstream, sub.pattern)
-		if err != nil || len(newPatterns) == 0 {
-			return err
-		}
-
-		return pubsub.PSubscribe(context.Background(), newPatterns...)
-	}
-
-	newTopics := r.router.Subscribe(sub.upstream, sub.topics...)
-	if len(newTopics) == 0 {
-		return nil
-	}
-
-	return pubsub.Subscribe(context.Background(), newTopics...)
-}
-
-func (r *Redis) handleLeave(sub subscriber, pubsub *redis.PubSub) error {
-	if sub.pattern != "" {
-		dropPatterns := r.router.UnsubscribePattern(sub.upstream, sub.pattern)
-		if len(dropPatterns) == 0 {
-			return nil
-		}
-
-		return pubsub.PUnsubscribe(context.Background(), dropPatterns...)
-	}
-
-	dropTopics := r.router.Unsubscribe(sub.upstream, sub.topics...)
-	if len(dropTopics) == 0 {
-		return nil
-	}
-
-	return pubsub.Unsubscribe(context.Background(), dropTopics...)
-}
-
-// Stop closes the consumption and disconnects broker.
-func (r *Redis) Stop() {
-	if atomic.CompareAndSwapInt32(&r.stopped, 0, 1) {
-		close(r.stop)
-	}
-}
-
-// Subscribe broker to one or multiple channels.
-func (r *Redis) Subscribe(upstream chan *Message, topics ...string) error {
-	if atomic.LoadInt32(&r.stopped) == 1 {
-		return errors.New("broker has been stopped")
-	}
-
-	ctx := subscriber{upstream: upstream, topics: topics, done: make(chan error)}
-
-	r.listen <- ctx
-	return <-ctx.done
-}
-
-// SubscribePattern broker to pattern.
-func (r *Redis) SubscribePattern(upstream chan *Message, pattern string) error {
-	if atomic.LoadInt32(&r.stopped) == 1 {
-		return errors.New("broker has been stopped")
-	}
-
-	ctx := subscriber{upstream: upstream, pattern: pattern, done: make(chan error)}
-
-	r.listen <- ctx
-	return <-ctx.done
-}
-
-// Unsubscribe broker from one or multiple channels.
-func (r *Redis) Unsubscribe(upstream chan *Message, topics ...string) error {
-	if atomic.LoadInt32(&r.stopped) == 1 {
-		return errors.New("broker has been stopped")
-	}
-
-	ctx := subscriber{upstream: upstream, topics: topics, done: make(chan error)}
-
-	r.leave <- ctx
-	return <-ctx.done
-}
-
-// UnsubscribePattern broker from pattern.
-func (r *Redis) UnsubscribePattern(upstream chan *Message, pattern string) error {
-	if atomic.LoadInt32(&r.stopped) == 1 {
-		return errors.New("broker has been stopped")
-	}
-
-	ctx := subscriber{upstream: upstream, pattern: pattern, done: make(chan error)}
-
-	r.leave <- ctx
-	return <-ctx.done
-}
-
-// Publish one or multiple Channel.
-func (r *Redis) Publish(messages ...*Message) error {
-	if atomic.LoadInt32(&r.stopped) == 1 {
-		return errors.New("broker has been stopped")
-	}
-
-	for _, msg := range messages {
-		if err := r.client.Publish(context.Background(), msg.Topic, []byte(msg.Payload)).Err(); err != nil {
-			return err
-		}
-	}
-
-	return nil
-}
diff --git a/plugins/broadcast/redis/redis_test.go b/plugins/broadcast/redis/redis_test.go
deleted file mode 100644
index 37027e01..00000000
--- a/plugins/broadcast/redis/redis_test.go
+++ /dev/null
@@ -1,98 +0,0 @@
-package redis
-
-import (
-	"fmt"
-	"testing"
-
-	"github.com/sirupsen/logrus"
-	"github.com/sirupsen/logrus/hooks/test"
-	"github.com/stretchr/testify/assert"
-)
-
-func TestRedis_Error(t *testing.T) {
-	logger, _ := test.NewNullLogger()
-	logger.SetLevel(logrus.DebugLevel)
-
-	//c := service.NewContainer(logger)
-	//c.Register(rpc.ID, &rpc.Service{})
-	//c.Register(ID, &Service{})
-	//
-	//err := c.Init(&testCfg{
-	//	broadcast: `{"redis":{"addr":"localhost:6372"}}`,
-	//	rpc:       fmt.Sprintf(`{"join":"tcp://:%v"}`, rpcPort),
-	//})
-
-	rpcPort++
-
-	assert.Error(t, err)
-}
-
-func TestRedis_Broadcast(t *testing.T) {
-	br, _, c := setup(`{"redis":{"addr":"localhost:6379"}}`)
-	defer c.Stop()
-
-	client := br.NewClient()
-	defer client.Close()
-
-	assert.NoError(t, br.Broker().Publish(newMessage("topic", "hello1"))) // must not be delivered
-
-	assert.NoError(t, client.Subscribe("topic"))
-
-	assert.NoError(t, br.Broker().Publish(newMessage("topic", "hello1")))
-	assert.Equal(t, `hello1`, readStr(<-client.Channel()))
-
-	assert.NoError(t, br.Broker().Publish(newMessage("topic", "hello2")))
-	assert.Equal(t, `hello2`, readStr(<-client.Channel()))
-
-	assert.NoError(t, client.Unsubscribe("topic"))
-
-	assert.NoError(t, br.Broker().Publish(newMessage("topic", "hello3")))
-
-	assert.NoError(t, client.Subscribe("topic"))
-
-	assert.NoError(t, br.Broker().Publish(newMessage("topic", "hello4")))
-	assert.Equal(t, `hello4`, readStr(<-client.Channel()))
-}
-
-func TestRedis_BroadcastPattern(t *testing.T) {
-	br, _, c := setup(`{"redis":{"addr":"localhost:6379"}}`)
-	defer c.Stop()
-
-	client := br.NewClient()
-	defer client.Close()
-
-	assert.NoError(t, br.Broker().Publish(newMessage("topic", "hello1"))) // must not be delivered
-
-	assert.NoError(t, client.SubscribePattern("topic/*"))
-
-	assert.NoError(t, br.Broker().Publish(newMessage("topic/1", "hello1")))
-	assert.Equal(t, `hello1`, readStr(<-client.Channel()))
-
-	assert.NoError(t, br.Broker().Publish(newMessage("topic/2", "hello2")))
-	assert.Equal(t, `hello2`, readStr(<-client.Channel()))
-
-	assert.NoError(t, br.Broker().Publish(newMessage("different", "hello4")))
-	assert.NoError(t, br.Broker().Publish(newMessage("topic/2", "hello5")))
-
-	assert.Equal(t, `hello5`, readStr(<-client.Channel()))
-
-	assert.NoError(t, client.UnsubscribePattern("topic/*"))
-
-	assert.NoError(t, br.Broker().Publish(newMessage("topic/3", "hello6")))
-
-	assert.NoError(t, client.SubscribePattern("topic/*"))
-
-	assert.NoError(t, br.Broker().Publish(newMessage("topic/4", "hello7")))
-	assert.Equal(t, `hello7`, readStr(<-client.Channel()))
-}
-
-func TestRedis_NotActive(t *testing.T) {
-	b := &Redis{}
-	b.stopped = 1
-
-	assert.Error(t, b.Publish(nil))
-	assert.Error(t, b.Subscribe(nil))
-	assert.Error(t, b.Unsubscribe(nil))
-	assert.Error(t, b.SubscribePattern(nil, ""))
-	assert.Error(t, b.UnsubscribePattern(nil, ""))
-}