Adds cli client

Author: ashishkujoy

.gitignore

@@ -1,3 +1,6 @@
 .vscode
 .idea
-main
+main
+data
+cli-client
+.vscode

cmd/cli/cli-client.go

@@ -0,0 +1,83 @@
+package main
+
+import (
+	netinternal "ashishkujoy/queue/proto"
+	"context"
+	"flag"
+	"fmt"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"io"
+	"log"
+	"time"
+)
+
+type CLIOptions struct {
+	msg        string
+	publish    bool
+	consumerId uint64
+}
+
+func NewCLIOptions() *CLIOptions {
+	msg := flag.String("msg", "", "message to send")
+	publish := flag.Bool("publish", false, "publish message to the queue")
+	consumerId := flag.Uint64("consumer-id", 0, "consumer id")
+
+	flag.Parse()
+
+	return &CLIOptions{
+		msg:        *msg,
+		publish:    *publish,
+		consumerId: *consumerId,
+	}
+}
+
+func createQueueClient() netinternal.QueueServiceClient {
+	conn, err := grpc.NewClient("localhost:50051", grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		log.Fatalf("failed to connect: %v", err)
+	}
+	client := netinternal.NewQueueServiceClient(conn)
+	return client
+}
+
+func enqueueMsg(cliOptions *CLIOptions, client netinternal.QueueServiceClient) {
+	fmt.Printf("publishing message to queue: %s\n", cliOptions.msg)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	_, err := client.Enqueue(ctx, &netinternal.EnqueueRequest{Message: []byte(cliOptions.msg)})
+	if err != nil {
+		log.Fatalf("failed to enqueue: %v", err)
+	}
+	return
+}
+
+func observeQueueMsg(cliOptions *CLIOptions, client netinternal.QueueServiceClient) {
+	fmt.Printf("Observing message from queue, consumer id = %d\n", cliOptions.consumerId)
+	queue, err := client.ObserveQueue(context.Background(), &netinternal.ObserveQueueRequest{ConsumerId: cliOptions.consumerId})
+	if err != nil {
+		log.Fatalf("failed to observe: %v", err)
+	}
+	for {
+		queueMessage, err := queue.Recv()
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			log.Fatalf("failed to receive: %v", err)
+		}
+		fmt.Printf("received message: %s\n", string(queueMessage.Message))
+	}
+}
+
+func main() {
+	cliOptions := NewCLIOptions()
+	client := createQueueClient()
+
+	if cliOptions.publish {
+		enqueueMsg(cliOptions, client)
+		return
+	}
+
+	observeQueueMsg(cliOptions, client)
+}

cmd/main.go cmd/server/main.gocmd/main.go renamed to cmd/server/main.go

@@ -11,7 +11,7 @@ import (
 
 func main() {
 	conf := config.NewConfigWithMetadataPath("data/segments", "data/metadata", 1024*1024)
-	server, err := netinternal.NewQueueServer(conf, ":8081")
+	server, err := netinternal.NewQueueServer(conf, ":50051")
 	if err != nil {
 		log.Fatalf("Failed to create server: %v", err)
 		return

internal/net/server.go

@@ -37,10 +37,12 @@ func NewQueueServer(config *config.Config, port string) (*QueueServer, error) {
 	return server, nil
 }
 
-func (qs *QueueServer) Enqueue(ctx context.Context, req *netinternal.EnqueuRequest) (*netinternal.EnqueuRequestResponse, error) {
-	qs.queueService.Enqueue(req.Message)
+func (qs *QueueServer) Enqueue(ctx context.Context, req *netinternal.EnqueueRequest) (*netinternal.EnqueueRequestResponse, error) {
+	if err := qs.queueService.Enqueue(req.Message); err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to enqueue")
+	}
 	qs.broadcastMessage()
-	return nil, status.Errorf(codes.Unimplemented, "method Enqueue not implemented")
+	return &netinternal.EnqueueRequestResponse{Success: true}, nil
 }
 
 func (qs *QueueServer) broadcastMessage() {
@@ -57,6 +59,7 @@ func (qs *QueueServer) serveMessages(req *netinternal.ObserveQueueRequest, strea
 		if err != nil {
 			break
 		}
+		fmt.Printf("Message dequeued: %v\n", msg)
 		err = stream.Send(&netinternal.QueueMessage{Message: msg})
 		if err != nil {
 			return err

proto/queue.pb.go

@@ -2,11 +2,11 @@ syntax = "proto3";
 
 option go_package = "ashishkujoy/queue/net;netinternal";
 
-message EnqueuRequest {
+message EnqueueRequest {
     bytes message = 1;
 }
 
-message EnqueuRequestResponse {
+message EnqueueRequestResponse {
     bool success = 1;
 }
 
@@ -19,6 +19,6 @@ message QueueMessage {
 }
 
 service QueueService {
-    rpc Enqueue(EnqueuRequest) returns (EnqueuRequestResponse);
+    rpc Enqueue(EnqueueRequest) returns (EnqueueRequestResponse);
     rpc ObserveQueue(ObserveQueueRequest) returns (stream QueueMessage);
 }