Adding a queue implementation using RedisStreams

- queue implementation still implements a pubsub interface, but uses
RedisStreams with groups instead, which enables one subscriber in a
group to receive each message, rather than all subscribers
- Added some comments for pubsub

Author: TerrenceHo

go.mod

@@ -27,6 +27,7 @@ require (
 	golang.org/x/crypto v0.0.0-20190422183909-d864b10871cd
 	golang.org/x/net v0.0.0-20190424024845-afe8014c977f // indirect
 	golang.org/x/sys v0.0.0-20190516110030-61b9204099cb // indirect
+	golang.org/x/xerrors v0.0.0-20190513163551-3ee3066db522
 	google.golang.org/appengine v1.5.0 // indirect
 	gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 // indirect
 )

go.sum

@@ -111,6 +111,8 @@ golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20190516110030-61b9204099cb h1:k07iPOt0d6nEnwXF+kHB+iEg+WSuKe/SOQuFM2QoD+E=
 golang.org/x/sys v0.0.0-20190516110030-61b9204099cb/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/xerrors v0.0.0-20190513163551-3ee3066db522 h1:bhOzK9QyoD0ogCnFro1m2mz41+Ib0oOhfJnBp5MR4K4=
+golang.org/x/xerrors v0.0.0-20190513163551-3ee3066db522/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/appengine v1.5.0 h1:KxkO13IPW4Lslp2bz+KHP2E3gtFlrIGNThxkZQ3g+4c=
 google.golang.org/appengine v1.5.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

stores/redis/pubsub.go

@@ -2,6 +2,8 @@ package redis
 
 import "github.com/go-redis/redis"
 
+// RedisPubSub implements the PublishSubscribe interface, but publishes to all
+// subscribers. All subscribers will receive the message.
 type RedisPubSub struct {
 	client *redis.Client
 }

stores/redis/queue.go

@@ -0,0 +1,109 @@
+package redis
+
+import (
+	"bytes"
+	"encoding/gob"
+	"fmt"
+	"os"
+
+	"github.com/go-redis/redis"
+)
+
+const defaultKey = "value" // a default key needed to use RedisStreams
+
+// RedisQueue implements the PublishSubscribe interface, but only publishes to
+// one subscriber. Implemented by having Publishers push into a queue, and
+// Subscribers read from the queue and pop off from the top.
+type RedisQueue struct {
+	client *redis.Client
+}
+
+// NewRedisQueue creates a new queue and a group associated with that queue.
+// Underlying mecahnism uses Redis Streams.
+func NewRedisQueue(client *redis.Client, queueID, groupID string) *RedisQueue {
+	_, err := client.XGroupCreateMkStream(queueID, groupID, "$").Result()
+	if err != nil {
+		fmt.Println("Error creating redis group stream")
+		os.Exit(1)
+	}
+	return &RedisQueue{
+		client: client,
+	}
+}
+
+// Subscribe creates a goroutine that subscribes to a RedisStream, based on
+// queueID, groupID, consumerID. Sends data values to a msg []byte channel.
+func (rq *RedisQueue) Subscribe(queueID, groupID, consumerID string, msg chan []byte) error {
+	// Create Subscription
+
+	// Read from stram (do in loop)
+	// XREADGROUP GROUP queueGROUP ConsumerID COUNT 1 STREAMS queueID >
+
+	// Acknowledge that message was processed
+	// XACK queueID queueGROUP MSGID
+
+	go func() {
+		args := &redis.XReadGroupArgs{
+			Group:    groupID,
+			Consumer: consumerID,
+			Streams:  []string{queueID},
+			Count:    1,
+			Block:    0,
+			NoAck:    false,
+		}
+
+		for {
+			xstreams, err := rq.client.XReadGroup(args).Result()
+			if err != nil {
+				// handle error, prob by logging
+			}
+			xstream := xstreams[0]                          // only asking for one stream
+			message := xstream.Messages[0]                  // asking for one message
+			msgID := message.ID                             // retrieve message ID
+			value, err := getBytes(message.Values["value"]) // retrieve value using defaultKey, transform to bytes
+			if err != nil {
+				// log gob decoding error
+			}
+
+			msg <- value // send the value
+
+			// Ack the read
+			_, err = rq.client.XAck(queueID, groupID, msgID).Result()
+			if err != nil {
+				// log ack error
+			}
+		}
+	}()
+
+	return nil
+}
+
+// Publish value to a queue, based on queueID.
+func (rq *RedisQueue) Publish(queueID string, value []byte) error {
+	// XADD queueID * field value
+	var m map[string]interface{}
+	m["value"] = value
+	args := &redis.XAddArgs{
+		Stream:       queueID,
+		MaxLen:       1, // MAXLEN N
+		MaxLenApprox: 1, // MAXLEN ~ N
+		ID:           "*",
+		Values:       m,
+	}
+
+	err := rq.client.XAdd(args).Err()
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func getBytes(data interface{}) ([]byte, error) {
+	var buf bytes.Buffer
+	enc := gob.NewEncoder(&buf)
+	err := enc.Encode(data)
+	if err != nil {
+		return nil, err
+	}
+	return buf.Bytes(), nil
+}