Feature: implement grpc server for publish and subscribe

Author: ThomasBoom89

Dockerfile

@@ -15,6 +15,7 @@ WORKDIR /app
 COPY --from=Builder /app/application /app
 
 EXPOSE 3000
+EXPOSE 3001
 
 CMD ["./application"]
 
@@ -26,3 +27,13 @@ RUN go install github.com/go-delve/delve/cmd/dlv@v1.22.1
 WORKDIR /app
 
 CMD ["air"]
+
+
+FROM golang:1.22-alpine3.19 AS protobuf-grpc-build
+
+WORKDIR /app
+
+RUN go install google.golang.org/protobuf/cmd/protoc-gen-go@v1.34.2
+RUN go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@v1.4
+RUN apk update && apk add protobuf
+

Makefile

@@ -7,3 +7,6 @@ live:
 
 down:
 	docker compose -f compose.yml -f compose.development.yml down
+
+proto:
+	docker compose -f compose.development.yml up application-proto

README.md

@@ -23,6 +23,8 @@ You can publish messages with http POST on /{topic}/publish message must be sent
 To receive messages you can http GET on /{topic}/subscribe or connect via websocket to /{topic}/ws
 On the websocket you need to send "next" in plaintext to get a new message
 
+You can use the gRPC endpoints to publish and subscribe to messages, see internal/grpc/messagebroker.proto
+
 ## License
 
 Simple Http Message Queue \

compose.development.yml

@@ -33,3 +33,10 @@ services:
       - ./go.mod:/app/go.mod
       - ./go.sum:/app/go.sum
 
+  application-proto:
+    container_name: application-proto
+    command: sh -c "protoc grpc/*.proto --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative --proto_path=."
+    build:
+      target: protobuf-grpc-build
+    volumes:
+      - ./internal/grpc:/app/grpc

go.mod

@@ -5,6 +5,8 @@ go 1.22
 require (
 	github.com/gofiber/contrib/websocket v1.3.1
 	github.com/gofiber/fiber/v2 v2.52.4
+	google.golang.org/grpc v1.64.0
+	google.golang.org/protobuf v1.34.2
 )
 
 require (
@@ -22,4 +24,6 @@ require (
 	github.com/valyala/tcplisten v1.0.0 // indirect
 	golang.org/x/net v0.24.0 // indirect
 	golang.org/x/sys v0.19.0 // indirect
+	golang.org/x/text v0.14.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240318140521-94a12d6c2237 // indirect
 )

go.sum

@@ -8,6 +8,8 @@ github.com/gofiber/contrib/websocket v1.3.1 h1:iINEnUIT7Wi1ttGWW5fY1fnKQlIEa5KTD
 github.com/gofiber/contrib/websocket v1.3.1/go.mod h1:oDLA6uM7x4hFq1zjy3US3HuvmrlWJKO5nrsw2ZKNSfY=
 github.com/gofiber/fiber/v2 v2.52.4 h1:P+T+4iK7VaqUsq2PALYEfBBo6bJZ4q3FP8cZ84EggTM=
 github.com/gofiber/fiber/v2 v2.52.4/go.mod h1:KEOE+cXMhXG0zHc9d8+E38hoX+ZN7bhOtgeF2oT6jrQ=
+github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
+github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/klauspost/compress v1.17.8 h1:YcnTYrq7MikUT7k0Yb5eceMmALQPYBW/Xltxn0NAMnU=
@@ -40,5 +42,13 @@ golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.19.0 h1:q5f1RH2jigJ1MoAWp2KTp3gm5zAGFUTarQZ5U386+4o=
 golang.org/x/sys v0.19.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
+golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240318140521-94a12d6c2237 h1:NnYq6UN9ReLM9/Y01KWNOWyI5xQ9kbIms5GGJVwS/Yc=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240318140521-94a12d6c2237/go.mod h1:WtryC6hu0hhx87FDGxWCDptyssuo68sk10vYjF+T9fY=
+google.golang.org/grpc v1.64.0 h1:KH3VH9y/MgNQg1dE7b3XfVK0GsPSIzJwdF617gUSbvY=
+google.golang.org/grpc v1.64.0/go.mod h1:oxjF8E3FBnjp+/gVFYdWacaLDx9na1aqy9oovLpxQYg=
+google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg=
+google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

internal/grpc/grpc.go

@@ -0,0 +1,34 @@
+package grpc
+
+import (
+	"context"
+	"github.com/ThomasBoom89/simple-http-message-queue/internal"
+	"github.com/gofiber/fiber/v2/log"
+)
+
+type Server struct {
+	topicManager *internal.TopicManager
+}
+
+func NewServer(topicManager *internal.TopicManager) *Server {
+	return &Server{topicManager: topicManager}
+}
+
+func (S *Server) Publish(ctx context.Context, request *PublishRequest) (*Empty, error) {
+	log.Debug(request.GetMessage())
+	S.topicManager.AddMessage(internal.Topic(request.GetTopic()), request.GetMessage())
+
+	return &Empty{}, nil
+}
+
+func (S *Server) Subscribe(ctx context.Context, request *SubscribeRequest) (*Response, error) {
+	message, err := S.topicManager.GetNextMessage(internal.Topic(request.GetTopic()))
+	if err != nil {
+		return &Response{Message: []byte{}}, nil
+	}
+
+	return &Response{Message: message}, nil
+}
+
+func (S *Server) mustEmbedUnimplementedMessageBrokerServer() {
+}