upd(sam): pubsub events with redis option

sam/websocket/event_queue.go

@@ -2,9 +2,9 @@ package websocket
 
 import (
 	"context"
+	"github.com/crusttech/crust/sam/service"
 	"github.com/crusttech/crust/sam/types"
 	"github.com/titpetric/factory"
-	"log"
 	"time"
 )
 
@@ -23,10 +23,15 @@ type (
 
 	eventQueue struct {
 		origin uint64
+		pubsub *service.PubSub
 		queue  chan *types.EventQueueItem
 	}
 )
 
+const (
+	eventQueueBacklog = 512
+)
+
 var eq *eventQueue
 
 func init() {
@@ -36,7 +41,7 @@ func init() {
 func EventQueue(origin uint64) *eventQueue {
 	return &eventQueue{
 		origin: origin,
-		queue:  make(chan *types.EventQueueItem, 512),
+		queue:  make(chan *types.EventQueueItem, eventQueueBacklog),
 	}
 }
 
@@ -52,63 +57,89 @@ func (eq *eventQueue) store(ctx context.Context, qp eventQueuePusher) {
 	}()
 }
 
-func (eq *eventQueue) feedSessions(ctx context.Context, qp eventQueuePuller, store eventQueueWalker) {
-	var items []*types.EventQueueItem
+func (eq *eventQueue) feedSessions(ctx context.Context, qp eventQueuePuller, store eventQueueWalker) error {
+	newMessageEvent := make(chan struct{}, eventQueueBacklog)
+	done := make(chan error, 1)
 
-	go func() {
-		var err error
-	mainLoop:
-		for {
-			select {
-			case <-ctx.Done():
-				log.Printf("Error: %v", ctx.Err())
-			case <-time.After(time.Second * 1):
-				// How often do we check the database for new events?
-				// @todo make this interval configurable
-			}
+	// feed events from redis into newMessageEvent channel
+	if config.pubSubMode == "redis" && config.pubSubRedis != "" {
+		onConnect := func() error {
+			return nil
+		}
+		onMessage := func(message string, payload []byte) error {
+			newMessageEvent <- struct{}{}
+			return nil
+		}
+		pubsub := service.PubSub{}.New(config.pubSubRedis, ctx)
+		go func() {
+			done <- pubsub.Subscribe(onConnect, onMessage, "events")
+		}()
+	}
 
+	if config.pubSubMode == "poll" {
+		polling := func() error {
 			for {
-				items, err = qp.EventQueuePull(eq.origin)
-				if err != nil {
-					log.Printf("Error: %v", err)
-					return
+				select {
+				case <-ctx.Done():
+				case <-time.After(config.pubSubInterval):
+					newMessageEvent <- struct{}{}
 				}
-
-				if len(items) == 0 {
-					// No more items to sync, continue the mainLoop loop
-					continue mainLoop
-				}
-
-				var lastSyncedId uint64
-
-				for _, item := range items {
-					if item.Subscriber == "" {
-						// Distribute payload to all connected sessions
-						store.Walk(func(s *Session) {
-							s.sendBytes(item.Payload)
-						})
-					} else {
-						// Distribute payload to specific subscribers
-						store.Walk(func(s *Session) {
-							if s.subs.Get(item.Subscriber) != nil {
-								s.sendBytes(item.Payload)
-							}
-						})
-					}
-
-					lastSyncedId = item.ID
-
-				}
-
-				if lastSyncedId > 0 {
-					qp.EventQueueSync(eq.origin, lastSyncedId)
-				}
-
 			}
 		}
-	}()
+		go func() {
+			done <- polling()
+		}()
+	}
 
-	return
+	poll := func() error {
+		for {
+			items, err := qp.EventQueuePull(eq.origin)
+			if err != nil {
+				return err
+			}
+			if len(items) == 0 {
+				return nil
+			}
+
+			var lastSyncedId uint64
+
+			for _, item := range items {
+				if item.Subscriber == "" {
+					// Distribute payload to all connected sessions
+					store.Walk(func(s *Session) {
+						s.sendBytes(item.Payload)
+					})
+				} else {
+					// Distribute payload to specific subscribers
+					store.Walk(func(s *Session) {
+						if s.subs.Get(item.Subscriber) != nil {
+							s.sendBytes(item.Payload)
+						}
+					})
+				}
+
+				lastSyncedId = item.ID
+
+			}
+
+			if lastSyncedId > 0 {
+				qp.EventQueueSync(eq.origin, lastSyncedId)
+			}
+		}
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-newMessageEvent:
+			if err := poll(); err != nil {
+				return err
+			}
+		case err := <-done:
+			return err
+		}
+	}
 }
 
 // Adds origin to the event and puts it into queue.

sam/websocket/flags.go

@@ -1,6 +1,8 @@
 package websocket
 
 import (
+	"github.com/namsral/flag"
+	"github.com/pkg/errors"
 	"time"
 )
 
@@ -9,12 +11,24 @@ type (
 		writeTimeout time.Duration
 		pingTimeout  time.Duration
 		pingPeriod   time.Duration
+
+		pubSubMode     string
+		pubSubRedis    string
+		pubSubInterval time.Duration
 	}
 )
 
 var config configuration
 
 func (c configuration) validate() error {
+	switch c.pubSubMode {
+	case "redis", "poll":
+	default:
+		return errors.Errorf("Unknown pubSubMode: %s", c.pubSubMode)
+	}
+	if c.pubSubMode == "redis" && c.pubSubRedis == "" {
+		return errors.New("No host defined for mode=redis, pubSubRedis is empty")
+	}
 	return nil
 }
 
@@ -23,4 +37,8 @@ func Flags() {
 	config.writeTimeout = 15 * time.Second
 	config.pingTimeout = 120 * time.Second
 	config.pingPeriod = (config.pingTimeout * 10) / 9
+
+	flag.StringVar(&config.pubSubMode, "pubsub", "poll", "Pubsub mode (poll, redis)")
+	flag.StringVar(&config.pubSubRedis, "pubsub-redis", "", "Redis Pub/Sub hostname")
+	flag.DurationVar(&config.pubSubInterval, "pubsub-poll-interval", 3*time.Second, "Pub/Sub polling interval (3s, 12m, 3h...)")
 }

sam/websocket/router.go

@@ -16,7 +16,7 @@ func MountRoutes(ctx context.Context) func(chi.Router) {
 
 		repo := repository.New()
 
-		eq.feedSessions(ctx, repo, store)
+		go eq.feedSessions(ctx, repo, store)
 		eq.store(ctx, repo)
 
 		websocket := Websocket{}.New(svcUser)