From 8d675100ba4e77909b0dcea7ad60ae4cbbe8ff37 Mon Sep 17 00:00:00 2001 From: Valery Piashchynski Date: Wed, 12 May 2021 08:49:27 +0300 Subject: - Update diagramm - Remove old code Signed-off-by: Valery Piashchynski --- plugins/broadcast/doc/broadcast.drawio | 152 ++++++++++++++++++++++++++++- plugins/broadcast/interface.go | 2 +- plugins/broadcast/old/redis.go | 172 --------------------------------- plugins/broadcast/old/redis_test.go | 98 ------------------- plugins/broadcast/plugin.go | 10 +- plugins/broadcast/redis/storage.go | 1 + plugins/broadcast/ws/config.go | 1 + plugins/broadcast/ws/plugin.go | 1 + plugins/broadcast/ws/subscriber.go | 1 + 9 files changed, 161 insertions(+), 277 deletions(-) delete mode 100644 plugins/broadcast/old/redis.go delete mode 100644 plugins/broadcast/old/redis_test.go create mode 100644 plugins/broadcast/redis/storage.go create mode 100644 plugins/broadcast/ws/config.go create mode 100644 plugins/broadcast/ws/plugin.go create mode 100644 plugins/broadcast/ws/subscriber.go (limited to 'plugins') diff --git a/plugins/broadcast/doc/broadcast.drawio b/plugins/broadcast/doc/broadcast.drawio index f610396d..2339f5b1 100644 --- a/plugins/broadcast/doc/broadcast.drawio +++ b/plugins/broadcast/doc/broadcast.drawio @@ -1 +1,151 @@ -7VtZc+I4EP41VM08QNnyySNHrppkkgnJzmRfpoQtwBtjsbIJML9+JR/YsoTDYSCp2iRF7JYspO6vT8kNrTddXhE4m9xhF/kNoLjLhtZvAKBqBqD/GGWVUCzbTAhj4rlpp5ww8P6glKik1LnnopDrGGHsR96MJzo4CJATcTRICF7w3UbY5791BsdIIAwc6IvUn54bTVJqGyh5wzXyxpPsq4GStkxh1jslhBPo4kWBpF00tB7BOEqupsse8hn3MsYkz11uaF3PjKAg2uaBlX53+/TShYHzpPmvP4inBJ1mO51btMpWjFzKgPQWk2iCxziA/kVO7RI8D1zERlXoXd7nFuMZJaqU+A+KolUqTTiPMCVNoqmftqKlF/0qXL+woVpGetdfpiPHN6v0RlxtyoAQz4mDKpaY9mPrKjyY8ugK4SmKyIp2IMiHkffGCx6m+Bmv+60ffcAenQpQUqybmZxTpOu6wg8RQTJGUfpULid6UZhGToqlJ5dkr//iT/+23OV81vnmurc/+lHYVI1zilJpAaMozUpZUhGS1a9c6Oz2pdiWPxbfZc+NPN/vYR+TeHmaC5E9cig9jAh+RYUW07HRcFQHahKhVfQDR0FXhxC4KnSYMdSEm8Gnmzz4gF6yBaX+wKrsTy+SGeyLUE3/3R9No2/qsr304e9u7/blr6aamv436M9TPugfCLPqdphVOcTmAN6AWSk8aoOknM+WHIKp7JWWYgCTE3/mbQ80gbrRbllAVzWrramKppsG9y2Gup1FFMZtqop0utk4eDQK0aFGtYrlBcR2H+87/V5n8ETJ189UvMqXh9vnq5vvXwUo80BdTLwIDWYwFvGCRks8KGu0bW+IRGjZqLJGmcC0ks/KOLsoxDqZX5sUwhxT2YwNTga7MvysTmyXeGRfH7a/+oMtPVLt8c62ApUbo7NGmLWEJerpTLxExnKuniPIyIKEtRk2q4MMs21W9T9OkAGAYLJNKQJv4ZDmqhxqoO+NA3rtULkham67zJJ6NBnspA1Tz3UTgKLQ+wOH8XgMAinr6OBGt2H016AQTPE6U00fboA8k8zhUqFJG215k2JdtUAtjlxV7JYJOOE1T+d0RQn+HNTqXEcG+5U61/iHPYGDqEBPfupxuqAUzYD2lk63nFDW5nTlCvKhne4+Qfg+Bln7lE4XaGd1ujuXdWrxnVodPlEwhYZQ1WkBxbTbpmFQa6sDfrz6ajxVaCxYxseL/k3NxnE0Ao4083DNoWmYxzGCOji3EbQ+nxHcJ0zdR7P0j24Eq2ZdUJab7827i7v7x5eaFQaZGxTGag+VStbvrzCGeW6FsQX+0qX7NLQNGW/nw+bMn4+9IBSYTVccHS/wZkoWJvqm1sN6rcR6zdqS9aAG1stTAUPg/Qfy+NaW9qtUSbVOWUpVdjFpFaVUVQdtDhyZaPYNO05Q01TEKOKBDqV0n+uNJJDqGsiSGca2aWmwpkhC2Hc7ZQ1TuhN3nnQqjwqsj7CpJt+j/OiBhHTWMkeXnDZghxHYxwTRz6t7NuXAnaWa/GVE8HTd6uBg5I3FTYKTlaKK8qqC7dYlq6pKlGlavFnMijoHZmNNTWtp/MCqxg9Si+GsQm8BB4I0YThLTqGMvCXT5qJIJaFKWcjeND6OElefUk1XQU7ve9MxnbnvDekn/DMniC1yjAJEIJ39ZZedd0GkFb6Na7KtBmjxe3iaaFxtK+tTtK7W0WIfkesn3VD4qPsJUshuXb7aoNEnKleJEYlcyJ+vZg/eiR+BZvAKllmZAw1lKSgFLbN9KjMpFqkenru3N4Prw4LLOsyZZrL9QD6Ntrc0aNrRDJroV1SBVf+buE1wM0QTtxmWVZpYUsO9jsTsuqVq2nzykiUzG7dUq/vzW6pHSiqrpFBU+uuH8ys8KLNL1HZpbliHskv51Bb4NIBvaJ0lLNAwxM4rk1HmdDwcCHz8bBU0UIItkFX77ZNW0AQ5dB/vv108NrRL9geUZgOYcMogOY5iJiiD+TB0iDek7K0B15IoQ2BxRVzOJ0GqIgvLT+jDRFhT9vmMcTOOU+a/c3bcvTul1tqjsOzQVmW2pJ8JjxN6M2Iui7XphTYG+maKc9aWQp1rdpGDCYx1Ju7DxEJ8L0D5V9OrVKLJBIcZYRCRuRPRpCproZwYlntT2qxMm5AyZa9F+2gUJY02ayzP9wnPPCd+ph+PTrxgvF7DegYPcOVj6K77EbgQO1Ut5hhT77JKElnPacE2CLLqyLouonyZQOI6mOrR122nXFLDss2LYbQ2jckku5j2Gvnxixis9CXJttn9JZx6PtOua+S/ITZq7UpslYyioWuCEgOZc7KPZhTFPEwMRAO3w96roXcBDtCmIiRf2n+vsr//CyHHypI3WzlRngV5GRJxGXtFkkLoZ5VO35mKzQ+RLPP909RWu4Q7ozRQfYcY5Cg766H/0+207wpNObN2OoB0rgKOKub8nbsfB8b/+26zH2SUy9vsprFlxlDHNruct/Y51WWXs1wfQV2sLdVlQ13uuNm+WnpLy3gn2y/31+zd3uoq9T/SgWsxlZK/xPH5ird2paloKi0tw1vG8Fo8fZMflIZlLU3nB6mlkFOpQZJUbu6Xo3Hfyyi9CXJe46WPvZDKErEUJGIpS1gI3fP+4giPyPVIYVPVJZSDpPJhSizO6RBnU0oS6vAkpXjNkOToqiUpNO/hSeht/h55goT8dXzt4j8= \ No newline at end of file + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/plugins/broadcast/interface.go b/plugins/broadcast/interface.go index 093943af..716b3aac 100644 --- a/plugins/broadcast/interface.go +++ b/plugins/broadcast/interface.go @@ -18,7 +18,7 @@ type Subscriber interface { } type Storage interface { - + Store() } type Publisher interface { diff --git a/plugins/broadcast/old/redis.go b/plugins/broadcast/old/redis.go deleted file mode 100644 index 62970bc2..00000000 --- a/plugins/broadcast/old/redis.go +++ /dev/null @@ -1,172 +0,0 @@ -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 deleted file mode 100644 index 8148c155..00000000 --- a/plugins/broadcast/old/redis_test.go +++ /dev/null @@ -1,98 +0,0 @@ -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, "")) -} diff --git a/plugins/broadcast/plugin.go b/plugins/broadcast/plugin.go index f1ea258e..ec7c2587 100644 --- a/plugins/broadcast/plugin.go +++ b/plugins/broadcast/plugin.go @@ -13,6 +13,7 @@ const ( type Plugin struct { broker Subscriber + driver Storage log logger.Logger cfg *Config @@ -48,17 +49,16 @@ func (p *Plugin) Serve() chan error { // start the underlying broker go func() { - err := p.broker.Serve() - if err != nil { - errCh <- errors.E(op, err) - } + // err := p.broker.Serve() + // if err != nil { + // errCh <- errors.E(op, err) + // } }() return errCh } func (p *Plugin) Stop() error { - return nil } diff --git a/plugins/broadcast/redis/storage.go b/plugins/broadcast/redis/storage.go new file mode 100644 index 00000000..65a229e1 --- /dev/null +++ b/plugins/broadcast/redis/storage.go @@ -0,0 +1 @@ +package redis diff --git a/plugins/broadcast/ws/config.go b/plugins/broadcast/ws/config.go new file mode 100644 index 00000000..33da2485 --- /dev/null +++ b/plugins/broadcast/ws/config.go @@ -0,0 +1 @@ +package ws \ No newline at end of file diff --git a/plugins/broadcast/ws/plugin.go b/plugins/broadcast/ws/plugin.go new file mode 100644 index 00000000..98592950 --- /dev/null +++ b/plugins/broadcast/ws/plugin.go @@ -0,0 +1 @@ +package ws diff --git a/plugins/broadcast/ws/subscriber.go b/plugins/broadcast/ws/subscriber.go new file mode 100644 index 00000000..98592950 --- /dev/null +++ b/plugins/broadcast/ws/subscriber.go @@ -0,0 +1 @@ +package ws -- cgit v1.2.3