Merge pull request #19 from yusank/dev

Merge dev into main

apps/gateway/cmd/main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"flag"
 
 	"github.com/gin-gonic/gin"
@@ -9,6 +10,7 @@ import (
 	"github.com/yusank/goim/apps/gateway/internal/app"
 	"github.com/yusank/goim/apps/gateway/internal/router"
 	"github.com/yusank/goim/apps/gateway/internal/service"
+	"github.com/yusank/goim/pkg/graceful"
 
 	"github.com/go-kratos/kratos/v2/log"
 )
@@ -39,5 +41,8 @@ func main() {
 		log.Info(err)
 	}
 
-	application.Stop()
+	graceful.Register(application.Shutdown)
+	if err = graceful.Shutdown(context.TODO()); err != nil {
+		log.Infof("graceful shutdown error: %v", err)
+	}
 }

apps/msg/cmd/main.go

@@ -1,13 +1,15 @@
 package main
 
 import (
+	"context"
 	"flag"
 
 	"github.com/go-kratos/kratos/v2/log"
 
 	messagev1 "github.com/yusank/goim/api/message/v1"
 	"github.com/yusank/goim/apps/msg/internal/app"
 	"github.com/yusank/goim/apps/msg/internal/service"
+	"github.com/yusank/goim/pkg/graceful"
 	"github.com/yusank/goim/pkg/mq"
 )
 
@@ -44,5 +46,8 @@ func main() {
 		log.Info(err)
 	}
 
-	application.Stop()
+	graceful.Register(application.Shutdown)
+	if err = graceful.Shutdown(context.TODO()); err != nil {
+		log.Infof("graceful shutdown error: %v", err)
+	}
 }

apps/msg/internal/service/msg_service.go

@@ -7,13 +7,12 @@ import (
 	"strings"
 	"sync"
 
-	redisv8 "github.com/go-redis/redis/v8"
-
 	"github.com/apache/rocketmq-client-go/v2/consumer"
 	"github.com/apache/rocketmq-client-go/v2/primitive"
 	"github.com/go-kratos/kratos/v2/log"
 	"github.com/go-kratos/kratos/v2/selector"
 	"github.com/go-kratos/kratos/v2/transport/grpc"
+	redisv8 "github.com/go-redis/redis/v8"
 	ggrpc "google.golang.org/grpc"
 
 	messagev1 "github.com/yusank/goim/api/message/v1"
@@ -61,41 +60,39 @@ func (s *MqMessageService) Consume(ctx context.Context, msg ...*primitive.Messag
 
 func (s *MqMessageService) handleSingleMsg(ctx context.Context, msg *primitive.MessageExt) error {
 	// PushMessageReq contains all MqMessage fields.
-	req := &messagev1.PushMessageReq{}
+	req := &messagev1.MqMessage{}
 	if err := json.Unmarshal(msg.Body, req); err != nil {
 		return err
 	}
 
+	in := &messagev1.PushMessageReq{
+		FromUser:        req.GetFromUser(),
+		ToUser:          req.GetToUser(),
+		PushMessageType: req.GetPushMessageType(),
+		ContentType:     req.GetContentType(),
+		Content:         req.GetContent(),
+		MsgSeq:          msg.MsgId,
+	}
+
 	if req.GetPushMessageType() == messagev1.PushMessageType_Broadcast {
-		return s.broadcast(ctx, req)
+		return s.broadcast(ctx, in)
 	}
 
-	var agentID string
 	str, err := s.rdb.Get(ctx, data.GetUserOnlineAgentKey(req.GetToUser())).Result()
 	if err != nil {
 		if err == redisv8.Nil {
 			log.Infof("user=%s not online, put to offline queue", req.GetToUser())
-			return s.putToRedis(ctx, msg, req)
+			return s.putToRedis(ctx, msg, in)
 		}
 		return err
 	}
-	agentID = str
 
-	cc, err := s.loadGrpcConn(ctx, agentID)
+	in.AgentId = str
+	cc, err := s.loadGrpcConn(ctx, in.AgentId)
 	if err != nil {
 		return err
 	}
 
-	in := &messagev1.PushMessageReq{
-		FromUser:        req.GetFromUser(),
-		ToUser:          req.GetToUser(),
-		PushMessageType: messagev1.PushMessageType_User,
-		ContentType:     req.GetContentType(),
-		Content:         req.GetContent(),
-		AgentId:         agentID,
-		MsgSeq:          msg.MsgId,
-	}
-
 	out, err := messagev1.NewPushMessagerClient(cc).PushMessage(ctx, in)
 	if err != nil {
 		log.Info("MSG send msg err=", err)

apps/push/cmd/main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"flag"
 
 	"github.com/gin-gonic/gin"
@@ -10,6 +11,7 @@ import (
 	"github.com/yusank/goim/apps/push/internal/app"
 	"github.com/yusank/goim/apps/push/internal/router"
 	"github.com/yusank/goim/apps/push/internal/service"
+	"github.com/yusank/goim/pkg/graceful"
 )
 
 var (
@@ -27,12 +29,21 @@ func main() {
 	if err != nil {
 		log.Fatal(err)
 	}
+
+	// register grpc
 	messagev1.RegisterPushMessagerServer(application.GrpcSrv, service.GetPushMessager())
+
+	// register router
 	g := gin.Default()
 	router.RegisterRouter(g.Group("/push/service"))
 	application.HTTPSrv.HandlePrefix("/", g)
 
 	if err = application.Run(); err != nil {
 		log.Fatal(err)
 	}
+
+	graceful.Register(application.Shutdown)
+	if err = graceful.Shutdown(context.TODO()); err != nil {
+		log.Infof("graceful shutdown error: %s", err)
+	}
 }

apps/push/internal/service/push_message.go

@@ -11,6 +11,7 @@ import (
 	messagev1 "github.com/yusank/goim/api/message/v1"
 	"github.com/yusank/goim/pkg/conn/pool"
 	"github.com/yusank/goim/pkg/conn/wrapper"
+	"github.com/yusank/goim/pkg/graceful"
 	"github.com/yusank/goim/pkg/worker"
 )
 
@@ -28,6 +29,7 @@ func GetPushMessager() *PushMessager {
 	pmOnce.Do(func() {
 		pm = new(PushMessager)
 		pm.workerPool = worker.NewPool(100, 20)
+		graceful.Register(pm.workerPool.Shutdown)
 	})
 
 	return pm

docs/content/docs/advance/configuration.md

@@ -1,3 +1,80 @@
+---
+weight: 1
+---
+
 # Configuration
 
-> config
+配置为两份文件分别为 service config 和 registry config
+
+- service config 关注服务启停以及声明周期中需要的各类配置
+- registry config 关注服务注册相关配置
+
+## server config definition
+
+```proto
+// Service 为一个服务的全部配置
+message Service {
+    string name = 1;
+    string version = 2;
+    optional Server http = 3;
+    optional Server grpc = 4;
+    Log log = 5;
+    map<string, string> metadata = 6;
+    Redis redis = 7;
+    MQ mq = 8;
+}
+
+message Server {
+    string scheme = 1;
+    string addr = 2;
+    int32 port = 3;
+}
+
+
+enum Level {
+    DEBUG = 0;
+    INFO = 1;
+    WARING = 2;
+    ERROR = 3;
+    FATAL = 4;
+}
+
+message Log {
+    optional string log_path = 1;
+    repeated Level level = 2;
+}
+
+message Redis {
+    string addr = 1;
+    string password = 2;
+    int32 max_conns = 3;
+    int32 min_idle_conns = 4;
+    google.protobuf.Duration dial_timeout = 5;
+    google.protobuf.Duration idle_timeout = 6;
+}
+
+message MQ {
+    repeated string addr = 1;
+    int32 max_retry = 2;
+}
+```
+
+## registry config definition 
+
+```proto
+message RegistryInfo {
+    repeated string addr = 1;
+    string scheme = 2;
+    google.protobuf.Duration dial_timeout_sec = 3;
+    google.protobuf.Duration dial_keep_alive_time_sec = 4;
+    google.protobuf.Duration dial_keep_alive_timeout_sec = 5;
+}
+
+message Registry {
+    string name = 1;
+    oneof reg {
+        RegistryInfo consul = 2;
+        RegistryInfo etcd = 3;
+    }
+}
+```

docs/content/docs/example/terminal_cli/index.md

@@ -0,0 +1,22 @@
+---
+title: "Terminal CLI"
+---
+
+## How to run it
+
+服务提供一个简单的终端 GUI 可以测试消息的发送和接受，代码在 `tests` 目录下。
+
+在 `goim/test` 目录下执行如下命令：
+
+```shell
+# 支持参数
+# ADDR ?= 127.0.0.1:18071
+# UID ?= user1
+# TOUID ?= user2
+
+make run-gui UID=user3 TOUID=user2
+```
+
+界面如下：
+
+![gui](./gui.png)

docs/content/docs/quick_start/prepare.md

@@ -2,10 +2,43 @@
 weight: 2
 title: "Prepare"
 ---
-# Prepare
 
-## install mq
+## requirement
 
-## install consul
+### environment
 
-## install redis
+- apache/rocketmq 4.6.0+
+- consul 1.11.4+
+- redis 2.0+
+
+> 关于部署 rocketmq：docker 部署 rocketmq 过程中遇到过一些问题，如果你有疑问可以参考这篇文章 [Docker 部署 RocketMQ](https://yusank.space/posts/rocketmq-deploy/)
+
+### config
+
+msg service 为例，`apps/msg/config/config.yaml`:
+
+    name: goim.msg.service
+    version: v0.0.0
+    grpc:
+    scheme: grpc
+    port: 18063
+    log:
+    level:
+    - INFO
+    - DEBUG
+    metadata:
+    grpcSrv: yes
+    redis:
+    addr: 127.0.0.1:6379
+    mq:
+    addr:
+        - 127.0.0.1:9876
+
+`apps/msg/config/registry.yaml` :
+
+    consul:
+    addr:
+        - 127.0.0.1:8500
+    scheme: http
+
+根据自己的环境去修改各个组件的地址和端口。

docs/content/docs/quick_start/quick_start.md

@@ -4,14 +4,43 @@ weight: 1
 
 # Quick Start
 
-## build
+## run
 
 ```shell
-$ make build all
+# run msg service
+$ make run Srv=msg
+# run gateway service
+$ make run Srv=gateway
+# run push service
+$ make run Srv=push
 ```
 
-## run
+## other make command
 
 ```shell
-$ make run Srv=xxx
+make help
+
+Usage:
+  make <target>
+
+Development
+  vet              Run go vet against code.
+  lint             Run go lint against code.
+  test             Run test against code.
+
+Generate
+  protoc           Run protoc command to generate pb code.
+
+Build
+  build            build provided server
+  build-all        build all apps
+
+Docker
+  docker-build     build docker image
+
+Run
+  run              run provided server
+
+General
+  help             Display this help.
 ```