Temp commit, ammend it later

Author: ashishkujoy

cmd/main.go

@@ -0,0 +1,29 @@
+package main
+
+import (
+	"ashishkujoy/queue/internal/config"
+	netinternal "ashishkujoy/queue/internal/net"
+	"log"
+	"os"
+	"os/signal"
+	"syscall"
+)
+
+func main() {
+	conf := config.NewConfigWithMetadataPath("data/segments", "data/metadata", 1024*1024)
+	server, err := netinternal.NewQueueServer(conf, ":8081")
+	if err != nil {
+		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)
+	go func() {
+		println("About to call run")
+		server.Run(serverStop)
+		println("Calling run done")
+	}()
+	<-stop
+	serverStop <- "stop"
+}

generate-grcp-code.sh

@@ -0,0 +1,5 @@
+#! /bin/bash
+
+protoc --go_out=. --go_opt=paths=source_relative \
+       --go-grpc_out=. --go-grpc_opt=paths=source_relative \
+       proto/queue.proto

go.mod

@@ -10,6 +10,11 @@ require (
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/rakyll/gotest v0.0.6 // indirect
 	github.com/stretchr/testify v1.10.0 // indirect
+	golang.org/x/net v0.35.0 // indirect
 	golang.org/x/sys v0.32.0 // indirect
+	golang.org/x/text v0.22.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20250218202821-56aae31c358a // indirect
+	google.golang.org/grpc v1.72.0 // indirect
+	google.golang.org/protobuf v1.36.5 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -16,11 +16,21 @@ github.com/rakyll/gotest v0.0.6 h1:hBTqkO3jiuwYW/M9gL4bu0oTYcm8J6knQAAPUsJsz1I=
 github.com/rakyll/gotest v0.0.6/go.mod h1:SkoesdNCWmiD4R2dljIUcfSnNdVZ12y8qK4ojDkc2Sc=
 github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
 github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+golang.org/x/net v0.35.0 h1:T5GQRQb2y08kTAByq9L4/bz8cipCdA8FbRTXewonqY8=
+golang.org/x/net v0.35.0/go.mod h1:EglIi67kWsHKlRzzVMUD93VMSWGFOMSZgxFjparz1Qk=
 golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.32.0 h1:s77OFDvIQeibCmezSnk/q6iAfkdiQaJi4VzroCFrN20=
 golang.org/x/sys v0.32.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
+golang.org/x/text v0.22.0 h1:bofq7m3/HAFvbF51jz3Q9wLg3jkvSPuiZu/pD1XwgtM=
+golang.org/x/text v0.22.0/go.mod h1:YRoo4H8PVmsu+E3Ou7cqLVH8oXWIHVoX0jqUWALQhfY=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20250218202821-56aae31c358a h1:51aaUVRocpvUOSQKM6Q7VuoaktNIaMCLuhZB6DKksq4=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20250218202821-56aae31c358a/go.mod h1:uRxBH1mhmO8PGhU89cMcHaXKZqO+OfakD8QQO0oYwlQ=
+google.golang.org/grpc v1.72.0 h1:S7UkcVa60b5AAQTaO6ZKamFp1zMZSU0fGDK2WZLbBnM=
+google.golang.org/grpc v1.72.0/go.mod h1:wH5Aktxcg25y1I3w7H69nHfXdOG3UiadoBtjh3izSDM=
+google.golang.org/protobuf v1.36.5 h1:tPhr+woSbjfYvY6/GPufUoYizxw1cF/yFoxJ2fmpwlM=
+google.golang.org/protobuf v1.36.5/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

internal/net/server.go

@@ -0,0 +1,86 @@
+package netinternal
+
+import (
+	"ashishkujoy/queue/internal/config"
+	queueinternal "ashishkujoy/queue/internal/queue"
+	context "context"
+	"fmt"
+	"log"
+	"net"
+
+	grpc "google.golang.org/grpc"
+	codes "google.golang.org/grpc/codes"
+	status "google.golang.org/grpc/status"
+)
+
+type QueueServer struct {
+	UnimplementedQueueServiceServer
+	queueService *queueinternal.QueueService
+	port         string
+	gpServer     *grpc.Server
+}
+
+func NewQueueServer(config *config.Config, port string) (*QueueServer, error) {
+	service, err := queueinternal.NewQueueService(config)
+	if err != nil {
+		return nil, err
+	}
+
+	gpServer := grpc.NewServer()
+	server := &QueueServer{
+		port:         port,
+		queueService: service,
+		gpServer:     gpServer,
+	}
+	RegisterQueueServiceServer(gpServer, server)
+	return server, nil
+}
+
+func (qs *QueueServer) Enqueue(ctx context.Context, req *EnqueuRequest) (*EnqueuRequestResponse, error) {
+	qs.queueService.Enqueue(req.Message)
+	qs.broadcastMessage()
+	return nil, status.Errorf(codes.Unimplemented, "method Enqueue not implemented")
+}
+
+func (qs *QueueServer) broadcastMessage() {
+	// panic("unimplemented")
+}
+
+func (qs *QueueServer) ObserveQueue(req *ObserveQueueRequest, stream grpc.ServerStreamingServer[QueueMessage]) error {
+	return qs.serveMessages(req, stream)
+}
+
+func (qs *QueueServer) serveMessages(req *ObserveQueueRequest, stream grpc.ServerStreamingServer[QueueMessage]) error {
+	for {
+		msg, err := qs.queueService.Dequeue(int(req.ConsumerId))
+		if err != nil {
+			break
+		}
+		err = stream.Send(&QueueMessage{Message: msg})
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (qs *QueueServer) Run(cancel <-chan interface{}) 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()
+	return nil
+}

proto/queue.proto

@@ -0,0 +1,24 @@
+syntax = "proto3";
+
+option go_package = "ashishkujoy/queue/net;netinternal";
+
+message EnqueuRequest {
+    bytes message = 1;
+}
+
+message EnqueuRequestResponse {
+    bool success = 1;
+}
+
+message ObserveQueueRequest {
+    uint64 consumerId = 1;
+}
+
+message QueueMessage {
+    bytes message = 1;
+}
+
+service QueueService {
+    rpc Enqueue(EnqueuRequest) returns (EnqueuRequestResponse);
+    rpc ObserveQueue(ObserveQueueRequest) returns (stream QueueMessage);
+}