Publishes new messages to online consumers

Author: ashishkujoy

cmd/server/main.go

@@ -19,11 +19,11 @@ func main() {
 	serverStop := make(chan interface{})
 	stop := make(chan os.Signal, 1)
 	signal.Notify(stop, syscall.SIGTERM, syscall.SIGINT)
-	go func() {
-		println("About to call run")
-		server.Run(serverStop)
-		println("Calling run done")
-	}()
+
+	err = server.Run(serverStop)
+	if err != nil {
+		return
+	}
 	<-stop
 	serverStop <- "stop"
 }

internal/arrayUtils.go

@@ -0,0 +1,22 @@
+package internal
+
+func Contains[T comparable](arr []T, element T) bool {
+	for _, e := range arr {
+		if e == element {
+			return true
+		}
+	}
+
+	return false
+}
+
+func Filter[T any](elements []T, f func(T) bool) []T {
+	var filtered []T
+	for _, element := range elements {
+		if f(element) {
+			filtered = append(filtered, element)
+		}
+	}
+
+	return filtered
+}

internal/net/server.go

@@ -1,24 +1,34 @@
 package netinternal
 
 import (
+	"ashishkujoy/queue/internal"
 	"ashishkujoy/queue/internal/config"
 	queueinternal "ashishkujoy/queue/internal/queue"
 	netinternal "ashishkujoy/queue/proto"
 	context "context"
 	"fmt"
 	"log"
 	"net"
+	"sync"
 
 	grpc "google.golang.org/grpc"
 	codes "google.golang.org/grpc/codes"
 	status "google.golang.org/grpc/status"
 )
 
+type MessageOutputStream = grpc.ServerStreamingServer[netinternal.QueueMessage]
+type OnlineConsumer struct {
+	id           uint64
+	stream       MessageOutputStream
+	closeChannel chan<- interface{}
+}
 type QueueServer struct {
 	netinternal.UnimplementedQueueServiceServer
-	queueService *queueinternal.QueueService
-	port         string
-	gpServer     *grpc.Server
+	queueService   *queueinternal.QueueService
+	port           string
+	gpServer       *grpc.Server
+	onlineConsumer []*OnlineConsumer
+	mu             *sync.RWMutex
 }
 
 func NewQueueServer(config *config.Config, port string) (*QueueServer, error) {
@@ -29,15 +39,19 @@ func NewQueueServer(config *config.Config, port string) (*QueueServer, error) {
 
 	gpServer := grpc.NewServer()
 	server := &QueueServer{
-		port:         port,
-		queueService: service,
-		gpServer:     gpServer,
+		port:           port,
+		queueService:   service,
+		gpServer:       gpServer,
+		onlineConsumer: make([]*OnlineConsumer, 0),
+		mu:             &sync.RWMutex{},
 	}
 	netinternal.RegisterQueueServiceServer(gpServer, server)
 	return server, nil
 }
 
-func (qs *QueueServer) Enqueue(ctx context.Context, req *netinternal.EnqueueRequest) (*netinternal.EnqueueRequestResponse, error) {
+func (qs *QueueServer) Enqueue(_ context.Context, req *netinternal.EnqueueRequest) (*netinternal.EnqueueRequestResponse, error) {
+	//qs.mu.Lock()
+	//defer qs.mu.Unlock()
 	if err := qs.queueService.Enqueue(req.Message); err != nil {
 		return nil, status.Errorf(codes.Internal, "failed to enqueue")
 	}
@@ -46,21 +60,38 @@ func (qs *QueueServer) Enqueue(ctx context.Context, req *netinternal.EnqueueRequ
 }
 
 func (qs *QueueServer) broadcastMessage() {
-	// panic("unimplemented")
+	//qs.mu.RLock()
+	//defer qs.mu.RUnlock()
+	var closedChannels []uint64
+	for _, consumer := range qs.onlineConsumer {
+		if err := qs.serveMessages(consumer); err != nil {
+			consumer.closeChannel <- "closed"
+			closedChannels = append(closedChannels, consumer.id)
+		}
+	}
+	qs.onlineConsumer = internal.Filter(qs.onlineConsumer, func(consumer *OnlineConsumer) bool {
+		return !internal.Contains(closedChannels, consumer.id)
+	})
 }
 
 func (qs *QueueServer) ObserveQueue(req *netinternal.ObserveQueueRequest, stream grpc.ServerStreamingServer[netinternal.QueueMessage]) error {
-	return qs.serveMessages(req, stream)
+	closeChannel := make(chan interface{})
+	consumer := &OnlineConsumer{id: req.ConsumerId, stream: stream, closeChannel: closeChannel}
+	_ = qs.serveMessages(consumer)
+	//qs.mu.Lock()
+	//defer qs.mu.Unlock()
+	qs.onlineConsumer = append(qs.onlineConsumer, consumer)
+	<-closeChannel
+	return nil
 }
 
-func (qs *QueueServer) serveMessages(req *netinternal.ObserveQueueRequest, stream grpc.ServerStreamingServer[netinternal.QueueMessage]) error {
+func (qs *QueueServer) serveMessages(consumer *OnlineConsumer) error {
 	for {
-		msg, err := qs.queueService.Dequeue(int(req.ConsumerId))
+		msg, err := qs.queueService.Dequeue(int(consumer.id))
 		if err != nil {
 			break
 		}
-		fmt.Printf("Message dequeued: %v\n", msg)
-		err = stream.Send(&netinternal.QueueMessage{Message: msg})
+		err = consumer.stream.Send(&netinternal.QueueMessage{Message: msg})
 		if err != nil {
 			return err
 		}
@@ -84,7 +115,7 @@ func (qs *QueueServer) Run(cancel <-chan interface{}) error {
 
 	}()
 	<-cancel
-	listener.Close()
+	_ = listener.Close()
 	qs.gpServer.Stop()
 	return nil
 }