Add gRPC and Http SSE (#28)

* Add grpc

* Ignore generated files

* Update makefile command

* Fix docker

* Add rpc server

* Fix make test

* Fix build

* Fix make check

* Add SSE streaming

* Remove generated swagger docs

* Adjust gitignore

* Fix makefile

.gitignore

@@ -24,4 +24,5 @@ dist.zip
 frontend/dist/
 
 # Ignore generated files
-*.pb.go
+*.pb.go
+**/swaggo-gen/docs.go

Makefile

@@ -19,15 +19,19 @@ gen-proto: install-tools
 		--go-grpc_out=. --go-grpc_opt=paths=source_relative \
 		service/proto/*.proto
 
-.PHONY: build
-build: install-tools
-	@echo "Building on $(OS)"
+.PHONY: gen-swagger
+gen-swagger: install-tools
 	swag fmt -d application/server
 	swag init -d application/server -ot go -o application/server/swaggo-gen
+
+.PHONY: build
+build: gen-proto gen-swagger
+	@echo "Building on $(OS)"
 	go mod tidy
 	go build -o bin/migrate_db application/migrate_db/main.go
 	go build -o bin/service application/server/main.go
 	go build -o bin/asynq_worker application/asynq_worker/main.go
+	go build -o bin/rpc_server application/rpc_server/main.go
 
 .PHONY: clear-db
 clear-db:
@@ -42,11 +46,11 @@ setup-db: clear-db build
 	./bin/migrate_db
 
 .PHONY: check
-check:
+check: gen-proto
 	go vet ./...
 
 .PHONY: test
-test: setup-db check 
+test: gen-swagger check setup-db
 	go test -cover -v ./...
 
 .PHONY: run-task-worker
@@ -57,6 +61,10 @@ run-task-worker: build check
 run-server: build check
 	./bin/service
 
+.PHONY: run-rpc-server
+run-rpc-server: build check
+	./bin/rpc_server
+
 .PHONY: run
 run: build check
 	make -j run-task-worker run-server

application/rpc_server/impls/greeter.go

@@ -0,0 +1,17 @@
+package impls
+
+import (
+	"context"
+	"log"
+
+	"github.com/OJ-lab/oj-lab-services/service/proto"
+)
+
+type GreeterServer struct {
+	proto.UnimplementedGreeterServer
+}
+
+func (s *GreeterServer) Greeting(ctx context.Context, request *proto.GreetingRequest) (*proto.GreetingResponse, error) {
+	log.Printf("Received: %v", request.GetName())
+	return &proto.GreetingResponse{Message: "Hello " + request.GetName()}, nil
+}

application/rpc_server/impls/streamer.go

@@ -0,0 +1,32 @@
+package impls
+
+import (
+	"log"
+	"time"
+
+	"github.com/OJ-lab/oj-lab-services/service/proto"
+)
+
+type StreamerServer struct {
+	proto.UnimplementedStreamerServer
+}
+
+func (s *StreamerServer) StartStream(request *proto.StreamRequest, server proto.Streamer_StartStreamServer) error {
+	tick := time.NewTicker(1 * time.Second)
+	for range tick.C {
+		if server.Context().Err() != nil {
+			if server.Context().Err().Error() == "context canceled" {
+				log.Printf("client closed stream")
+				return nil
+			}
+			log.Printf("client closed stream with: %v", server.Context().Err().Error())
+			return nil
+		}
+
+		server.Send(&proto.StreamResponse{Body: &proto.StreamResponse_Health{
+			Health: &proto.ServerHealth{},
+		}})
+	}
+
+	return nil
+}

application/rpc_server/main.go

@@ -0,0 +1,39 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+	"log"
+	"net"
+
+	"github.com/OJ-lab/oj-lab-services/application/rpc_server/impls"
+	"github.com/OJ-lab/oj-lab-services/core"
+	"github.com/OJ-lab/oj-lab-services/service/proto"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/reflection"
+)
+
+const (
+	portProp = "rpc-server.port"
+)
+
+var (
+	port = core.AppConfig.GetInt(portProp)
+)
+
+func main() {
+	flag.Parse()
+	lis, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
+	if err != nil {
+		log.Fatalf("failed to listen: %v", err)
+	}
+	s := grpc.NewServer()
+	proto.RegisterGreeterServer(s, &impls.GreeterServer{})
+	proto.RegisterStreamerServer(s, &impls.StreamerServer{})
+
+	reflection.Register(s)
+	log.Printf("server listening at %v", lis.Addr())
+	if err := s.Serve(lis); err != nil {
+		log.Fatalf("failed to serve: %v", err)
+	}
+}

application/server/handler/user.go

@@ -1,12 +1,16 @@
 package handler
 
 import (
+	"fmt"
+	"io"
 	"net/http"
+	"time"
 
 	"github.com/OJ-lab/oj-lab-services/core"
 	"github.com/OJ-lab/oj-lab-services/core/middleware"
 	"github.com/OJ-lab/oj-lab-services/service"
 	"github.com/gin-gonic/gin"
+	"github.com/sirupsen/logrus"
 )
 
 func SetupUserRouter(baseRoute *gin.RouterGroup) {
@@ -18,9 +22,33 @@ func SetupUserRouter(baseRoute *gin.RouterGroup) {
 		g.POST("/login", login)
 		g.GET("/me", middleware.HandleRequireLogin, me)
 		g.GET("/check-exist", checkUserExist)
+		g.GET("/stream", Stream)
 	}
 }
 
+// Stream
+//
+//	@Summary		Stream
+//	@Description	Stream
+//	@Tags			user
+//	@Router			/user/stream [get]
+//	@Accept			text/event-stream
+//	@Produce		text/event-stream
+//	@Success		200	{string}	string	"data: {message}"
+//	@Router			/user/stream [get]
+func Stream(ginCtx *gin.Context) {
+	ginCtx.Header("Content-Type", "text/event-stream")
+	ginCtx.Header("Cache-Control", "no-cache")
+
+	ginCtx.Stream(func(w io.Writer) bool {
+		// 每秒钟向客户端发送一条消息
+		logrus.Info("send message")
+		fmt.Fprintf(w, "data: %s\n\n", time.Now().String())
+		time.Sleep(1 * time.Second)
+		return true
+	})
+}
+
 type loginBody struct {
 	Account  string `json:"account" example:"admin"`
 	Password string `json:"password" example:"admin"`

config/development.toml

@@ -19,6 +19,9 @@ mode = "debug"
 swagger_on = true
 serve_front = false
 
+[rpc-server]
+port = 50051
+
 [judger]
 host = "http://localhost:8000"
 

config/production.toml

@@ -19,6 +19,9 @@ mode = "debug"
 swagger_on = true
 serve_front = true
 
+[rpc-server]
+port = 50051
+
 [judger]
 host = "http://host.docker.internal:8000"
 

service/proto/oj_lab_greeting.proto

@@ -0,0 +1,17 @@
+syntax = "proto3";
+
+package oj_lab_greeting.protos;
+
+option go_package = "github.com/OJ-lab/oj-lab-services/service/proto";
+
+service Greeter {
+    rpc Greeting (GreetingRequest) returns (GreetingResponse) {}
+}
+
+message GreetingRequest {
+    string name = 1;
+}
+
+message GreetingResponse {
+    string message = 1;
+}