Removes redundant go routine

Author: ashishkujoy

cmd/server/main.go

@@ -4,9 +4,6 @@ import (
 	"ashishkujoy/queue/internal/config"
 	netinternal "ashishkujoy/queue/internal/net"
 	"log"
-	"os"
-	"os/signal"
-	"syscall"
 )
 
 func main() {
@@ -16,14 +13,6 @@ func main() {
 		log.Fatalf("Failed to create server: %v", err)
 		return
 	}
-	serverStop := make(chan interface{})
-	stop := make(chan os.Signal, 1)
-	signal.Notify(stop, syscall.SIGTERM, syscall.SIGINT)
 
-	err = server.Run(serverStop)
-	if err != nil {
-		return
-	}
-	<-stop
-	serverStop <- "stop"
+	err = server.Run()
 }

internal/net/server.go

@@ -7,7 +7,6 @@ import (
 	netinternal "ashishkujoy/queue/proto"
 	context "context"
 	"fmt"
-	"log"
 	"net"
 	"sync"
 
@@ -50,25 +49,30 @@ func NewQueueServer(config *config.Config, port string) (*QueueServer, 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")
 	}
-	qs.broadcastMessage()
+	go qs.broadcastMessage()
 	return &netinternal.EnqueueRequestResponse{Success: true}, nil
 }
 
 func (qs *QueueServer) broadcastMessage() {
-	//qs.mu.RLock()
-	//defer qs.mu.RUnlock()
 	var closedChannels []uint64
+	mu := sync.Mutex{}
+	wg := sync.WaitGroup{}
 	for _, consumer := range qs.onlineConsumer {
-		if err := qs.serveMessages(consumer); err != nil {
-			consumer.closeChannel <- "closed"
-			closedChannels = append(closedChannels, consumer.id)
-		}
+		wg.Add(1)
+		go func(consumer *OnlineConsumer) {
+			defer wg.Done()
+			if err := qs.serveMessages(consumer); err != nil {
+				mu.Lock()
+				defer mu.Unlock()
+				consumer.closeChannel <- "closed"
+				closedChannels = append(closedChannels, consumer.id)
+			}
+		}(consumer)
 	}
+	wg.Wait()
 	qs.onlineConsumer = internal.Filter(qs.onlineConsumer, func(consumer *OnlineConsumer) bool {
 		return !internal.Contains(closedChannels, consumer.id)
 	})
@@ -78,8 +82,6 @@ func (qs *QueueServer) ObserveQueue(req *netinternal.ObserveQueueRequest, 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
@@ -100,22 +102,15 @@ func (qs *QueueServer) serveMessages(consumer *OnlineConsumer) error {
 	return nil
 }
 
-func (qs *QueueServer) Run(cancel <-chan interface{}) error {
+func (qs *QueueServer) Run() error {
 	listener, err := net.Listen("tcp", ":50051")
 	if err != nil {
 		fmt.Printf("Error Creating listener %v\n", err)
 		return err
 	}
 	fmt.Println("Created Listener")
-	go func() {
-		if err := qs.gpServer.Serve(listener); err != nil {
-			fmt.Printf("Error Starting grpc serve %v", err)
-		}
-		log.Printf("GRPC server listening on %s", qs.port)
-
-	}()
-	<-cancel
-	_ = listener.Close()
-	qs.gpServer.Stop()
+	if err := qs.gpServer.Serve(listener); err != nil {
+		return nil
+	}
 	return nil
 }