tests

Author: kristoff-it

.github/workflows/test.yml

@@ -0,0 +1,17 @@
+name: Python package
+
+on:
+  push:
+    branches:
+      - master
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v2
+    - uses: actions/setup-go@v2-beta
+      with:
+        go-version: '1.14' # The Go version to download (if necessary) and use.
+    - run: go test
+
+

README.md

@@ -1,16 +1,20 @@
 # go-mpx
-RedisMPX implementation for the Go programming language.
+RedisMPX is a Redis Pub/Sub multiplexer library written in multiple languages and [live coded on Twitch](https://twitch.tv/kristoff_it).
+
+The live coding of this implementation is [archived on YouTube](https://www.youtube.com/watch?v=w6CyJ5dlFd4&list=PLVnMKELF_8IRJ37dW799IxkztIm1Dk3Az).
 
 ## Abstract
-RedisMPX is a Redis Pub/Sub multiplexer written in multiple languages and livecoded on Twitch.
+When bridging multiple application instances through Redis Pub/Sub it's easy to end up needing
+support for multiplexing. RedisMPX streamlines this process in a consistent way across multiple
+languages by offering a consistent set of features that cover the most common use cases.
 
-- [Twitch channel](https://twitch.tv/kristoff_it)
-- [YouTube VOD archive](https://www.youtube.com/user/Kappaloris/videos)
+The library works under the assumption that you are going to create separate subscriptions
+for each client connected to your service (e.g. WebSockets clients):
 
-## Status
-Main functionality completed. No unit tests have been written as the real
-source of complexity is how goroutines and concurrency mix with network I/O etc.
-So instead of wasting effort with mostly useless unit tests, I aim to write a fuzzer for that purpose.
+- ChannelSubscription allows you to add and remove individual Redis
+  PubSub channels similarly to how a multi-room chat application would need to.
+- PatternSubscription allows you to subscribe to a single Redis Pub/Sub pattern.
+- PromiseSubscription allows you to create a networked promise system.
 
 ## Features
 - Simple channel subscriptions

channel_subscription.go renamed to channel.go

@@ -51,11 +51,11 @@ func (s *ChannelSubscription) Add(channels ...string) {
 		panic("tried to use a closed subscription")
 	}
 	for _, ch := range channels {
-		_, ok := s.channels[ch]
+		_, ok := s.Channels[ch]
 		if !ok {
 			node := list.NewElement(s)
 			s.mpx.reqCh <- request{subscriptionAdd, ch, node}
-			s.channels[ch] = node
+			s.Channels[ch] = node
 		}
 	}
 }
@@ -68,10 +68,10 @@ func (s *ChannelSubscription) Remove(channels ...string) {
 		panic("tried to use a closed subscription")
 	}
 	for _, ch := range channels {
-		node, ok := s.channels[ch]
+		node, ok := s.Channels[ch]
 		if ok {
 			s.mpx.reqCh <- request{subscriptionRemove, ch, node}
-			delete(s.channels, ch)
+			delete(s.Channels, ch)
 		}
 	}
 }
@@ -81,12 +81,12 @@ func (s *ChannelSubscription) Clear() {
 	if s.closed {
 		panic("tried to use a closed ChannelSubscription")
 	}
-	for ch := range s.channels {
+	for ch := range s.Channels {
 		s.Remove(ch)
 	}
 
 	// Reset our internal state
-	s.channels = make(map[string]*list.Element)
+	s.Channels = make(map[string]*list.Element)
 }
 
 // Calls Clear and frees all references from the Multiplexer.

channel_test.go

@@ -0,0 +1,78 @@
+package mpx
+
+import (
+	"fmt"
+	"github.com/gomodule/redigo/redis"
+	"testing"
+	"time"
+)
+
+func TestChannel(t *testing.T) {
+	connBuilder := func() (redis.Conn, error) {
+		return redis.Dial("tcp", ":6379")
+	}
+
+	conn, err := connBuilder()
+	if err != nil {
+		panic(err)
+	}
+
+	multiplexer := New(connBuilder)
+
+	messages := make(chan []byte, 100)
+	onMessage := func(_ string, msg []byte) {
+		messages <- msg
+	}
+
+	errors := make(chan error, 100)
+	onDisconnect := func(err error) {
+		errors <- err
+	}
+
+	activations := make(chan string, 100)
+	onActivation := func(ch string) {
+		activations <- ch
+	}
+
+	sub := multiplexer.NewChannelSubscription(onMessage, onDisconnect, onActivation)
+
+	// Activation works
+	{
+		// Add a Redis Pub/Sub channel to the Subscription
+		sub.Add("mychannel")
+		timer := time.NewTimer(3 * time.Second)
+		select {
+		case <-activations:
+			break
+		case <-timer.C:
+			t.Errorf("timed out while waiting for the 1st activation")
+		}
+	}
+
+	// Can receive messages
+	{
+		_, err := conn.Do("PUBLISH", "mychannel", "Hello World 1!")
+		if err != nil {
+			panic(err)
+		}
+
+		_, err = conn.Do("PUBLISH", "mychannel", "Hello World 2!")
+		if err != nil {
+			panic(err)
+		}
+
+		timer := time.NewTimer(3 * time.Second)
+		for i := 1; i <= 2; i += 1 {
+			select {
+			case m := <-messages:
+				mm := fmt.Sprintf("Hello World %v!", i)
+				if string(m) != mm {
+					t.Errorf("Messages are not equal. Got [%v], expected [%v]", string(m), mm)
+				}
+			case <-timer.C:
+				t.Errorf("timed out while waiting for messages")
+			}
+		}
+	}
+
+}

multiplexer.go

@@ -7,7 +7,6 @@ package mpx
 
 import (
 	"errors"
-	"fmt"
 	"github.com/RedisMPX/go-mpx/internal"
 	"github.com/RedisMPX/go-mpx/internal/list"
 	"github.com/gomodule/redigo/redis"
@@ -57,15 +56,17 @@ type request struct {
 type Multiplexer struct {
 	// Options
 	pingTimeout     time.Duration
-	minBackOff      time.Duration
-	maxBackOff      time.Duration
+	minBackoff      time.Duration
+	maxBackoff      time.Duration
 	pipeliningSlots []request
 
 	// state
 	createConn       func() (redis.Conn, error)
 	pubsub           redis.PubSubConn
 	channels         map[string]*list.List
 	patterns         map[string]*list.List
+	active_channels  map[string]struct{}
+	active_patterns  map[string]struct{}
 	reqCh            chan request
 	stop             chan struct{}
 	exit             chan struct{}
@@ -100,20 +101,22 @@ func New(createConn func() (redis.Conn, error)) *Multiplexer {
 func NewWithOpts(
 	createConn func() (redis.Conn, error),
 	pingTimeout time.Duration,
-	minBackOff time.Duration,
-	maxBackOff time.Duration,
+	minBackoff time.Duration,
+	maxBackoff time.Duration,
 	messagesBufSize uint,
 	pipeliningBufSize uint,
 ) *Multiplexer {
 	mpx := Multiplexer{
 		pingTimeout:      pingTimeout,
-		minBackoff:       minBackOff,
+		minBackoff:       minBackoff,
 		maxBackoff:       maxBackoff,
 		pipeliningSlots:  make([]request, pipeliningBufSize),
 		pubsub:           redis.PubSubConn{Conn: nil},
 		createConn:       createConn,
 		channels:         make(map[string]*list.List),
 		patterns:         make(map[string]*list.List),
+		active_channels:  make(map[string]struct{}),
+		active_patterns:  make(map[string]struct{}),
 		reqCh:            make(chan request, 100),
 		stop:             make(chan struct{}),
 		exit:             make(chan struct{}),
@@ -267,7 +270,7 @@ func (mpx *Multiplexer) openNewConnection() {
 			break
 		}
 		if errorCount > 0 {
-			time.Sleep(internal.RetryBackoff(errorCount, mpx.minBackOff, mpx.maxBackOff))
+			time.Sleep(internal.RetryBackoff(errorCount, mpx.minBackoff, mpx.maxBackoff))
 		}
 		errorCount++
 	}
@@ -301,6 +304,10 @@ func (mpx *Multiplexer) connectionGoroutine() {
 					return
 				}
 			}
+
+			// Reset the active channel / pattern sets.
+			mpx.active_channels = make(map[string]struct{})
+			mpx.active_patterns = make(map[string]struct{})
 		}
 
 		// Start an "emergency" goroutine that keeps processing
@@ -458,27 +465,21 @@ func (mpx *Multiplexer) processRequests(requests []request, networkIO bool) erro
 					// Subscribe in Redis
 					sub = append(sub, req.name)
 				}
-			} else {
-				// We were already subscribed to that channel
-				if networkIO {
-					// If we are not in offline-mode, we immediately send confirmation
-					// that the subscription is active. If we are in the case where
-					// we just lost connectivity and processRequestGoroutine has not yet
-					// noticed, we will send a wrong notification, but we will also soon
-					// send a onDisconnect notification, after this goroutine exits, thus
-					// rectifying our wrong communication.
-					if onActivation := req.node.Value.(*ChannelSubscription).onActivation; onActivation != nil {
-						onActivation(req.name)
-					}
+			}
+
+			// Trigger onActivation if the subscription is already active.
+			if _, active := mpx.active_channels[req.name]; active {
+				if onActivation := req.node.Value.(*ChannelSubscription).onActivation; onActivation != nil {
+					onActivation(req.name)
 				}
 			}
+
 		case subscriptionRemove:
 			if listeners := req.node.DetachFromList(); listeners != nil {
 				if listeners.Len() > 0 {
-					fmt.Printf("[ws] unsubbed but more remaining (%v)\n", listeners.Len())
 				} else {
-					fmt.Printf("[ws] unsubbed also from Redis\n")
 					delete(mpx.channels, req.name)
+					delete(mpx.active_channels, req.name)
 					if networkIO {
 						unsub = append(unsub, req.name)
 					}
@@ -509,18 +510,12 @@ func (mpx *Multiplexer) processRequests(requests []request, networkIO bool) erro
 					// PSubscribe in Redis
 					psub = append(psub, req.name)
 				}
-			} else {
-				// We were already subscribed to that channel
-				if networkIO {
-					// If we are not in offline-mode, we immediately send confirmation
-					// that the subscription is active. If we are in the case where
-					// we just lost connectivity and processRequestGoroutine has not yet
-					// noticed, we will send a wrong notification, but we will also soon
-					// send a onDisconnect notification, after this goroutine exits, thus
-					// rectifying our wrong communication.
-					if onActivation := req.node.Value.(*PatternSubscription).onActivation; onActivation != nil {
-						onActivation(req.name)
-					}
+			}
+
+			// Trigger onActivation if the subscription is already active.
+			if _, active := mpx.active_patterns[req.name]; active {
+				if onActivation := req.node.Value.(*PatternSubscription).onActivation; onActivation != nil {
+					onActivation(req.name)
 				}
 			}
 		case patternClose:
@@ -531,10 +526,9 @@ func (mpx *Multiplexer) processRequests(requests []request, networkIO bool) erro
 			onMessageNode := req.node.Value.(*PatternSubscription).onMessageNode
 			if listeners := onMessageNode.DetachFromList(); listeners != nil {
 				if listeners.Len() > 0 {
-					fmt.Printf("[ws] unsubbed but more remaining (%v)\n", listeners.Len())
 				} else {
-					fmt.Printf("[ws] unsubbed also from Redis\n")
 					delete(mpx.patterns, req.name)
+					delete(mpx.active_patterns, req.name)
 					if networkIO {
 						punsub = append(punsub, req.name)
 					}
@@ -711,7 +705,6 @@ func (mpx *Multiplexer) messageReadingGoroutine(size int) {
 			// TODO: ask redigo do accept pings also from the pubsub
 			//       interface.
 			if !strings.HasSuffix(msg.Error(), "got type string") {
-				fmt.Printf("error in Pubsub: %v", msg)
 				mpx.triggerReconnect(msg)
 				return
 			}

multiplexer_test.go

@@ -0,0 +1,75 @@
+package mpx
+
+import (
+	"github.com/gomodule/redigo/redis"
+	"testing"
+	"time"
+)
+
+func TestMultiplexer(t *testing.T) {
+	connBuilder := func() (redis.Conn, error) {
+		return redis.Dial("tcp", ":6379")
+	}
+
+	conn, err := connBuilder()
+	if err != nil {
+		panic(err)
+	}
+
+	multiplexer := New(connBuilder)
+
+	messages := make(chan []byte, 100)
+	onMessage := func(_ string, msg []byte) {
+		messages <- msg
+	}
+
+	errors := make(chan error, 100)
+	onDisconnect := func(err error) {
+		errors <- err
+	}
+
+	activations := make(chan string, 100)
+	onActivation := func(ch string) {
+		activations <- ch
+	}
+
+	sub := multiplexer.NewChannelSubscription(onMessage, onDisconnect, onActivation)
+
+	// Activation works
+	{
+		// Add a Redis Pub/Sub channel to the Subscription
+		sub.Add("mychannel")
+		timer := time.NewTimer(3 * time.Second)
+		select {
+		case <-activations:
+			break
+		case <-timer.C:
+			t.Errorf("timed out while waiting for the 1st activation")
+		}
+	}
+
+	// Able to reconnect
+	{
+		_, err := conn.Do("client", "kill", "type", "pubsub")
+		if err != nil {
+			panic(err)
+		}
+
+		timer := time.NewTimer(3 * time.Second)
+
+		select {
+		case <-errors:
+			break
+		case <-timer.C:
+			t.Errorf("timed out while waiting for the error notification")
+		}
+
+		select {
+		case <-activations:
+			break
+		case <-timer.C:
+			t.Errorf("timed out while waiting for the 2nd activation")
+		}
+	}
+
+}