- Initial broadcast commit

Signed-off-by: Valery Piashchynski <[email protected]>

2 files changed, 270 insertions, 0 deletions

diff --git a/plugins/broadcast/old/redis.go b/plugins/broadcast/old/redis.go
new file mode 100644
index 00000000..62970bc2
--- /dev/null
+++ b/plugins/broadcast/old/redis.go
@@ -0,0 +1,172 @@
+package old
+
+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/old/redis_test.go b/plugins/broadcast/old/redis_test.go
new file mode 100644
index 00000000..8148c155
--- /dev/null
+++ b/plugins/broadcast/old/redis_test.go
@@ -0,0 +1,98 @@
+package old
+
+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, ""))
+}