feat: add openim info test

Signed-off-by: Xinwei Xiong(cubxxw) <3293172751nss@gmail.com>

.docker-compose_cfg/config.yaml

@@ -0,0 +1,79 @@
+# Copyright © 2023 OpenIM. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+#---------------Infrastructure configuration---------------------#
+etcd:
+  etcdSchema: openim  #默认即可
+  etcdAddr: [ 127.0.0.1:2379 ] #单机部署时，默认即可
+  userName:
+  password:
+  secret: openIM123
+
+mysql:
+  dbMysqlDatabaseName: admin_chat # 数据库名字 默认即可
+
+# 默认管理员账号
+admin:
+  defaultAccount:
+    account: [ "admin1", "admin2" ]
+    defaultPassword: [ "password1", "password2" ]
+    openIMUserID: [ "openIM123456", "openIMAdmin" ]
+    faceURL: [ "", "" ]
+    nickname: [ "admin1", "admin2" ]
+    level: [ 1, 100 ]
+
+
+adminapi:
+  openImAdminApiPort: [ 10009 ] #管理后台api服务端口，默认即可，需要开放此端口或做nginx转发
+  listenIP: 0.0.0.0
+
+chatapi:
+  openImChatApiPort: [ 10008 ] #登录注册，默认即可，需要开放此端口或做nginx转发
+  listenIP: 0.0.0.0
+
+rpcport: # rpc服务端口 默认即可
+  openImAdminPort: [ 30200 ]
+  openImChatPort: [ 30300 ]
+
+
+rpcregistername: #rpc注册服务名，默认即可
+  openImChatName: Chat
+  openImAdminCMSName: Admin
+
+chat:
+  codeTTL: 300 #短信验证码有效时间(秒)
+  superVerificationCode: 666666 # 超级验证码
+  alismsverify: #阿里云短信配置，在阿里云申请成功后修改以下四项
+    accessKeyId:
+    accessKeySecret:
+    signName:
+    verificationCodeTemplateCode:
+
+
+oss:
+  tempDir: enterprise-temp # 临时密钥上传的目录
+  dataDir: enterprise-data # 最终存放目录
+  aliyun:
+    endpoint: https://oss-cn-chengdu.aliyuncs.com
+    accessKeyID: ""
+    accessKeySecret: ""
+    bucket: ""
+  tencent:
+    BucketURL: ""
+    serviceURL: https://cos.COS_REGION.myqcloud.com
+    secretID: ""
+    secretKey: ""
+    sessionToken: ""
+    bucket: ""
+  use: "minio"

.docker-compose_cfg/datasource-compose.yaml

@@ -0,0 +1,27 @@
+# Copyright © 2023 OpenIM. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+#more datasource-compose.yaml 
+apiVersion: 1
+
+datasources:
+- name: Prometheus
+  type: prometheus
+  access: proxy
+  orgId: 1
+  url: http://127.0.0.1:9091
+  basicAuth: false
+  isDefault: true
+  version: 1
+  editable: true

.docker-compose_cfg/prometheus-compose.yml

@@ -0,0 +1,85 @@
+# Copyright © 2023 OpenIM. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+#more prometheus-compose.yml 
+global:
+  scrape_interval:     15s 
+  evaluation_interval: 15s 
+  external_labels:
+    monitor: 'openIM-monitor'
+
+scrape_configs:
+    - job_name: 'prometheus'
+      static_configs:
+      - targets: ['localhost:9091']
+
+    - job_name: 'openIM-server'
+      metrics_path: /metrics
+      static_configs:
+      - targets: ['localhost:10002']
+        labels:
+          group: 'api'
+
+      - targets: ['localhost:20110']
+        labels:
+          group: 'user'
+
+      - targets: ['localhost:20120']
+        labels:
+          group: 'friend'
+
+      - targets: ['localhost:20130']
+        labels:
+          group: 'message'
+
+      - targets: ['localhost:20140']
+        labels:
+          group: 'msg-gateway'    
+
+      - targets: ['localhost:20150']
+        labels:
+          group: 'group'
+
+      - targets: ['localhost:20160']
+        labels:
+          group: 'auth'
+
+      - targets: ['localhost:20170']
+        labels:
+          group: 'push'
+
+      - targets: ['localhost:20120']
+        labels:
+          group: 'friend'
+
+      
+      - targets: ['localhost:20230']
+        labels:
+          group: 'conversation' 
+
+      
+      - targets: ['localhost:21400', 'localhost:21401', 'localhost:21402', 'localhost:21403']
+        labels:
+          group: 'msg-transfer'
+       
+ 
+
+
+      
+
+    - job_name: 'node'  
+      scrape_interval: 8s
+      static_configs:
+      - targets: ['localhost:9100']
+     

.env

@@ -29,8 +29,8 @@ PASSWORD=openIM123
 MINIO_ENDPOINT=http://172.28.0.1:10005
 
 # Base URL for the application programming interface (API).
-# Default: API_URL=http://172.28.0.1:10002
-API_URL=http://172.28.0.1:10002
+# Default: API_URL=http://172.0.0.1:10002
+API_URL=http://172.0.0.1:10002
 
 # Directory path for storing data files or related information.
 # Default: DATA_DIR=./
@@ -98,14 +98,7 @@ PROMETHEUS_NETWORK_ADDRESS=172.28.0.11
 # Address or hostname for the Grafana network.
 # Default: GRAFANA_NETWORK_ADDRESS=172.28.0.12
 GRAFANA_NETWORK_ADDRESS=172.28.0.12
-  
-# Address or hostname for the node_exporter network.
-# Default: NODE_EXPORTER_NETWORK_ADDRESS=172.28.0.13
-NODE_EXPORTER_NETWORK_ADDRESS=172.28.0.13
-  
-# Address or hostname for the OpenIM admin network.
-# Default: OPENIM_ADMIN_NETWORK_ADDRESS=172.28.0.14
-OPENIM_ADMIN_FRONT_NETWORK_ADDRESS=172.28.0.14
+
 
 # ===============================================
 # = Component Extension Configuration =
@@ -290,19 +283,3 @@ SERVER_BRANCH=main
 # Port for the OpenIM admin API.
 # Default: OPENIM_ADMIN_API_PORT=10009
 OPENIM_ADMIN_API_PORT=10009
-  
-# Port for the node exporter.
-# Default: NODE_EXPORTER_PORT=19100
-NODE_EXPORTER_PORT=19100
-  
-# Port for the prometheus.
-# Default: PROMETHEUS_PORT=19090
-PROMETHEUS_PORT=19090
-  
-# Port for the grafana.
-# Default: GRAFANA_PORT=3000
-GRAFANA_PORT=3000
-  
-# Port for the admin front.
-# Default: OPENIM_ADMIN_FRONT_PORT=11002
-OPENIM_ADMIN_FRONT_PORT=11002

.github/.codecov.yml

@@ -20,20 +20,4 @@ coverage:
         paths:
           - pkg/* # only include coverage in "pkg/" folder
         informational: true # Always pass check
-      tools: # declare a new status context "tools"
-        paths:
-          - tools/* # only include coverage in "tools/" folder
-        informational: true # Always pass check
-      test: # declare a new status context "test"
-        paths:
-          - test/* # only include coverage in "test/" folder
-        informational: true # Always pass check
-      # internal: # declare a new status context "internal"
-      #   paths:
-      #     - internal/* # only include coverage in "internal/" folder
-      #   informational: true # Always pass check
-      # cmd: # declare a new status context "cmd"
-      #   paths:
-      #     - cmd/* # only include coverage in "cmd/" folder
-      #   informational: true # Always pass check
-    patch: off # disable the commit only checks
+    patch: off # disable the commit only checks

.github/workflows/api-test.yml

@@ -0,0 +1,90 @@
+# Copyright © 2023 OpenIM. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+name: OpenIM API TEST
+
+on: 
+  push:
+    branches:
+        - main
+    paths-ignore:
+        - "docs/**"
+        - "README.md"
+        - "README_zh-CN.md"
+        - "CONTRIBUTING.md"
+  pull_request:
+    branches:
+        - main
+    paths-ignore:
+        - "README.md"
+        - "README_zh-CN.md"
+        - "CONTRIBUTING.md"
+        - "docs/**"
+
+env:
+  GO_VERSION: "1.19"
+  GOLANGCI_VERSION: "v1.50.1"
+
+jobs:
+  execute-linux-systemd-scripts:
+    name: Execute OpenIM script on ${{ matrix.os }}
+    runs-on: ${{ matrix.os }}
+    environment:
+      name: openim
+    strategy:
+      matrix:
+        go_version: ["1.20"]
+        os: ["ubuntu-latest"]
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@v4
+
+    - name: Set up Go ${{ matrix.go_version }}
+      uses: actions/setup-go@v4
+      with:
+        go-version: ${{ matrix.go_version }}
+      id: go
+
+    - name: Install Task
+      uses: arduino/setup-task@v1
+      with:
+        version: '3.x' # If available, use the latest major version that's compatible
+        repo-token: ${{ secrets.GITHUB_TOKEN }}
+
+    - name: Docker Operations
+      run: |
+        curl -o docker-compose.yml https://raw.githubusercontent.com/OpenIMSDK/openim-docker/main/example/basic-openim-server-dependency.yml
+        sudo docker compose up -d
+        sudo sleep 60
+
+    - name: Module Operations
+      run: |
+        sudo make tidy
+        sudo make tools.verify.go-gitlint
+
+    - name: Build, Start, Check Services and Print Logs
+      run: |
+        sudo ./scripts/install/install.sh -i && \
+        sudo ./scripts/install/install.sh -s && \
+        (echo "An error occurred, printing logs:" && sudo cat ./_output/logs/* 2>/dev/null)
+
+    - name: Run Test
+      run: |
+        sudo make test-api && \
+        (echo "An error occurred, printing logs:" && sudo cat ./_output/logs/* 2>/dev/null)
+
+    - name: Stop Services
+      run: |
+        sudo ./scripts/install/install.sh -u && \
+        (echo "An error occurred, printing logs:" && sudo cat ./_output/logs/* 2>/dev/null)

.github/workflows/build-docker-image.yml

@@ -52,7 +52,6 @@ jobs:
             type=ref,event=branch
             type=ref,event=pr
             type=semver,pattern={{version}}
-            type=semver,pattern=v{{version}}
             type=semver,pattern={{major}}.{{minor}}
             type=semver,pattern={{major}}
             type=sha
@@ -88,16 +87,6 @@ jobs:
         uses: docker/metadata-action@v5.0.0
         with:
           images: registry.cn-hangzhou.aliyuncs.com/openimsdk/openim-server
-          # generate Docker tags based on the following events/attributes
-          tags: |
-            type=schedule
-            type=ref,event=branch
-            type=ref,event=pr
-            type=semver,pattern={{version}}
-            type=semver,pattern=v{{version}}
-            type=semver,pattern={{major}}.{{minor}}
-            type=semver,pattern={{major}}
-            type=sha
 
       - name: Log in to AliYun Docker Hub
         uses: docker/login-action@v3
@@ -137,7 +126,6 @@ jobs:
             type=ref,event=branch
             type=ref,event=pr
             type=semver,pattern={{version}}
-            type=semver,pattern=v{{version}}
             type=semver,pattern={{major}}.{{minor}}
             type=semver,pattern={{major}}
             type=sha

.github/workflows/build-openim-web-image.yml

@@ -15,11 +15,11 @@
 name: Build OpenIM Web Docker image
 
 on:
-#   schedule:
-#   - cron: '30 3 * * *'
+  schedule:
+  - cron: '30 3 * * *'
   push:
     branches:
-    #   - main
+      - main
       - release-*
     tags:
       - v*

.github/workflows/e2e-test.yml

@@ -79,33 +79,10 @@ jobs:
         sudo make tidy
         sudo make tools.verify.go-gitlint
 
-    - name: Build, Start
+    - name: Build, Start, Test, Check Services and Print Logs
       run: |
-        sudo ./scripts/install/install.sh -i
-
-    - name: Exec OpenIM System Status Chack
-      run: |
-        sudo ./scripts/install/install.sh -s
-
-    - name: Exec OpenIM API test
-      run: |
-        sudo make test-api
-    
-    - name: Exec OpenIM E2E test
-      run: |
-        sudo make test-e2e
-
-    - name: Exec OpenIM System uninstall
-      run: |
-        sudo ./scripts/install/install.sh -u
-
-    - name: gobenchdata publish
-      uses: bobheadxi/gobenchdata@v1
-      with:
-        PRUNE_COUNT: 30
-        GO_TEST_FLAGS: -cpu 1,2
-        PUBLISH: true
-        PUBLISH_BRANCH: gh-pages
-      env:
-        GITHUB_TOKEN: ${{ secrets.BOT_GITHUB_TOKEN }}
-      continue-on-error: true
+        sudo ./scripts/install/install.sh -i && \
+        sudo make test-api && \
+        sudo ./scripts/install/install.sh -s && \
+        sudo ./scripts/install/install.sh -u || \
+        (echo "An error occurred, printing logs:" && sudo cat ./_output/logs/* 2>/dev/null)

.github/workflows/link-pr.yml

@@ -41,7 +41,7 @@ jobs:
             # ./*.md all markdown files in the root directory
           args: --verbose  -E -i --no-progress --exclude-path './CHANGELOG' './**/*.md'
         env:
-          GITHUB_TOKEN: ${{secrets.BOT_GITHUB_TOKEN}}
+          GITHUB_TOKEN: ${{secrets.GH_PAT}}
 
       - name: Create Issue From File
         if: env.lychee_exit_code != 0

Makefile

@@ -171,10 +171,6 @@ cover:
 test-api:
 	@$(MAKE) go.test.api
 
-## test-e2e: Run e2e test
-test-e2e:
-	@$(MAKE) go.test.e2e
-
 ## updates: Check for updates to go.mod dependencies. ✨
 .PHONY: updates
 	@$(MAKE) go.updates

README-zh_CN.md

@@ -29,10 +29,6 @@
 
 </p>
 
-## 🟢 扫描微信进群交流
-<img src="./docs/images/Wechat.jpg" width="300">
-
-
 ## Ⓜ️ 关于 OpenIM
 
 OpenIM 不仅仅是一个开源的即时消息组件，它是你的应用程序生态系统的一个不可或缺的部分。查看下面的图表，了解 AppServer、AppClient、OpenIMServer 和 OpenIMSDK 是如何交互的。

README.md

@@ -231,7 +231,7 @@ Before you start, please make sure your changes are in demand. The best for that
 - [OpenIM Makefile Utilities](https://github.com/openimsdk/open-im-server/tree/main/docs/contrib/util-makefile.md)
 - [OpenIM Script Utilities](https://github.com/openimsdk/open-im-server/tree/main/docs/contrib/util-scripts.md)
 - [OpenIM Versioning](https://github.com/openimsdk/open-im-server/tree/main/docs/contrib/version.md)
-- [Manage backend and monitor deployment](https://github.com/openimsdk/open-im-server/tree/main/docs/contrib/prometheus-grafana.md)
+
 
 ## :busts_in_silhouette: Community
 

cmd/openim-api/main.go

@@ -17,12 +17,13 @@ package main
 import (
 	"context"
 	"fmt"
+	ginProm "github.com/openimsdk/open-im-server/v3/pkg/common/ginPrometheus"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/prom_metrics"
 	"net"
 	_ "net/http/pprof"
 	"strconv"
 
-	ginProm "github.com/openimsdk/open-im-server/v3/pkg/common/ginprometheus"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/discovery_register"
 
 	"github.com/OpenIMSDK/protocol/constant"
 	"github.com/OpenIMSDK/tools/discoveryregistry"
@@ -32,7 +33,6 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/cmd"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
-	kdisc "github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister"
 )
 
 func main() {
@@ -65,7 +65,7 @@ func run(port int, proPort int) error {
 	var client discoveryregistry.SvcDiscoveryRegistry
 
 	// Determine whether zk is passed according to whether it is a clustered deployment
-	client, err = kdisc.NewDiscoveryRegister(config.Config.Envs.Discovery)
+	client, err = discovery_register.NewDiscoveryRegister(config.Config.Envs.Discovery)
 	if err != nil {
 		log.ZError(context.Background(), "Failed to initialize discovery register", err)
 
@@ -86,7 +86,7 @@ func run(port int, proPort int) error {
 	router := api.NewGinRouter(client, rdb)
 	//////////////////////////////
 	if config.Config.Prometheus.Enable {
-		p := ginProm.NewPrometheus("app", prommetrics.GetGinCusMetrics("Api"))
+		p := ginProm.NewPrometheus("app", prom_metrics.GetGinCusMetrics("Api"))
 		p.SetListenAddress(fmt.Sprintf(":%d", proPort))
 		p.Use(router)
 	}

config/alertmanager.yml

@@ -1,32 +0,0 @@
-###################### AlertManager Configuration ######################
-# AlertManager configuration using environment variables
-#
-# Resolve timeout
-# SMTP configuration for sending alerts
-# Templates for email notifications
-# Routing configurations for alerts
-# Receiver configurations
-global:
-  resolve_timeout: 5m
-  smtp_from: alert@openim.io
-  smtp_smarthost: smtp.163.com:465
-  smtp_auth_username: alert@openim.io
-  smtp_auth_password: YOURAUTHPASSWORD
-  smtp_require_tls: false
-  smtp_hello: xxx监控告警
-
-templates:
-  - /etc/alertmanager/email.tmpl
-
-route:
-  group_wait: 5s
-  group_interval: 5s
-  repeat_interval: 5m
-  receiver: email
-receivers:
-  - name: email
-    email_configs:
-      - to: {EMAIL_TO:-'alert@example.com'}
-        html: '{{ template "email.to.html" . }}'
-        headers: { Subject: "[OPENIM-SERVER]Alarm" }
-        send_resolved: true

config/config.yaml

@@ -198,7 +198,7 @@ rpcRegisterName:
 # Whether to output in json format
 # Whether to include stack trace in logs
 log:
-  storageLocation: ../logs/
+  storageLocation: ./logs/
   rotationTime: 24
   remainRotationCount: 2
   remainLogLevel: 6
@@ -382,7 +382,7 @@ callback:
 # The number of Prometheus ports per service needs to correspond to rpcPort
 # The number of ports needs to be consistent with msg_transfer_service_num in script/path_info.sh
 prometheus:
-  enable: false
+  enable: true
   prometheusUrl: "https://openim.prometheus"
   apiPrometheusPort: [20100]
   userPrometheusPort: [ 20110 ]

config/email.tmpl

@@ -1,16 +0,0 @@
-{{ define "email.to.html" }}
-{{ range .Alerts }}
-<!-- Begin of OpenIM Alert -->
-<div style="border:1px solid #ccc; padding:10px; margin-bottom:10px;">
-    <h3>OpenIM Alert</h3>
-    <p><strong>Alert Program:</strong> Prometheus Alert</p>
-    <p><strong>Severity Level:</strong> {{ .Labels.severity }}</p>
-    <p><strong>Alert Type:</strong> {{ .Labels.alertname }}</p>
-    <p><strong>Affected Host:</strong> {{ .Labels.instance }}</p>
-    <p><strong>Affected Service:</strong> {{ .Labels.job }}</p>
-    <p><strong>Alert Subject:</strong> {{ .Annotations.summary }}</p>
-    <p><strong>Trigger Time:</strong> {{ .StartsAt.Format "2006-01-02 15:04:05" }}</p>
-</div>
-<!-- End of OpenIM Alert -->
-{{ end }}
-{{ end }}

config/instance-down-rules.yml

@@ -1,11 +0,0 @@
-groups:
-  - name: instance_down
-    rules:
-      - alert: InstanceDown
-        expr: up == 0
-        for: 1m
-        labels:
-          severity: critical
-        annotations:
-          summary: "Instance {{ $labels.instance }} down"
-          description: "{{ $labels.instance }} of job {{ $labels.job }} has been down for more than 1 minutes."

config/prometheus.yml

@@ -1,85 +0,0 @@
-# my global config
-global:
-  scrape_interval:     15s # Set the scrape interval to every 15 seconds. Default is every 1 minute.
-  evaluation_interval: 15s # Evaluate rules every 15 seconds. The default is every 1 minute.
-  # scrape_timeout is set to the global default (10s).
-
-# Alertmanager configuration
-alerting:
-  alertmanagers:
-    - static_configs:
-        - targets: ['172.28.0.1:19093']
-
-# Load rules once and periodically evaluate them according to the global 'evaluation_interval'.
-rule_files:
-  - "instance-down-rules.yml"
-# - "first_rules.yml"
-# - "second_rules.yml"
-
-# A scrape configuration containing exactly one endpoint to scrape:
-# Here it's Prometheus itself.
-scrape_configs:
-  # The job name is added as a label "job='job_name'"" to any timeseries scraped from this config.
-  # Monitored information captured by prometheus
-  - job_name: 'node-exporter'
-    static_configs:
-      - targets: [ '172.28.0.1:19100' ]
-        labels:
-          namespace: 'default'
-
-  # prometheus fetches application services
-  - job_name: 'openimserver-openim-api'
-    static_configs:
-      - targets: [ '172.28.0.1:20100' ]
-        labels:
-          namespace: 'default'
-  - job_name: 'openimserver-openim-msggateway'
-    static_configs:
-      - targets: [ '172.28.0.1:20140' ]
-        labels:
-          namespace: 'default'
-  - job_name: 'openimserver-openim-msgtransfer'
-    static_configs:
-      - targets: [ 172.28.0.1:21400, 172.28.0.1:21401, 172.28.0.1:21402, 172.28.0.1:21403 ]
-        labels:
-          namespace: 'default'
-  - job_name: 'openimserver-openim-push'
-    static_configs:
-      - targets: [ '172.28.0.1:20170' ]
-        labels:
-          namespace: 'default'
-  - job_name: 'openimserver-openim-rpc-auth'
-    static_configs:
-      - targets: [ '172.28.0.1:20160' ]
-        labels:
-          namespace: 'default'
-  - job_name: 'openimserver-openim-rpc-conversation'
-    static_configs:
-      - targets: [ '172.28.0.1:20230' ]
-        labels:
-          namespace: 'default'
-  - job_name: 'openimserver-openim-rpc-friend'
-    static_configs:
-      - targets: [ '172.28.0.1:20120' ]
-        labels:
-          namespace: 'default'
-  - job_name: 'openimserver-openim-rpc-group'
-    static_configs:
-      - targets: [ '172.28.0.1:20150' ]
-        labels:
-          namespace: 'default'
-  - job_name: 'openimserver-openim-rpc-msg'
-    static_configs:
-      - targets: [ '172.28.0.1:20130' ]
-        labels:
-          namespace: 'default'
-  - job_name: 'openimserver-openim-rpc-third'
-    static_configs:
-      - targets: [ '172.28.0.1:21301' ]
-        labels:
-          namespace: 'default'
-  - job_name: 'openimserver-openim-rpc-user'
-    static_configs:
-      - targets: [ '172.28.0.1:20110' ]
-        labels:
-          namespace: 'default'

deployments/README.md

@@ -123,30 +123,6 @@ Explore our Helm-Charts repository and read through: [Helm-Charts Repository](ht
 
 Using the helm charts repository, you can ignore the following configuration, but if you want to just use the server and scale on top of it, you can go ahead:
 
-**Use the Helm template to generate the deployment yaml file: `openim-charts.yaml`**
-
-**Gen Image:**
-
-```bash
-../scripts/genconfig.sh ../scripts/install/environment.sh ./templates/helm-image.yaml > ./charts/generated-configs/helm-image.yaml
-```
-
-**Gen Charts:**
-
-```bash
-for chart in ./charts/*/; do
-    if [[ "$chart" == *"generated-configs"* || "$chart" == *"helmfile.yaml"* ]]; then
-        continue
-    fi
-
-    if [ -f "${chart}values.yaml" ]; then
-        helm template "$chart" -f "./charts/generated-configs/helm-image.yaml" -f "./charts/generated-configs/config.yaml" -f "./charts/generated-configs/notification.yaml" >> openim-charts.yaml
-    else
-        helm template "$chart" >> openim-charts.yaml
-    fi
-done
-```
-
 **Use Helmfile:**
 
 ```bash

deployments/charts/openim-api/templates/deployment.yaml

@@ -48,10 +48,10 @@ spec:
           resources:
             {{- toYaml .Values.resources | nindent 12 }}
           volumeMounts:
-            - mountPath: /openim/openim-server/config/config.yaml
+            - mountPath: /config/config.yaml
               name: config
               subPath: config.yaml
-            - mountPath: /openim/openim-server/config/
+            - mountPath: /config/notification.yaml
               name: config
               subPath: notification.yaml
       volumes:

deployments/charts/openim-msggateway/templates/deployment.yaml

@@ -65,10 +65,10 @@ spec:
           resources:
             {{- toYaml .Values.resources | nindent 12 }}
           volumeMounts:
-            - mountPath: /openim/openim-server/config/config.yaml
+            - mountPath: /config/config.yaml
               name: config
               subPath: config.yaml
-            - mountPath: /openim/openim-server/config/
+            - mountPath: /config/notification.yaml
               name: config
               subPath: notification.yaml
       volumes:

deployments/charts/openim-msgtransfer/templates/deployment.yaml

@@ -48,10 +48,10 @@ spec:
           resources:
             {{- toYaml .Values.resources | nindent 12 }}
           volumeMounts:
-            - mountPath: /openim/openim-server/config/config.yaml
+            - mountPath: /config/config.yaml
               name: config
               subPath: config.yaml
-            - mountPath: /openim/openim-server/config/
+            - mountPath: /config/notification.yaml
               name: config
               subPath: notification.yaml
       volumes:

deployments/charts/openim-push/templates/deployment.yaml

@@ -48,10 +48,10 @@ spec:
           resources:
             {{- toYaml .Values.resources | nindent 12 }}
           volumeMounts:
-            - mountPath: /openim/openim-server/config/config.yaml
+            - mountPath: /config/config.yaml
               name: config
               subPath: config.yaml
-            - mountPath: /openim/openim-server/config/
+            - mountPath: /config/notification.yaml
               name: config
               subPath: notification.yaml
       volumes:

deployments/charts/openim-rpc-auth/templates/deployment.yaml

@@ -48,10 +48,10 @@ spec:
           resources:
             {{- toYaml .Values.resources | nindent 12 }}
           volumeMounts:
-            - mountPath: /openim/openim-server/config/config.yaml
+            - mountPath: /config/config.yaml
               name: config
               subPath: config.yaml
-            - mountPath: /openim/openim-server/config/
+            - mountPath: /config/notification.yaml
               name: config
               subPath: notification.yaml
       volumes:

deployments/charts/openim-rpc-conversation/templates/deployment.yaml

@@ -48,10 +48,10 @@ spec:
           resources:
             {{- toYaml .Values.resources | nindent 12 }}
           volumeMounts:
-            - mountPath: /openim/openim-server/config/config.yaml
+            - mountPath: /config/config.yaml
               name: config
               subPath: config.yaml
-            - mountPath: /openim/openim-server/config/
+            - mountPath: /config/notification.yaml
               name: config
               subPath: notification.yaml
       volumes:

deployments/charts/openim-rpc-friend/templates/deployment.yaml

@@ -48,10 +48,10 @@ spec:
           resources:
             {{- toYaml .Values.resources | nindent 12 }}
           volumeMounts:
-            - mountPath: /openim/openim-server/config/config.yaml
+            - mountPath: /config/config.yaml
               name: config
               subPath: config.yaml
-            - mountPath: /openim/openim-server/config/
+            - mountPath: /config/notification.yaml
               name: config
               subPath: notification.yaml
       volumes:

deployments/charts/openim-rpc-group/templates/deployment.yaml

@@ -48,10 +48,10 @@ spec:
           resources:
             {{- toYaml .Values.resources | nindent 12 }}
           volumeMounts:
-            - mountPath: /openim/openim-server/config/config.yaml
+            - mountPath: /config/config.yaml
               name: config
               subPath: config.yaml
-            - mountPath: /openim/openim-server/config/
+            - mountPath: /config/notification.yaml
               name: config
               subPath: notification.yaml
       volumes:

deployments/charts/openim-rpc-msg/templates/deployment.yaml

@@ -48,10 +48,10 @@ spec:
           resources:
             {{- toYaml .Values.resources | nindent 12 }}
           volumeMounts:
-            - mountPath: /openim/openim-server/config/config.yaml
+            - mountPath: /config/config.yaml
               name: config
               subPath: config.yaml
-            - mountPath: /openim/openim-server/config/
+            - mountPath: /config/notification.yaml
               name: config
               subPath: notification.yaml
       volumes:

deployments/charts/openim-rpc-third/templates/deployment.yaml

@@ -48,10 +48,10 @@ spec:
           resources:
             {{- toYaml .Values.resources | nindent 12 }}
           volumeMounts:
-            - mountPath: /openim/openim-server/config/config.yaml
+            - mountPath: /config/config.yaml
               name: config
               subPath: config.yaml
-            - mountPath: /openim/openim-server/config/
+            - mountPath: /config/notification.yaml
               name: config
               subPath: notification.yaml
       volumes:

deployments/charts/openim-rpc-user/templates/deployment.yaml

@@ -48,10 +48,10 @@ spec:
           resources:
             {{- toYaml .Values.resources | nindent 12 }}
           volumeMounts:
-            - mountPath: /openim/openim-server/config/config.yaml
+            - mountPath: /config/config.yaml
               name: config
               subPath: config.yaml
-            - mountPath: /openim/openim-server/config/
+            - mountPath: /config/notification.yaml
               name: config
               subPath: notification.yaml
       volumes:

deployments/templates/alertmanager.yml

@@ -1,33 +0,0 @@
-###################### AlertManager Configuration ######################
-# AlertManager configuration using environment variables
-#
-# Resolve timeout
-# SMTP configuration for sending alerts
-# Templates for email notifications
-# Routing configurations for alerts
-# Receiver configurations
-global:
-  resolve_timeout: ${ALERTMANAGER_RESOLVE_TIMEOUT}
-  smtp_from: ${ALERTMANAGER_SMTP_FROM}
-  smtp_smarthost: ${ALERTMANAGER_SMTP_SMARTHOST}
-  smtp_auth_username: ${ALERTMANAGER_SMTP_AUTH_USERNAME}
-  smtp_auth_password: ${ALERTMANAGER_SMTP_AUTH_PASSWORD}
-  smtp_require_tls: ${ALERTMANAGER_SMTP_REQUIRE_TLS}
-  smtp_hello: ${ALERTMANAGER_SMTP_HELLO}
-
-templates:
-  - /etc/alertmanager/email.tmpl
-
-route:
-  group_by: ['alertname']
-  group_wait: 5s
-  group_interval: 5s
-  repeat_interval: 5m
-  receiver: email
-receivers:
-  - name: email
-    email_configs:
-      - to: ${ALERTMANAGER_EMAIL_TO}
-        html: '{{ template "email.to.html" . }}'
-        headers: { Subject: "[OPENIM-SERVER]Alarm" }
-        send_resolved: true

deployments/templates/chat.yaml

@@ -96,14 +96,7 @@ verifyCode:
     accessKeySecret: ""
     signName: ""
     verificationCodeTemplateCode: ""
-  mail: # 根据对应的发件邮箱更改 sendMail、senderAuthorizationCode、smtpAddr、smtpPort 即可
-    title: ""
-    senderMail: "" # 发送者
-    senderAuthorizationCode: "" # 授权码
-    smtpAddr: "smtp.qq.com"  # smtp 服务器地址
-    smtpPort: 25   # smtp 服务器邮件发送端口
-  testDepartMentID: 001
-  imAPIURL: http://127.0.0.1:10002
+
 
 ###################### Proxy Header ######################
 # 获取ip的header,没有配置直接获取远程地址

deployments/templates/env_template.yaml

@@ -94,22 +94,12 @@ OPENIM_CHAT_NETWORK_ADDRESS=${OPENIM_CHAT_NETWORK_ADDRESS}
 # Address or hostname for the Prometheus network.
 # Default: PROMETHEUS_NETWORK_ADDRESS=172.28.0.11
 PROMETHEUS_NETWORK_ADDRESS=${PROMETHEUS_NETWORK_ADDRESS}
-  
+
 # Address or hostname for the Grafana network.
 # Default: GRAFANA_NETWORK_ADDRESS=172.28.0.12
 GRAFANA_NETWORK_ADDRESS=${GRAFANA_NETWORK_ADDRESS}
-  
-# Address or hostname for the node_exporter network.
-# Default: NODE_EXPORTER_NETWORK_ADDRESS=172.28.0.13
-NODE_EXPORTER_NETWORK_ADDRESS=${NODE_EXPORTER_NETWORK_ADDRESS}
-  
-# Address or hostname for the OpenIM admin network.
-# Default: OPENIM_ADMIN_NETWORK_ADDRESS=172.28.0.14
-OPENIM_ADMIN_FRONT_NETWORK_ADDRESS=${OPENIM_ADMIN_FRONT_NETWORK_ADDRESS}
-  
-# Address or hostname for the alertmanager network.
-# Default: ALERT_MANAGER_NETWORK_ADDRESS=172.28.0.15
-ALERT_MANAGER_NETWORK_ADDRESS=${ALERT_MANAGER_NETWORK_ADDRESS}
+
+
 # ===============================================
 # = Component Extension Configuration =
 # ===============================================
@@ -225,7 +215,7 @@ PROMETHEUS_PORT=${PROMETHEUS_PORT}
 GRAFANA_ADDRESS=${GRAFANA_NETWORK_ADDRESS}
 
 # Port on which Grafana service is running.
-# Default: GRAFANA_PORT=13000
+# Default: GRAFANA_PORT=3000
 GRAFANA_PORT=${GRAFANA_PORT}
 
 # ======================================
@@ -293,23 +283,3 @@ SERVER_BRANCH=${SERVER_BRANCH}
 # Port for the OpenIM admin API.
 # Default: OPENIM_ADMIN_API_PORT=10009
 OPENIM_ADMIN_API_PORT=${OPENIM_ADMIN_API_PORT}
-  
-# Port for the node exporter.
-# Default: NODE_EXPORTER_PORT=19100
-NODE_EXPORTER_PORT=${NODE_EXPORTER_PORT}
-  
-# Port for the prometheus.
-# Default: PROMETHEUS_PORT=19090
-PROMETHEUS_PORT=${PROMETHEUS_PORT}
-  
-# Port for the grafana.
-# Default: GRAFANA_PORT=13000
-GRAFANA_PORT=${GRAFANA_PORT}
-  
-# Port for the admin front.
-# Default: OPENIM_ADMIN_FRONT_PORT=11002
-OPENIM_ADMIN_FRONT_PORT=${OPENIM_ADMIN_FRONT_PORT}
-
-# Port for the alertmanager.
-# Default: ALERT_MANAGER_PORT=19093
-ALERT_MANAGER_PORT=${ALERT_MANAGER_PORT}

deployments/templates/prometheus.yml

@@ -1,85 +0,0 @@
-# my global config
-global:
-  scrape_interval:     15s # Set the scrape interval to every 15 seconds. Default is every 1 minute.
-  evaluation_interval: 15s # Evaluate rules every 15 seconds. The default is every 1 minute.
-  # scrape_timeout is set to the global default (10s).
-
-# Alertmanager configuration
-alerting:
-  alertmanagers:
-    - static_configs:
-        - targets: ['${ALERT_MANAGER_ADDRESS}:${ALERT_MANAGER_PORT}']
-
-# Load rules once and periodically evaluate them according to the global 'evaluation_interval'.
-rule_files:
-  - "instance-down-rules.yml"
-# - "first_rules.yml"
-# - "second_rules.yml"
-
-# A scrape configuration containing exactly one endpoint to scrape:
-# Here it's Prometheus itself.
-scrape_configs:
-  # The job name is added as a label "job='job_name'"" to any timeseries scraped from this config.
-  # Monitored information captured by prometheus
-  - job_name: 'node-exporter'
-    static_configs:
-      - targets: [ '${NODE_EXPORTER_ADDRESS}:${NODE_EXPORTER_PORT}' ]
-        labels:
-          namespace: 'default'
-
-  # prometheus fetches application services
-  - job_name: 'openimserver-openim-api'
-    static_configs:
-      - targets: [ '${OPENIM_SERVER_ADDRESS}:${API_PROM_PORT}' ]
-        labels:
-          namespace: 'default'
-  - job_name: 'openimserver-openim-msggateway'
-    static_configs:
-      - targets: [ '${OPENIM_SERVER_ADDRESS}:${MSG_GATEWAY_PROM_PORT}' ]
-        labels:
-          namespace: 'default'
-  - job_name: 'openimserver-openim-msgtransfer'
-    static_configs:
-      - targets: [ ${MSG_TRANSFER_PROM_ADDRESS_PORT} ]
-        labels:
-          namespace: 'default'
-  - job_name: 'openimserver-openim-push'
-    static_configs:
-      - targets: [ '${OPENIM_SERVER_ADDRESS}:${PUSH_PROM_PORT}' ]
-        labels:
-          namespace: 'default'
-  - job_name: 'openimserver-openim-rpc-auth'
-    static_configs:
-      - targets: [ '${OPENIM_SERVER_ADDRESS}:${AUTH_PROM_PORT}' ]
-        labels:
-          namespace: 'default'
-  - job_name: 'openimserver-openim-rpc-conversation'
-    static_configs:
-      - targets: [ '${OPENIM_SERVER_ADDRESS}:${CONVERSATION_PROM_PORT}' ]
-        labels:
-          namespace: 'default'
-  - job_name: 'openimserver-openim-rpc-friend'
-    static_configs:
-      - targets: [ '${OPENIM_SERVER_ADDRESS}:${FRIEND_PROM_PORT}' ]
-        labels:
-          namespace: 'default'
-  - job_name: 'openimserver-openim-rpc-group'
-    static_configs:
-      - targets: [ '${OPENIM_SERVER_ADDRESS}:${GROUP_PROM_PORT}' ]
-        labels:
-          namespace: 'default'
-  - job_name: 'openimserver-openim-rpc-msg'
-    static_configs:
-      - targets: [ '${OPENIM_SERVER_ADDRESS}:${MESSAGE_PROM_PORT}' ]
-        labels:
-          namespace: 'default'
-  - job_name: 'openimserver-openim-rpc-third'
-    static_configs:
-      - targets: [ '${OPENIM_SERVER_ADDRESS}:${THIRD_PROM_PORT}' ]
-        labels:
-          namespace: 'default'
-  - job_name: 'openimserver-openim-rpc-user'
-    static_configs:
-      - targets: [ '${OPENIM_SERVER_ADDRESS}:${USER_PROM_PORT}' ]
-        labels:
-          namespace: 'default'

docker-compose.yml

@@ -67,17 +67,17 @@ services:
         ipv4_address: ${REDIS_NETWORK_ADDRESS}
 
   zookeeper:
-    image: bitnami/zookeeper:3.8
-    container_name: zookeeper
-    ports:
-     - "${ZOOKEEPER_PORT}:2181"
-    volumes:
-      - "/etc/localtime:/etc/localtime"
-    environment:
-      - ALLOW_ANONYMOUS_LOGIN=yes
-      - TZ="Asia/Shanghai"
-    restart: always
-    networks:
+   image: bitnami/zookeeper:3.8
+   container_name: zookeeper
+   ports:
+    - "${ZOOKEEPER_PORT}:2181"
+   volumes:
+     - "/etc/localtime:/etc/localtime"
+   environment:
+     - ALLOW_ANONYMOUS_LOGIN=yes
+     - TZ="Asia/Shanghai"
+   restart: always
+   networks:
       server:
         ipv4_address: ${ZOOKEEPER_NETWORK_ADDRESS}
 
@@ -142,68 +142,67 @@ services:
       server:
         ipv4_address: ${OPENIM_WEB_NETWORK_ADDRESS}
 
-  openim-admin:
-    image: ${IMAGE_REGISTRY}/openim-admin-front:v3.4.0
-    # image: ghcr.io/openimsdk/openim-admin-front:v3.4.0
-    # image: registry.cn-hangzhou.aliyuncs.com/openimsdk/openim-admin-front:v3.4.0
-    # image: openim/openim-admin-front:v3.4.0
-    container_name: openim-admin
-    restart: always
-    ports:
-      - "${OPENIM_ADMIN_FRONT_PORT}:80"
-    networks:
-      server:
-        ipv4_address: ${OPENIM_ADMIN_FRONT_NETWORK_ADDRESS}
+#   openim-server:
+#     # image: ghcr.io/openimsdk/openim-server:main
+#     # image: registry.cn-hangzhou.aliyuncs.com/openimsdk/openim-server:main
+#     # image: openim/openim-server:main
+#     image: ${IMAGE_REGISTRY}/openim-server:main
+# #    build: .
+#     container_name: openim-server
+#     ports:
+#       - ${OPENIM_WS_PORT}:10001
+#       - ${API_OPENIM_PORT}:10002
+#     healthcheck:
+#       test: ["CMD", "/openim/openim-server/scripts/check-all.sh"]
+#       interval: 300s
+#       timeout: 10s
+#       retries: 5
+#     volumes:
+#       - ./logs:/openim/openim-server/logs
+#       - ./_output/logs:/openim/openim-server/_output/logs
+#       - ./config:/openim/openim-server/config
+#       - ./scripts:/openim/openim-server/scripts
+#     restart: always
+#     depends_on:
+#       - kafka
+#       - mysql
+#       - mongodb
+#       - redis
+#       - minio
+#     logging:
+#       driver: json-file
+#       options:
+#         max-size: "1g"
+#         max-file: "2"
+#     networks:
+#       server:
+#         ipv4_address: ${OPENIM_SERVER_NETWORK_ADDRESS}
 
-  prometheus:
-    image: prom/prometheus
-    container_name: prometheus
-    hostname: prometheus
-    restart: always
-    volumes:
-      - ./config/prometheus.yml:/etc/prometheus/prometheus.yml
-      - ./config/instance-down-rules.yml:/etc/prometheus/instance-down-rules.yml
-    ports:
-      - "${PROMETHEUS_PORT}:9090"
-    networks:
-      server:
-        ipv4_address: ${PROMETHEUS_NETWORK_ADDRESS}
+#   prometheus:
+#     image: prom/prometheus
+#     volumes:
+#       - ./.docker-compose_cfg/prometheus-compose.yml:/etc/prometheus/prometheus.yml
+#     container_name: prometheus
+#     ports:
+#       - ${PROMETHEUS_PORT}:9091
+#     depends_on:
+#       - openim-server
+#     command: --web.listen-address=:9091 --config.file="/etc/prometheus/prometheus.yml"
+#     networks:
+#       openim-server:
+#         ipv4_address: ${PROMETHEUS_NETWORK_ADDRESS}
 
-  alertmanager:
-    image: prom/alertmanager
-    container_name: alertmanager
-    hostname: alertmanager
-    restart: always
-    volumes:
-      - ./config/alertmanager.yml:/etc/alertmanager/alertmanager.yml
-      - ./config/email.tmpl:/etc/alertmanager/email.tmpl
-    ports:
-      - "${ALERT_MANAGER_PORT}:9093"
-    networks:
-      server:
-        ipv4_address: ${ALERT_MANAGER_NETWORK_ADDRESS}
-
-  grafana:
-    image: grafana/grafana
-    container_name: grafana
-    hostname: grafana
-    user: root
-    restart: always
-    ports:
-      - "${GRAFANA_PORT}:3000"
-    volumes:
-      - ${DATA_DIR}/components/grafana:/var/lib/grafana
-    networks:
-      server:
-        ipv4_address: ${GRAFANA_NETWORK_ADDRESS}
-
-  node-exporter:
-    image: quay.io/prometheus/node-exporter
-    container_name: node-exporter
-    hostname: node-exporter
-    restart: always
-    ports:
-      - "${NODE_EXPORTER_PORT}:9100"
-    networks:
-      server:
-        ipv4_address: ${NODE_EXPORTER_NETWORK_ADDRESS}
+#   grafana:
+#     image: grafana/grafana
+#     volumes:
+#       - ./.docker-compose_cfg/datasource-compose.yaml:/etc/grafana/provisioning/datasources/datasource.yaml
+#       - ./.docker-compose_cfg/grafana.ini:/etc/grafana/grafana.ini
+#       - ./.docker-compose_cfg/node-exporter-full_rev1.json:/var/lib/grafana/dashboards/node-exporter-full_rev1.json
+#     container_name: grafana
+#     ports:
+#       - ${GRAFANA_PORT}:3000
+#     depends_on:
+#       - prometheus
+#     networks:
+#       openim-server:
+#         ipv4_address: ${GRAFANA_NETWORK_ADDRESS}

docs/contrib/environment.md

@@ -150,7 +150,7 @@ For convenience, configuration through modifying environment variables is recomm
   + **Description**: API address.
   + **Note**: If the server has an external IP, it will be automatically obtained. For internal networks, set this variable to the IP serving internally.
 
-  ```bash
+  ```
   export API_URL="http://ip:10002"
   ```
 
@@ -412,7 +412,7 @@ Configuration for Grafana, including its port and address.
 
 | Parameter       | Example Value              | Description           |
 | --------------- | -------------------------- | --------------------- |
-| GRAFANA_PORT    | "13000"                     | Port used by Grafana. |
+| GRAFANA_PORT    | "3000"                     | Port used by Grafana. |
 | GRAFANA_ADDRESS | "${DOCKER_BRIDGE_GATEWAY}" | Address for Grafana.  |
 
 ###  2.16. <a name='RPCPortConfigurationVariables'></a>RPC Port Configuration Variables

docs/contrib/prometheus-grafana.md

@@ -1,323 +0,0 @@
-# Deployment and Design of OpenIM's Management Backend and Monitoring
-
-<!-- vscode-markdown-toc -->
-* 1. [Source Code & Docker](#SourceCodeDocker)
-	* 1.1. [Deployment](#Deployment)
-	* 1.2. [Configuration](#Configuration)
-	* 1.3. [Monitoring Running in Docker Guide](#MonitoringRunninginDockerGuide)
-		* 1.3.1. [Introduction](#Introduction)
-		* 1.3.2. [Prerequisites](#Prerequisites)
-		* 1.3.3. [Step 1: Clone the Repository](#Step1:ClonetheRepository)
-		* 1.3.4. [Step 2: Start Docker Compose](#Step2:StartDockerCompose)
-		* 1.3.5. [Step 3: Use the OpenIM Web Interface](#Step3:UsetheOpenIMWebInterface)
-		* 1.3.6. [Running Effect](#RunningEffect)
-		* 1.3.7. [Step 4: Access the Admin Panel](#Step4:AccesstheAdminPanel)
-		* 1.3.8. [Step 5: Access the Monitoring Interface](#Step5:AccesstheMonitoringInterface)
-		* 1.3.9. [Next Steps](#NextSteps)
-		* 1.3.10. [Troubleshooting](#Troubleshooting)
-* 2. [Kubernetes](#Kubernetes)
-	* 2.1. [Middleware Monitoring](#MiddlewareMonitoring)
-	* 2.2. [Custom OpenIM Metrics](#CustomOpenIMMetrics)
-	* 2.3. [Node Exporter](#NodeExporter)
-* 3. [Setting Up and Configuring AlertManager Using Environment Variables and `make init`](#SettingUpandConfiguringAlertManagerUsingEnvironmentVariablesandmakeinit)
-	* 3.1. [Introduction](#Introduction-1)
-	* 3.2. [Prerequisites](#Prerequisites-1)
-	* 3.3. [Configuration Steps](#ConfigurationSteps)
-		* 3.3.1. [Exporting Environment Variables](#ExportingEnvironmentVariables)
-		* 3.3.2. [Initializing AlertManager](#InitializingAlertManager)
-		* 3.3.3. [Key Configuration Fields](#KeyConfigurationFields)
-		* 3.3.4. [Configuring SMTP Authentication Password](#ConfiguringSMTPAuthenticationPassword)
-		* 3.3.5. [Useful Links for Common Email Servers](#UsefulLinksforCommonEmailServers)
-	* 3.4. [Conclusion](#Conclusion)
-
-<!-- vscode-markdown-toc-config
-	numbering=true
-	autoSave=true
-	/vscode-markdown-toc-config -->
-<!-- /vscode-markdown-toc -->
-
-OpenIM offers various flexible deployment options to suit different environments and requirements. Here is a simplified and optimized description of these deployment options:
-
-1. Source Code Deployment:
-   + **Regular Source Code Deployment**: Deployment using the `nohup` method. This is a basic deployment method suitable for development and testing environments. For details, refer to the [Regular Source Code Deployment Guide](https://docs.openim.io/).
-   + **Production-Level Deployment**: Deployment using the `system` method, more suitable for production environments. This method provides higher stability and reliability. For details, refer to the [Production-Level Deployment Guide](https://docs.openim.io/guides/gettingStarted/install-openim-linux-system).
-2. Cluster Deployment:
-   + **Kubernetes Deployment**: Provides two deployment methods, including deployment through Helm and sealos. This is suitable for environments that require high availability and scalability. Specific methods can be found in the [Kubernetes Deployment Guide](https://docs.openim.io/guides/gettingStarted/k8s-deployment).
-3. Docker Deployment:
-   + **Regular Docker Deployment**: Suitable for quick deployments and small projects. For detailed information, refer to the [Docker Deployment Guide](https://docs.openim.io/guides/gettingStarted/dockerCompose).
-   + **Docker Compose Deployment**: Provides more convenient service management and configuration, suitable for complex multi-container applications.
-
-Next, we will introduce the specific steps, monitoring, and management backend configuration for each of these deployment methods, as well as usage tips to help you choose the most suitable deployment option according to your needs.
-
-##  1. <a name='SourceCodeDocker'></a>Source Code & Docker
-
-###  1.1. <a name='Deployment'></a>Deployment
-
-OpenIM deploys openim-server and openim-chat from source code, while other components are deployed via Docker.
-
-For Docker deployment, you can deploy all components with a single command using the [openimsdk/openim-docker](https://github.com/openimsdk/openim-docker) repository. The deployment configuration can be found in the [environment.sh](https://github.com/openimsdk/open-im-server/blob/main/scripts/install/environment.sh) document, which provides information on how to learn and familiarize yourself with various environment variables.
-
-For Prometheus, it is not enabled by default. To enable it, set the environment variable before executing `make init`:
-
-```bash
-export PROMETHEUS_ENABLE=true   # Default is false
-```
-
-Then, execute:
-
-```bash
-make init
-docker compose up -d
-```
-
-###  1.2. <a name='Configuration'></a>Configuration
-
-To configure Prometheus data sources in Grafana, follow these steps:
-
-1. **Log in to Grafana**: First, open your web browser and access the Grafana URL. If you haven't changed the port, the address is typically [http://localhost:13000](http://localhost:13000/).
-
-2. **Log in with default credentials**: Grafana's default username and password are both `admin`. You will be prompted to change the password on your first login.
-
-3. **Access Data Sources Settings**:
-
-   + In the left menu of Grafana, look for and click the "gear" icon representing "Configuration."
-   + In the configuration menu, select "Data Sources."
-
-4. **Add a New Data Source**:
-
-   + On the Data Sources page, click the "Add data source" button.
-   + In the list, find and select "Prometheus."
-
-   ![image-20231114175117374](http://sm.nsddd.top/sm202311141751692.png)
-
-   Click `Add New connection` to add more data sources, such as Loki (responsible for log storage and query processing).
-
-5. **Configure the Prometheus Data Source**:
-
-   + On the configuration page, fill in the details of the Prometheus server. This typically includes the URL of the Prometheus service (e.g., if Prometheus is running on the same machine as OpenIM, the URL might be `http://172.28.0.1:19090`, with the address matching the `DOCKER_BRIDGE_GATEWAY` variable address). OpenIM and the components are linked via a gateway. The default port used by OpenIM is `19090`.
-   + Adjust other settings as needed, such as authentication and TLS settings.
-
-   ![image-20231114180351923](http://sm.nsddd.top/sm202311141803076.png)
-
-6. **Save and Test**:
-
-   + After completing the configuration, click the "Save & Test" button to ensure that Grafana can successfully connect to Prometheus.
-
-**Importing Dashboards in Grafana**
-
-Importing Grafana Dashboards is a straightforward process and is applicable to OpenIM Server application services and Node Exporter. Here are detailed steps and necessary considerations:
-
-**Key Metrics Overview and Deployment Steps**
-
-To monitor OpenIM in Grafana, you need to focus on three categories of key metrics, each with its specific deployment and configuration steps:
-
-1. **OpenIM Metrics (`prometheus-dashboard.yaml`)**:
-   + **Configuration File Path**: Located at `config/prometheus-dashboard.yaml`.
-   + **Enabling Monitoring**: Set the environment variable `export PROMETHEUS_ENABLE=true` to enable Prometheus monitoring.
-   + **More Information**: Refer to the [OpenIM Configuration Guide](https://docs.openim.io/configurations/prometheus-integration).
-2. **Node Exporter**:
-   + **Container Deployment**: Deploy the `quay.io/prometheus/node-exporter` container for node monitoring.
-   + **Get Dashboard**: Access the [Node Exporter Full Feature Dashboard](https://grafana.com/grafana/dashboards/1860-node-exporter-full/) and import it using YAML file download or ID import.
-   + **Deployment Guide**: Refer to the [Node Exporter Deployment Documentation](https://prometheus.io/docs/guides/node-exporter/).
-3. **Middleware Metrics**: Each middleware requires specific steps and configurations to enable monitoring. Here is a list of common middleware and links to their respective setup guides:
-   + MySQL:
-     + **Configuration**: Ensure MySQL has performance monitoring enabled.
-     + **Link**: Refer to the [MySQL Monitoring Configuration Guide](https://grafana.com/docs/grafana/latest/datasources/mysql/).
-   + Redis:
-     + **Configuration**: Configure Redis to allow monitoring data export.
-     + **Link**: Refer to the [Redis Monitoring Guide](https://grafana.com/docs/grafana/latest/datasources/redis/).
-   + MongoDB:
-     + **Configuration**: Set up monitoring metrics for MongoDB.
-     + **Link**: Refer to the [MongoDB Monitoring Guide](https://grafana.com/grafana/plugins/grafana-mongodb-datasource/).
-   + Kafka:
-     + **Configuration**: Integrate Kafka with Prometheus monitoring.
-     + **Link**: Refer to the [Kafka Monitoring Guide](https://grafana.com/grafana/plugins/grafana-kafka-datasource/).
-   + Zookeeper:
-     + **Configuration**: Ensure Zookeeper can be monitored by Prometheus.
-     + **Link**: Refer to the [Zookeeper Monitoring Configuration](https://grafana.com/docs/grafana/latest/datasources/zookeeper/).
-
-
-
-**Importing Steps**:
-
-1. Access the Dashboard Import Interface:
-
-   + Click the `+` icon on the left menu or in the top right corner of Grafana, then select "Create."
-   + Choose "Import" to access the dashboard import interface.
-
-2. **Perform Dashboard Import**:
-   + **Upload via File**: Directly upload your YAML file.
-   + **Paste Content**: Open the YAML file, copy its content, and paste it into the import interface.
-   + **Import via Grafana.com Dashboard**: Visit [Grafana Dashboards](https://grafana.com/grafana/dashboards/), search for the desired dashboard, and import it using its ID.
-3. **Configure the Dashboard**:
-   + Select the appropriate data source, such as the previously configured Prometheus.
-   + Adjust other settings, such as the dashboard name or folder.
-4. **Save and View the Dashboard**:
-   + After configuring, click "Import" to complete the process.
-   + Immediately view the new dashboard after successful import.
-
-**Graph Examples:**
-
-![image-20231114194451673](http://sm.nsddd.top/sm202311141944953.png)
-
-
-
-###  1.3. <a name='MonitoringRunninginDockerGuide'></a>Monitoring Running in Docker Guide
-
-####  1.3.1. <a name='Introduction'></a>Introduction
-
-This guide provides the steps to run OpenIM using Docker. OpenIM is an open-source instant messaging solution that can be quickly deployed using Docker. For more information, please refer to the [OpenIM Docker GitHub](https://github.com/openimsdk/openim-docker).
-
-####  1.3.2. <a name='Prerequisites'></a>Prerequisites
-
-+ Ensure that Docker and Docker Compose are installed.
-+ Basic understanding of Docker and containerization technology.
-
-####  1.3.3. <a name='Step1:ClonetheRepository'></a>Step 1: Clone the Repository
-
-First, clone the OpenIM Docker repository:
-
-```bash
-git clone https://github.com/openimsdk/openim-docker.git
-```
-
-Navigate to the repository directory and check the `README` file for more information and configuration options.
-
-####  1.3.4. <a name='Step2:StartDockerCompose'></a>Step 2: Start Docker Compose
-
-In the repository directory, run the following command to start the service:
-
-```bash
-docker-compose up -d
-```
-
-This will download the required Docker images and start the OpenIM service.
-
-####  1.3.5. <a name='Step3:UsetheOpenIMWebInterface'></a>Step 3: Use the OpenIM Web Interface
-
-+ Open a browser in private mode and access [OpenIM Web](http://localhost:11001/).
-+ Register two users and try adding friends.
-+ Test sending messages and pictures.
-
-####  1.3.6. <a name='RunningEffect'></a>Running Effect
-
-![image-20231115100811208](http://sm.nsddd.top/sm202311151008639.png)
-
-####  1.3.7. <a name='Step4:AccesstheAdminPanel'></a>Step 4: Access the Admin Panel
-
-+ Access the [OpenIM Admin Panel](http://localhost:11002/).
-+ Log in using the default username and password (`admin1:admin1`).
-
-Running Effect Image:
-
-![image-20231115101039837](http://sm.nsddd.top/sm202311151010116.png)
-
-####  1.3.8. <a name='Step5:AccesstheMonitoringInterface'></a>Step 5: Access the Monitoring Interface
-
-+ Log in to the [Monitoring Interface](http://localhost:3000/login) using the credentials (`admin:admin`).
-
-####  1.3.9. <a name='NextSteps'></a>Next Steps
-
-+ Configure and manage the services following the steps provided in the OpenIM source code.
-+ Refer to the `README` file for advanced configuration and management.
-
-####  1.3.10. <a name='Troubleshooting'></a>Troubleshooting
-
-+ If you encounter any issues, please check the documentation on [OpenIM Docker GitHub](https://github.com/openimsdk/openim-docker) or search for related issues in the Issues section.
-+ If the problem persists, you can create an issue on the [openim-docker](https://github.com/openimsdk/openim-docker/issues/new/choose) repository or the [openim-server](https://github.com/openimsdk/open-im-server/issues/new/choose) repository.
-
-
-
-##  2. <a name='Kubernetes'></a>Kubernetes
-
-Refer to [openimsdk/helm-charts](https://github.com/openimsdk/helm-charts).
-
-When deploying and monitoring OpenIM in a Kubernetes environment, you will focus on three main metrics: middleware, custom OpenIM metrics, and Node Exporter. Here are detailed steps and guidelines:
-
-###  2.1. <a name='MiddlewareMonitoring'></a>Middleware Monitoring
-
-Middleware monitoring is crucial to ensure the overall system's stability. Typically, this includes monitoring the following components:
-
-+ **MySQL**: Monitor database performance, query latency, and more.
-+ **Redis**: Track operation latency, memory usage, and more.
-+ **MongoDB**: Observe database operations, resource usage, and more.
-+ **Kafka**: Monitor message throughput, latency, and more.
-+ **Zookeeper**: Keep an eye on cluster status, performance metrics, and more.
-
-For Kubernetes environments, you can use the corresponding Prometheus Exporters to collect monitoring data for these middleware components.
-
-###  2.2. <a name='CustomOpenIMMetrics'></a>Custom OpenIM Metrics
-
-Custom OpenIM metrics provide essential information about the OpenIM application itself, such as user activity, message traffic, system performance, and more. To monitor these metrics in Kubernetes:
-
-+ Ensure OpenIM application configurations expose Prometheus metrics.
-+ When deploying using Helm charts (refer to [OpenIM Helm Charts](https://github.com/openimsdk/helm-charts)), pay attention to configuring relevant monitoring settings.
-
-###  2.3. <a name='NodeExporter'></a>Node Exporter
-
-Node Exporter is used to collect hardware and operating system-level metrics for Kubernetes nodes, such as CPU, memory, disk usage, and more. To integrate Node Exporter in Kubernetes:
-
-+ Deploy Node Exporter using the appropriate Helm chart. You can find information and guides on [Prometheus Community](https://prometheus.io/docs/guides/node-exporter/).
-+ Ensure Node Exporter's data is collected by Prometheus instances within your cluster.
-
-
-
-##  3. <a name='SettingUpandConfiguringAlertManagerUsingEnvironmentVariablesandmakeinit'></a>Setting Up and Configuring AlertManager Using Environment Variables and `make init`
-
-###  3.1. <a name='Introduction-1'></a>Introduction
-
-AlertManager, a component of the Prometheus monitoring system, handles alerts sent by client applications such as the Prometheus server. It takes care of deduplicating, grouping, and routing them to the correct receiver. This document outlines how to set up and configure AlertManager using environment variables and the `make init` command. We will focus on configuring key fields like the sender's email, SMTP settings, and SMTP authentication password.
-
-###  3.2. <a name='Prerequisites-1'></a>Prerequisites
-
-+ Basic knowledge of terminal and command-line operations.
-+ AlertManager installed on your system.
-+ Access to an SMTP server for sending emails.
-
-###  3.3. <a name='ConfigurationSteps'></a>Configuration Steps
-
-####  3.3.1. <a name='ExportingEnvironmentVariables'></a>Exporting Environment Variables
-
-Before initializing AlertManager, you need to set environment variables. These variables are used to configure the AlertManager settings without altering the code. Use the `export` command in your terminal. Here are some key variables you might set:
-
-+ `export ALERTMANAGER_RESOLVE_TIMEOUT='5m'`
-+ `export ALERTMANAGER_SMTP_FROM='alert@example.com'`
-+ `export ALERTMANAGER_SMTP_SMARTHOST='smtp.example.com:465'`
-+ `export ALERTMANAGER_SMTP_AUTH_USERNAME='alert@example.com'`
-+ `export ALERTMANAGER_SMTP_AUTH_PASSWORD='your_password'`
-+ `export ALERTMANAGER_SMTP_REQUIRE_TLS='false'`
-
-####  3.3.2. <a name='InitializingAlertManager'></a>Initializing AlertManager
-
-After setting the necessary environment variables, you can initialize AlertManager by running the `make init` command. This command typically runs a script that prepares AlertManager with the provided configuration.
-
-####  3.3.3. <a name='KeyConfigurationFields'></a>Key Configuration Fields
-
-##### a. Sender's Email (`ALERTMANAGER_SMTP_FROM`)
-
-This variable sets the email address that will appear as the sender in the notifications sent by AlertManager.
-
-##### b. SMTP Configuration
-
-+ **SMTP Server (`ALERTMANAGER_SMTP_SMARTHOST`):** Specifies the address and port of the SMTP server used for sending emails.
-+ **SMTP Authentication Username (`ALERTMANAGER_SMTP_AUTH_USERNAME`):** The username for authenticating with the SMTP server.
-+ **SMTP Authentication Password (`ALERTMANAGER_SMTP_AUTH_PASSWORD`):** The password for SMTP server authentication. It's crucial to keep this value secure.
-
-####  3.3.4. <a name='ConfiguringSMTPAuthenticationPassword'></a>Configuring SMTP Authentication Password
-
-The SMTP authentication password can be set using the `ALERTMANAGER_SMTP_AUTH_PASSWORD` environment variable. It's recommended to use a secure method to set this variable to avoid exposing sensitive information. For instance, you might read the password from a secure file or a secret management tool.
-
-####  3.3.5. <a name='UsefulLinksforCommonEmailServers'></a>Useful Links for Common Email Servers
-
-For specific configurations related to common email servers, you may refer to their respective documentation:
-
-+ Gmail SMTP Settings:
-  + [Gmail SMTP Configuration](https://support.google.com/mail/answer/7126229?hl=en)
-+ Microsoft Outlook SMTP Settings:
-  + [Outlook Email Settings](https://support.microsoft.com/en-us/office/pop-imap-and-smtp-settings-8361e398-8af4-4e97-b147-6c6c4ac95353)
-+ Yahoo Mail SMTP Settings:
-  + [Yahoo SMTP Configuration](https://help.yahoo.com/kb/SLN4724.html)
-
-###  3.4. <a name='Conclusion'></a>Conclusion
-
-Setting up and configuring AlertManager with environment variables provides a flexible and secure way to manage alert settings. By following the above steps, you can easily configure AlertManager for your monitoring needs. Always ensure to secure sensitive information, especially when dealing with SMTP authentication credentials.

go.mod

@@ -44,8 +44,6 @@ require (
 	github.com/go-sql-driver/mysql v1.7.1
 	github.com/redis/go-redis/v9 v9.2.1
 	github.com/tencentyun/cos-go-sdk-v5 v0.7.45
-	go.uber.org/automaxprocs v1.5.3
-	golang.org/x/sync v0.4.0
 	gopkg.in/src-d/go-git.v4 v4.13.1
 	gotest.tools v2.2.0+incompatible
 )
@@ -133,6 +131,7 @@ require (
 	golang.org/x/arch v0.3.0 // indirect
 	golang.org/x/net v0.17.0 // indirect
 	golang.org/x/oauth2 v0.13.0 // indirect
+	golang.org/x/sync v0.4.0 // indirect
 	golang.org/x/sys v0.13.0 // indirect
 	golang.org/x/text v0.13.0 // indirect
 	golang.org/x/time v0.3.0 // indirect

go.sum

@@ -280,7 +280,6 @@ github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/prashantv/gostub v1.1.0 h1:BTyx3RfQjRHnUWaGF9oQos79AlQ5k8WNktv7VGvVH4g=
 github.com/prometheus/client_golang v1.17.0 h1:rl2sfwZMtSthVU752MqfjQozy7blglC+1SOtjMAMh+Q=
 github.com/prometheus/client_golang v1.17.0/go.mod h1:VeL+gMmOAxkS2IqfCq0ZmHSL+LjWfWDUmp1mBz9JgUY=
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
@@ -353,8 +352,6 @@ go.opencensus.io v0.24.0 h1:y73uSU6J157QMP2kn2r30vwW1A2W2WFwSCGnAVxeaD0=
 go.opencensus.io v0.24.0/go.mod h1:vNK8G9p7aAivkbmorf4v+7Hgx+Zs0yY+0fOtgBfjQKo=
 go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw=
 go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
-go.uber.org/automaxprocs v1.5.3 h1:kWazyxZUrS3Gs4qUpbwo5kEIMGe/DAvi5Z4tl2NW4j8=
-go.uber.org/automaxprocs v1.5.3/go.mod h1:eRbA25aqJrxAbsLO0xy5jVwPt7FQnRgjW+efnwa1WM0=
 go.uber.org/goleak v1.1.11 h1:wy28qYRKZgnJTxGxvye5/wgWr1EKjmUDGYox5mGlRlI=
 go.uber.org/multierr v1.6.0 h1:y6IPFStTAIT5Ytl7/XYmHvzXQ7S3g/IeZW9hyZ5thw4=
 go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU=

internal/api/third.go

@@ -15,12 +15,11 @@
 package api
 
 import (
+	config2 "github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"math/rand"
 	"net/http"
 	"strconv"
 
-	config2 "github.com/openimsdk/open-im-server/v3/pkg/common/config"
-
 	"github.com/gin-gonic/gin"
 
 	"github.com/OpenIMSDK/protocol/third"

internal/msggateway/client.go

@@ -167,7 +167,7 @@ func (c *Client) readMessage() {
 func (c *Client) handleMessage(message []byte) error {
 	if c.IsCompress {
 		var err error
-		message, err = c.longConnServer.DecompressWithPool(message)
+		message, err = c.longConnServer.DeCompress(message)
 		if err != nil {
 			return utils.Wrap(err, "")
 		}
@@ -317,7 +317,7 @@ func (c *Client) writeBinaryMsg(resp Resp) error {
 
 	_ = c.conn.SetWriteDeadline(writeWait)
 	if c.IsCompress {
-		resultBuf, compressErr := c.longConnServer.CompressWithPool(encodedBuf)
+		resultBuf, compressErr := c.longConnServer.Compress(encodedBuf)
 		if compressErr != nil {
 			return utils.Wrap(compressErr, "")
 		}

internal/msggateway/compressor.go

@@ -17,23 +17,14 @@ package msggateway
 import (
 	"bytes"
 	"compress/gzip"
-	"errors"
 	"io"
-	"sync"
 
 	"github.com/OpenIMSDK/tools/utils"
 )
 
-var (
-	gzipWriterPool = sync.Pool{New: func() any { return gzip.NewWriter(nil) }}
-	gzipReaderPool = sync.Pool{New: func() any { return new(gzip.Reader) }}
-)
-
 type Compressor interface {
 	Compress(rawData []byte) ([]byte, error)
-	CompressWithPool(rawData []byte) ([]byte, error)
 	DeCompress(compressedData []byte) ([]byte, error)
-	DecompressWithPool(compressedData []byte) ([]byte, error)
 }
 type GzipCompressor struct {
 	compressProtocol string
@@ -55,22 +46,6 @@ func (g *GzipCompressor) Compress(rawData []byte) ([]byte, error) {
 	return gzipBuffer.Bytes(), nil
 }
 
-func (g *GzipCompressor) CompressWithPool(rawData []byte) ([]byte, error) {
-	gz := gzipWriterPool.Get().(*gzip.Writer)
-	defer gzipWriterPool.Put(gz)
-
-	gzipBuffer := bytes.Buffer{}
-	gz.Reset(&gzipBuffer)
-
-	if _, err := gz.Write(rawData); err != nil {
-		return nil, utils.Wrap(err, "")
-	}
-	if err := gz.Close(); err != nil {
-		return nil, utils.Wrap(err, "")
-	}
-	return gzipBuffer.Bytes(), nil
-}
-
 func (g *GzipCompressor) DeCompress(compressedData []byte) ([]byte, error) {
 	buff := bytes.NewBuffer(compressedData)
 	reader, err := gzip.NewReader(buff)
@@ -84,23 +59,3 @@ func (g *GzipCompressor) DeCompress(compressedData []byte) ([]byte, error) {
 	_ = reader.Close()
 	return compressedData, nil
 }
-
-func (g *GzipCompressor) DecompressWithPool(compressedData []byte) ([]byte, error) {
-	reader := gzipReaderPool.Get().(*gzip.Reader)
-	if reader == nil {
-		return nil, errors.New("NewReader failed")
-	}
-	defer gzipReaderPool.Put(reader)
-
-	err := reader.Reset(bytes.NewReader(compressedData))
-	if err != nil {
-		return nil, utils.Wrap(err, "NewReader failed")
-	}
-
-	compressedData, err = io.ReadAll(reader)
-	if err != nil {
-		return nil, utils.Wrap(err, "ReadAll failed")
-	}
-	_ = reader.Close()
-	return compressedData, nil
-}

internal/msggateway/compressor_test.go

@@ -1,107 +0,0 @@
-package msggateway
-
-import (
-	"crypto/rand"
-	"sync"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func mockRandom() []byte {
-	bs := make([]byte, 50)
-	rand.Read(bs)
-	return bs
-}
-
-func TestCompressDecompress(t *testing.T) {
-
-	compressor := NewGzipCompressor()
-
-	for i := 0; i < 2000; i++ {
-		src := mockRandom()
-
-		// compress
-		dest, err := compressor.CompressWithPool(src)
-		assert.Equal(t, nil, err)
-
-		// decompress
-		res, err := compressor.DecompressWithPool(dest)
-		assert.Equal(t, nil, err)
-
-		// check
-		assert.EqualValues(t, src, res)
-	}
-}
-
-func TestCompressDecompressWithConcurrency(t *testing.T) {
-	wg := sync.WaitGroup{}
-	compressor := NewGzipCompressor()
-
-	for i := 0; i < 200; i++ {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			src := mockRandom()
-
-			// compress
-			dest, err := compressor.CompressWithPool(src)
-			assert.Equal(t, nil, err)
-
-			// decompress
-			res, err := compressor.DecompressWithPool(dest)
-			assert.Equal(t, nil, err)
-
-			// check
-			assert.EqualValues(t, src, res)
-
-		}()
-	}
-	wg.Wait()
-}
-
-func BenchmarkCompress(b *testing.B) {
-	src := mockRandom()
-	compressor := NewGzipCompressor()
-
-	for i := 0; i < b.N; i++ {
-		_, err := compressor.Compress(src)
-		assert.Equal(b, nil, err)
-	}
-}
-
-func BenchmarkCompressWithSyncPool(b *testing.B) {
-	src := mockRandom()
-
-	compressor := NewGzipCompressor()
-	for i := 0; i < b.N; i++ {
-		_, err := compressor.CompressWithPool(src)
-		assert.Equal(b, nil, err)
-	}
-}
-
-func BenchmarkDecompress(b *testing.B) {
-	src := mockRandom()
-
-	compressor := NewGzipCompressor()
-	comdata, err := compressor.Compress(src)
-	assert.Equal(b, nil, err)
-
-	for i := 0; i < b.N; i++ {
-		_, err := compressor.DeCompress(comdata)
-		assert.Equal(b, nil, err)
-	}
-}
-
-func BenchmarkDecompressWithSyncPool(b *testing.B) {
-	src := mockRandom()
-
-	compressor := NewGzipCompressor()
-	comdata, err := compressor.Compress(src)
-	assert.Equal(b, nil, err)
-
-	for i := 0; i < b.N; i++ {
-		_, err := compressor.DecompressWithPool(comdata)
-		assert.Equal(b, nil, err)
-	}
-}

internal/msggateway/hub_server.go

@@ -17,19 +17,22 @@ package msggateway
 import (
 	"context"
 
+	"github.com/OpenIMSDK/tools/mcontext"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
+
+	"github.com/OpenIMSDK/tools/errs"
 	"google.golang.org/grpc"
 
+	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
+
 	"github.com/OpenIMSDK/protocol/constant"
 	"github.com/OpenIMSDK/protocol/msggateway"
 	"github.com/OpenIMSDK/tools/discoveryregistry"
-	"github.com/OpenIMSDK/tools/errs"
 	"github.com/OpenIMSDK/tools/log"
-	"github.com/OpenIMSDK/tools/mcontext"
 	"github.com/OpenIMSDK/tools/utils"
 
-	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/startrpc"
 )
 
@@ -38,7 +41,6 @@ func (s *Server) InitServer(disCov discoveryregistry.SvcDiscoveryRegistry, serve
 	if err != nil {
 		return err
 	}
-
 	msgModel := cache.NewMsgCacheModel(rdb)
 	s.LongConnServer.SetDiscoveryRegistry(disCov)
 	s.LongConnServer.SetCacheHandler(msgModel)
@@ -95,25 +97,22 @@ func (s *Server) GetUsersOnlineStatus(
 		if !ok {
 			continue
 		}
-
-		uresp := new(msggateway.GetUsersOnlineStatusResp_SuccessResult)
-		uresp.UserID = userID
+		temp := new(msggateway.GetUsersOnlineStatusResp_SuccessResult)
+		temp.UserID = userID
 		for _, client := range clients {
-			if client == nil {
-				continue
+			if client != nil {
+				ps := new(msggateway.GetUsersOnlineStatusResp_SuccessDetail)
+				ps.Platform = constant.PlatformIDToName(client.PlatformID)
+				ps.Status = constant.OnlineStatus
+				ps.ConnID = client.ctx.GetConnID()
+				ps.Token = client.token
+				ps.IsBackground = client.IsBackground
+				temp.Status = constant.OnlineStatus
+				temp.DetailPlatformStatus = append(temp.DetailPlatformStatus, ps)
 			}
-
-			ps := new(msggateway.GetUsersOnlineStatusResp_SuccessDetail)
-			ps.Platform = constant.PlatformIDToName(client.PlatformID)
-			ps.Status = constant.OnlineStatus
-			ps.ConnID = client.ctx.GetConnID()
-			ps.Token = client.token
-			ps.IsBackground = client.IsBackground
-			uresp.Status = constant.OnlineStatus
-			uresp.DetailPlatformStatus = append(uresp.DetailPlatformStatus, ps)
 		}
-		if uresp.Status == constant.OnlineStatus {
-			resp.SuccessResult = append(resp.SuccessResult, uresp)
+		if temp.Status == constant.OnlineStatus {
+			resp.SuccessResult = append(resp.SuccessResult, temp)
 		}
 	}
 	return &resp, nil
@@ -130,55 +129,50 @@ func (s *Server) SuperGroupOnlineBatchPushOneMsg(
 	ctx context.Context,
 	req *msggateway.OnlineBatchPushOneMsgReq,
 ) (*msggateway.OnlineBatchPushOneMsgResp, error) {
-
-	var singleUserResults []*msggateway.SingleMsgToUserResults
-
+	var singleUserResult []*msggateway.SingleMsgToUserResults
 	for _, v := range req.PushToUserIDs {
 		var resp []*msggateway.SingleMsgToUserPlatform
-		results := &msggateway.SingleMsgToUserResults{
+		tempT := &msggateway.SingleMsgToUserResults{
 			UserID: v,
 		}
 		clients, ok := s.LongConnServer.GetUserAllCons(v)
 		if !ok {
 			log.ZDebug(ctx, "push user not online", "userID", v)
-			results.Resp = resp
-			singleUserResults = append(singleUserResults, results)
+			tempT.Resp = resp
+			singleUserResult = append(singleUserResult, tempT)
 			continue
 		}
-
 		log.ZDebug(ctx, "push user online", "clients", clients, "userID", v)
 		for _, client := range clients {
-			if client == nil {
-				continue
-			}
-
-			userPlatform := &msggateway.SingleMsgToUserPlatform{
-				RecvID:         v,
-				RecvPlatFormID: int32(client.PlatformID),
-			}
-			if !client.IsBackground ||
-				(client.IsBackground && client.PlatformID != constant.IOSPlatformID) {
-				err := client.PushMessage(ctx, req.MsgData)
-				if err != nil {
-					userPlatform.ResultCode = -2
-					resp = append(resp, userPlatform)
-				} else {
-					if utils.IsContainInt(client.PlatformID, s.pushTerminal) {
-						results.OnlinePush = true
-						resp = append(resp, userPlatform)
-					}
+			if client != nil {
+				temp := &msggateway.SingleMsgToUserPlatform{
+					RecvID:         v,
+					RecvPlatFormID: int32(client.PlatformID),
+				}
+				if !client.IsBackground ||
+					(client.IsBackground == true && client.PlatformID != constant.IOSPlatformID) {
+					err := client.PushMessage(ctx, req.MsgData)
+					if err != nil {
+						temp.ResultCode = -2
+						resp = append(resp, temp)
+					} else {
+						if utils.IsContainInt(client.PlatformID, s.pushTerminal) {
+							tempT.OnlinePush = true
+							resp = append(resp, temp)
+						}
+					}
+				} else {
+					temp.ResultCode = -3
+					resp = append(resp, temp)
 				}
-			} else {
-				userPlatform.ResultCode = -3
-				resp = append(resp, userPlatform)
 			}
 		}
-		results.Resp = resp
-		singleUserResults = append(singleUserResults, results)
+		tempT.Resp = resp
+		singleUserResult = append(singleUserResult, tempT)
 	}
 
 	return &msggateway.OnlineBatchPushOneMsgResp{
-		SinglePushResult: singleUserResults,
+		SinglePushResult: singleUserResult,
 	}, nil
 }
 
@@ -187,21 +181,17 @@ func (s *Server) KickUserOffline(
 	req *msggateway.KickUserOfflineReq,
 ) (*msggateway.KickUserOfflineResp, error) {
 	for _, v := range req.KickUserIDList {
-		clients, _, ok := s.LongConnServer.GetUserPlatformCons(v, int(req.PlatformID))
-		if !ok {
-			log.ZInfo(ctx, "conn not exist", "userID", v, "platformID", req.PlatformID)
-			continue
-		}
-
-		for _, client := range clients {
-			log.ZDebug(ctx, "kick user offline", "userID", v, "platformID", req.PlatformID, "client", client)
-			if err := client.longConnServer.KickUserConn(client); err != nil {
-				log.ZWarn(ctx, "kick user offline failed", err, "userID", v, "platformID", req.PlatformID)
+		if clients, _, ok := s.LongConnServer.GetUserPlatformCons(v, int(req.PlatformID)); ok {
+			for _, client := range clients {
+				log.ZDebug(ctx, "kick user offline", "userID", v, "platformID", req.PlatformID, "client", client)
+				if err := client.longConnServer.KickUserConn(client); err != nil {
+					log.ZWarn(ctx, "kick user offline failed", err, "userID", v, "platformID", req.PlatformID)
+				}
 			}
+		} else {
+			log.ZInfo(ctx, "conn not exist", "userID", v, "platformID", req.PlatformID)
 		}
-		continue
 	}
-
 	return &msggateway.KickUserOfflineResp{}, nil
 }
 

internal/msggateway/init.go

@@ -23,7 +23,7 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 )
 
-// RunWsAndServer run ws server.
+// RunWsAndServer run ws server
 func RunWsAndServer(rpcPort, wsPort, prometheusPort int) error {
 	fmt.Println(
 		"start rpc/msg_gateway server, port: ",
@@ -37,9 +37,7 @@ func RunWsAndServer(rpcPort, wsPort, prometheusPort int) error {
 		WithPort(wsPort),
 		WithMaxConnNum(int64(config.Config.LongConnSvr.WebsocketMaxConnNum)),
 		WithHandshakeTimeout(time.Duration(config.Config.LongConnSvr.WebsocketTimeout)*time.Second),
-		WithMessageMaxMsgLength(config.Config.LongConnSvr.WebsocketMaxMsgLen),
-		WithWriteBufferSize(config.Config.LongConnSvr.WebsocketWriteBufferSize),
-	)
+		WithMessageMaxMsgLength(config.Config.LongConnSvr.WebsocketMaxMsgLen))
 	if err != nil {
 		return err
 	}

internal/msggateway/long_conn.go

@@ -50,11 +50,10 @@ type GWebSocket struct {
 	protocolType     int
 	conn             *websocket.Conn
 	handshakeTimeout time.Duration
-	writeBufferSize  int
 }
 
-func newGWebSocket(protocolType int, handshakeTimeout time.Duration, wbs int) *GWebSocket {
-	return &GWebSocket{protocolType: protocolType, handshakeTimeout: handshakeTimeout, writeBufferSize: wbs}
+func newGWebSocket(protocolType int, handshakeTimeout time.Duration) *GWebSocket {
+	return &GWebSocket{protocolType: protocolType, handshakeTimeout: handshakeTimeout}
 }
 
 func (d *GWebSocket) Close() error {
@@ -66,10 +65,6 @@ func (d *GWebSocket) GenerateLongConn(w http.ResponseWriter, r *http.Request) er
 		HandshakeTimeout: d.handshakeTimeout,
 		CheckOrigin:      func(r *http.Request) bool { return true },
 	}
-	if d.writeBufferSize > 0 { // default is 4kb.
-		upgrader.WriteBufferSize = d.writeBufferSize
-	}
-
 	conn, err := upgrader.Upgrade(w, r, nil)
 	if err != nil {
 		return err

internal/msggateway/n_ws_server.go

@@ -17,28 +17,32 @@ package msggateway
 import (
 	"context"
 	"errors"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/prom_metrics"
 	"net/http"
 	"strconv"
 	"sync"
 	"sync/atomic"
 	"time"
 
-	"github.com/go-playground/validator/v10"
-	"github.com/redis/go-redis/v9"
-	"golang.org/x/sync/errgroup"
+	"github.com/OpenIMSDK/protocol/msggateway"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
+	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
 
 	"github.com/OpenIMSDK/protocol/constant"
-	"github.com/OpenIMSDK/protocol/msggateway"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
+
+	"github.com/redis/go-redis/v9"
+
 	"github.com/OpenIMSDK/tools/discoveryregistry"
+
+	"github.com/go-playground/validator/v10"
+
 	"github.com/OpenIMSDK/tools/errs"
 	"github.com/OpenIMSDK/tools/log"
 	"github.com/OpenIMSDK/tools/utils"
-
-	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
-	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
 )
 
 type LongConnServer interface {
@@ -57,6 +61,12 @@ type LongConnServer interface {
 	MessageHandler
 }
 
+var bufferPool = sync.Pool{
+	New: func() interface{} {
+		return make([]byte, 1024)
+	},
+}
+
 type WsServer struct {
 	port              int
 	wsMaxConnNum      int64
@@ -68,7 +78,7 @@ type WsServer struct {
 	onlineUserNum     atomic.Int64
 	onlineUserConnNum atomic.Int64
 	handshakeTimeout  time.Duration
-	writeBufferSize   int
+	hubServer         *Server
 	validate          *validator.Validate
 	cache             cache.MsgModel
 	userClient        *rpcclient.UserRpcClient
@@ -138,7 +148,6 @@ func NewWsServer(opts ...Option) (*WsServer, error) {
 	return &WsServer{
 		port:             config.port,
 		wsMaxConnNum:     config.maxConnNum,
-		writeBufferSize:  config.writeBufferSize,
 		handshakeTimeout: config.handshakeTimeout,
 		clientPool: sync.Pool{
 			New: func() interface{} {
@@ -174,39 +183,27 @@ func (ws *WsServer) Run() error {
 	return http.ListenAndServe(":"+utils.IntToString(ws.port), nil) // Start listening
 }
 
-var concurrentRequest = 3
-
 func (ws *WsServer) sendUserOnlineInfoToOtherNode(ctx context.Context, client *Client) error {
 	conns, err := ws.disCov.GetConns(ctx, config.Config.RpcRegisterName.OpenImMessageGatewayName)
 	if err != nil {
 		return err
 	}
-
-	wg := errgroup.Group{}
-	wg.SetLimit(concurrentRequest)
-
 	// Online push user online message to other node
 	for _, v := range conns {
-		v := v // safe closure var
 		if v.Target() == ws.disCov.GetSelfConnTarget() {
 			log.ZDebug(ctx, "Filter out this node", "node", v.Target())
 			continue
 		}
-
-		wg.Go(func() error {
-			msgClient := msggateway.NewMsgGatewayClient(v)
-			_, err := msgClient.MultiTerminalLoginCheck(ctx, &msggateway.MultiTerminalLoginCheckReq{
-				UserID:     client.UserID,
-				PlatformID: int32(client.PlatformID), Token: client.token,
-			})
-			if err != nil {
-				log.ZWarn(ctx, "MultiTerminalLoginCheck err", err, "node", v.Target())
-			}
-			return nil
+		msgClient := msggateway.NewMsgGatewayClient(v)
+		_, err := msgClient.MultiTerminalLoginCheck(ctx, &msggateway.MultiTerminalLoginCheckReq{
+			UserID:     client.UserID,
+			PlatformID: int32(client.PlatformID), Token: client.token,
 		})
+		if err != nil {
+			log.ZWarn(ctx, "MultiTerminalLoginCheck err", err, "node", v.Target())
+			continue
+		}
 	}
-
-	_ = wg.Wait()
 	return nil
 }
 
@@ -224,7 +221,7 @@ func (ws *WsServer) registerClient(client *Client) {
 	if !userOK {
 		ws.clients.Set(client.UserID, client)
 		log.ZDebug(client.ctx, "user not exist", "userID", client.UserID, "platformID", client.PlatformID)
-		prommetrics.OnlineUserGauge.Add(1)
+		prom_metrics.OnlineUserGauge.Add(1)
 		ws.onlineUserNum.Add(1)
 		ws.onlineUserConnNum.Add(1)
 	} else {
@@ -292,72 +289,70 @@ func (ws *WsServer) multiTerminalLoginChecker(clientOK bool, oldClients []*Clien
 		}
 		fallthrough
 	case constant.AllLoginButSameTermKick:
-		if !clientOK {
-			return
-		}
+		if clientOK {
+			isDeleteUser := ws.clients.deleteClients(newClient.UserID, oldClients)
+			if isDeleteUser {
+				ws.onlineUserNum.Add(-1)
+			}
+			for _, c := range oldClients {
+				err := c.KickOnlineMessage()
+				if err != nil {
+					log.ZWarn(c.ctx, "KickOnlineMessage", err)
+				}
+			}
+			m, err := ws.cache.GetTokensWithoutError(
+				newClient.ctx,
+				newClient.UserID,
+				newClient.PlatformID,
+			)
+			if err != nil && err != redis.Nil {
+				log.ZWarn(
+					newClient.ctx,
+					"get token from redis err",
+					err,
+					"userID",
+					newClient.UserID,
+					"platformID",
+					newClient.PlatformID,
+				)
+				return
+			}
+			if m == nil {
+				log.ZWarn(
+					newClient.ctx,
+					"m is nil",
+					errors.New("m is nil"),
+					"userID",
+					newClient.UserID,
+					"platformID",
+					newClient.PlatformID,
+				)
+				return
+			}
+			log.ZDebug(
+				newClient.ctx,
+				"get token from redis",
+				"userID",
+				newClient.UserID,
+				"platformID",
+				newClient.PlatformID,
+				"tokenMap",
+				m,
+			)
 
-		isDeleteUser := ws.clients.deleteClients(newClient.UserID, oldClients)
-		if isDeleteUser {
-			ws.onlineUserNum.Add(-1)
-		}
-		for _, c := range oldClients {
-			err := c.KickOnlineMessage()
+			for k := range m {
+				if k != newClient.ctx.GetToken() {
+					m[k] = constant.KickedToken
+				}
+			}
+			log.ZDebug(newClient.ctx, "set token map is ", "token map", m, "userID",
+				newClient.UserID, "token", newClient.ctx.GetToken())
+			err = ws.cache.SetTokenMapByUidPid(newClient.ctx, newClient.UserID, newClient.PlatformID, m)
 			if err != nil {
-				log.ZWarn(c.ctx, "KickOnlineMessage", err)
+				log.ZWarn(newClient.ctx, "SetTokenMapByUidPid err", err, "userID", newClient.UserID, "platformID", newClient.PlatformID)
+				return
 			}
 		}
-		m, err := ws.cache.GetTokensWithoutError(
-			newClient.ctx,
-			newClient.UserID,
-			newClient.PlatformID,
-		)
-		if err != nil && err != redis.Nil {
-			log.ZWarn(
-				newClient.ctx,
-				"get token from redis err",
-				err,
-				"userID",
-				newClient.UserID,
-				"platformID",
-				newClient.PlatformID,
-			)
-			return
-		}
-		if m == nil {
-			log.ZWarn(
-				newClient.ctx,
-				"m is nil",
-				errors.New("m is nil"),
-				"userID",
-				newClient.UserID,
-				"platformID",
-				newClient.PlatformID,
-			)
-			return
-		}
-		log.ZDebug(
-			newClient.ctx,
-			"get token from redis",
-			"userID",
-			newClient.UserID,
-			"platformID",
-			newClient.PlatformID,
-			"tokenMap",
-			m,
-		)
-
-		for k := range m {
-			if k != newClient.ctx.GetToken() {
-				m[k] = constant.KickedToken
-			}
-		}
-		log.ZDebug(newClient.ctx, "set token map is ", "token map", m, "userID",
-			newClient.UserID, "token", newClient.ctx.GetToken())
-		err = ws.cache.SetTokenMapByUidPid(newClient.ctx, newClient.UserID, newClient.PlatformID, m)
-		if err != nil {
-			log.ZWarn(newClient.ctx, "SetTokenMapByUidPid err", err, "userID", newClient.UserID, "platformID", newClient.PlatformID)
-			return
-		}
 	}
 }
 
@@ -366,11 +361,11 @@ func (ws *WsServer) unregisterClient(client *Client) {
 	isDeleteUser := ws.clients.delete(client.UserID, client.ctx.GetRemoteAddr())
 	if isDeleteUser {
 		ws.onlineUserNum.Add(-1)
-		prommetrics.OnlineUserGauge.Dec()
+		prom_metrics.OnlineUserGauge.Dec()
 	}
 	ws.onlineUserConnNum.Add(-1)
 	ws.SetUserOnlineStatus(client.ctx, client, constant.Offline)
-	log.ZInfo(client.ctx, "user offline", "close reason", client.closedErr, "online user Num", ws.onlineUserNum.Load(), "online user conn Num",
+	log.ZInfo(client.ctx, "user offline", "close reason", client.closedErr, "online user Num", ws.onlineUserNum, "online user conn Num",
 		ws.onlineUserConnNum.Load(),
 	)
 }
@@ -409,7 +404,7 @@ func (ws *WsServer) wsHandler(w http.ResponseWriter, r *http.Request) {
 		httpError(connContext, errs.ErrConnArgsErr)
 		return
 	}
-	if err = authverify.WsVerifyToken(token, userID, platformID); err != nil {
+	if err := authverify.WsVerifyToken(token, userID, platformID); err != nil {
 		httpError(connContext, err)
 		return
 	}
@@ -432,8 +427,7 @@ func (ws *WsServer) wsHandler(w http.ResponseWriter, r *http.Request) {
 		httpError(connContext, errs.ErrTokenNotExist.Wrap())
 		return
 	}
-
-	wsLongConn := newGWebSocket(WebSocket, ws.handshakeTimeout, ws.writeBufferSize)
+	wsLongConn := newGWebSocket(WebSocket, ws.handshakeTimeout)
 	err = wsLongConn.GenerateLongConn(w, r)
 	if err != nil {
 		httpError(connContext, err)

internal/msggateway/options.go

@@ -27,8 +27,6 @@ type (
 		handshakeTimeout time.Duration
 		// 允许消息最大长度
 		messageMaxMsgLength int
-		// websocket write buffer, default: 4096, 4kb.
-		writeBufferSize int
 	}
 )
 
@@ -55,9 +53,3 @@ func WithMessageMaxMsgLength(length int) Option {
 		opt.messageMaxMsgLength = length
 	}
 }
-
-func WithWriteBufferSize(size int) Option {
-	return func(opt *configs) {
-		opt.writeBufferSize = size
-	}
-}

internal/msgtransfer/init.go

@@ -17,16 +17,18 @@ package msgtransfer
 import (
 	"errors"
 	"fmt"
-	"log"
-	"net/http"
-	"sync"
-
-	"github.com/OpenIMSDK/tools/mw"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/discovery_register"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/prom_metrics"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/collectors"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
+	"log"
+	"net/http"
+	"sync"
+
+	"github.com/OpenIMSDK/tools/mw"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
@@ -34,8 +36,6 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/relation"
 	relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/unrelation"
-	kdisc "github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
 )
 
@@ -65,7 +65,7 @@ func StartTransfer(prometheusPort int) error {
 	if err := mongo.CreateMsgIndex(); err != nil {
 		return err
 	}
-	client, err := kdisc.NewDiscoveryRegister(config.Config.Envs.Discovery)
+	client, err := discovery_register.NewDiscoveryRegister(config.Config.Envs.Discovery)
 	/*
 		client, err := openkeeper.NewClient(config.Config.Zookeeper.ZkAddr, config.Config.Zookeeper.Schema,
 			openkeeper.WithFreq(time.Hour), openkeeper.WithRoundRobin(), openkeeper.WithUserNameAndPassword(config.Config.Zookeeper.Username,
@@ -123,7 +123,7 @@ func (m *MsgTransfer) Start(prometheusPort int) error {
 		reg.MustRegister(
 			collectors.NewGoCollector(),
 		)
-		reg.MustRegister(prommetrics.GetGrpcCusMetrics("Transfer")...)
+		reg.MustRegister(prom_metrics.GetGrpcCusMetrics("Transfer")...)
 		http.Handle("/metrics", promhttp.HandlerFor(reg, promhttp.HandlerOpts{Registry: reg}))
 		log.Fatal(http.ListenAndServe(fmt.Sprintf(":%d", prometheusPort), nil))
 	}

internal/msgtransfer/online_history_msg_handler.go

@@ -427,62 +427,49 @@ func (och *OnlineHistoryRedisConsumerHandler) ConsumeClaim(
 			break
 		}
 	}
+	rwLock := new(sync.RWMutex)
 	log.ZDebug(context.Background(), "online new session msg come", "highWaterMarkOffset",
 		claim.HighWaterMarkOffset(), "topic", claim.Topic(), "partition", claim.Partition())
-
-	split := 1000
-	rwLock := new(sync.RWMutex)
-	messages := make([]*sarama.ConsumerMessage, 0, 1000)
-	ticker := time.NewTicker(time.Millisecond * 100)
-
+	cMsg := make([]*sarama.ConsumerMessage, 0, 1000)
+	t := time.NewTicker(time.Millisecond * 100)
 	go func() {
 		for {
 			select {
-			case <-ticker.C:
-				if len(messages) == 0 {
-					continue
+			case <-t.C:
+				if len(cMsg) > 0 {
+					rwLock.Lock()
+					ccMsg := make([]*sarama.ConsumerMessage, 0, 1000)
+					for _, v := range cMsg {
+						ccMsg = append(ccMsg, v)
+					}
+					cMsg = make([]*sarama.ConsumerMessage, 0, 1000)
+					rwLock.Unlock()
+					split := 1000
+					ctx := mcontext.WithTriggerIDContext(context.Background(), utils.OperationIDGenerator())
+					log.ZDebug(ctx, "timer trigger msg consumer start", "length", len(ccMsg))
+					for i := 0; i < len(ccMsg)/split; i++ {
+						// log.Debug()
+						och.msgDistributionCh <- Cmd2Value{Cmd: ConsumerMsgs, Value: TriggerChannelValue{
+							ctx: ctx, cMsgList: ccMsg[i*split : (i+1)*split],
+						}}
+					}
+					if (len(ccMsg) % split) > 0 {
+						och.msgDistributionCh <- Cmd2Value{Cmd: ConsumerMsgs, Value: TriggerChannelValue{
+							ctx: ctx, cMsgList: ccMsg[split*(len(ccMsg)/split):],
+						}}
+					}
+					log.ZDebug(ctx, "timer trigger msg consumer end", "length", len(ccMsg))
 				}
-
-				rwLock.Lock()
-				buffer := make([]*sarama.ConsumerMessage, 0, len(messages))
-				buffer = append(buffer, messages...)
-
-				// reuse slice, set cap to 0
-				messages = messages[:0]
-				rwLock.Unlock()
-
-				start := time.Now()
-				ctx := mcontext.WithTriggerIDContext(context.Background(), utils.OperationIDGenerator())
-				log.ZDebug(ctx, "timer trigger msg consumer start", "length", len(buffer))
-				for i := 0; i < len(buffer)/split; i++ {
-					och.msgDistributionCh <- Cmd2Value{Cmd: ConsumerMsgs, Value: TriggerChannelValue{
-						ctx: ctx, cMsgList: buffer[i*split : (i+1)*split],
-					}}
-				}
-				if (len(buffer) % split) > 0 {
-					och.msgDistributionCh <- Cmd2Value{Cmd: ConsumerMsgs, Value: TriggerChannelValue{
-						ctx: ctx, cMsgList: buffer[split*(len(buffer)/split):],
-					}}
-				}
-
-				log.ZDebug(ctx, "timer trigger msg consumer end",
-					"length", len(buffer), "time_cost", time.Since(start),
-				)
 			}
 		}
 	}()
-
 	for msg := range claim.Messages() {
-		if len(msg.Value) == 0 {
-			continue
-		}
-
 		rwLock.Lock()
-		messages = append(messages, msg)
+		if len(msg.Value) != 0 {
+			cMsg = append(cMsg, msg)
+		}
 		rwLock.Unlock()
-
 		sess.MarkMessage(msg, "")
 	}
-
 	return nil
 }

internal/msgtransfer/online_msg_to_mongo_handler.go

@@ -16,6 +16,7 @@ package msgtransfer
 
 import (
 	"context"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/prom_metrics"
 
 	"github.com/IBM/sarama"
 	"google.golang.org/protobuf/proto"
@@ -26,7 +27,6 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/controller"
 	kfk "github.com/openimsdk/open-im-server/v3/pkg/common/kafka"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 )
 
 type OnlineHistoryMongoConsumerHandler struct {
@@ -75,9 +75,9 @@ func (mc *OnlineHistoryMongoConsumerHandler) handleChatWs2Mongo(
 			"conversationID",
 			msgFromMQ.ConversationID,
 		)
-		prommetrics.MsgInsertMongoFailedCounter.Inc()
+		prom_metrics.MsgInsertMongoFailedCounter.Inc()
 	} else {
-		prommetrics.MsgInsertMongoSuccessCounter.Inc()
+		prom_metrics.MsgInsertMongoSuccessCounter.Inc()
 	}
 	var seqs []int64
 	for _, msg := range msgFromMQ.MsgData {

internal/push/offlinepush/dummy/push.go

@@ -2,7 +2,6 @@ package dummy
 
 import (
 	"context"
-
 	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush"
 )
 

internal/push/offlinepush/fcm/push_test.go

@@ -0,0 +1,32 @@
+// Copyright © 2023 OpenIM. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package fcm
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
+)
+
+func Test_Push(t *testing.T) {
+	var redis cache.MsgModel
+	offlinePusher := NewClient(redis)
+	err := offlinePusher.Push(context.Background(), []string{"userID1"}, "test", "test", &offlinepush.Opts{})
+	assert.Nil(t, err)
+}

internal/push/push_to_client.go

@@ -18,12 +18,14 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
-	"sync"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/prom_metrics"
+	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/dummy"
 
-	"golang.org/x/sync/errgroup"
+	"github.com/OpenIMSDK/protocol/conversation"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
 
 	"github.com/OpenIMSDK/protocol/constant"
-	"github.com/OpenIMSDK/protocol/conversation"
 	"github.com/OpenIMSDK/protocol/msggateway"
 	"github.com/OpenIMSDK/protocol/sdkws"
 	"github.com/OpenIMSDK/tools/discoveryregistry"
@@ -32,7 +34,6 @@ import (
 	"github.com/OpenIMSDK/tools/utils"
 
 	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush"
-	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/dummy"
 	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/fcm"
 	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/getui"
 	"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/jpush"
@@ -40,8 +41,6 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/controller"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/localcache"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
-	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
 )
 
@@ -54,6 +53,7 @@ type Pusher struct {
 	msgRpcClient           *rpcclient.MessageRpcClient
 	conversationRpcClient  *rpcclient.ConversationRpcClient
 	groupRpcClient         *rpcclient.GroupRpcClient
+	successCount           int
 }
 
 var errNoOfflinePusher = errors.New("no offlinePusher is configured")
@@ -104,29 +104,24 @@ func (p *Pusher) Push2User(ctx context.Context, userIDs []string, msg *sdkws.Msg
 	if err := callbackOnlinePush(ctx, userIDs, msg); err != nil {
 		return err
 	}
-
 	// push
 	wsResults, err := p.GetConnsAndOnlinePush(ctx, msg, userIDs)
 	if err != nil {
 		return err
 	}
-
 	isOfflinePush := utils.GetSwitchFromOptions(msg.Options, constant.IsOfflinePush)
 	log.ZDebug(ctx, "push_result", "ws push result", wsResults, "sendData", msg, "isOfflinePush", isOfflinePush, "push_to_userID", userIDs)
-
-	if !isOfflinePush {
-		return nil
-	}
-
-	for _, v := range wsResults {
-		if msg.SendID != v.UserID && (!v.OnlinePush) {
-			if err = callbackOfflinePush(ctx, userIDs, msg, &[]string{}); err != nil {
-				return err
-			}
-
-			err = p.offlinePushMsg(ctx, msg.SendID, msg, []string{v.UserID})
-			if err != nil {
-				return err
+	p.successCount++
+	if isOfflinePush {
+		for _, v := range wsResults {
+			if msg.SendID != v.UserID && (!v.OnlinePush) {
+				if err := callbackOfflinePush(ctx, userIDs, msg, &[]string{}); err != nil {
+					return err
+				}
+				err = p.offlinePushMsg(ctx, msg.SendID, msg, []string{v.UserID})
+				if err != nil {
+					return err
+				}
 			}
 		}
 	}
@@ -145,16 +140,14 @@ func (p *Pusher) UnmarshalNotificationElem(bytes []byte, t interface{}) error {
 func (p *Pusher) Push2SuperGroup(ctx context.Context, groupID string, msg *sdkws.MsgData) (err error) {
 	log.ZDebug(ctx, "Get super group msg from msg_transfer and push msg", "msg", msg.String(), "groupID", groupID)
 	var pushToUserIDs []string
-	if err = callbackBeforeSuperGroupOnlinePush(ctx, groupID, msg, &pushToUserIDs); err != nil {
+	if err := callbackBeforeSuperGroupOnlinePush(ctx, groupID, msg, &pushToUserIDs); err != nil {
 		return err
 	}
-
 	if len(pushToUserIDs) == 0 {
 		pushToUserIDs, err = p.groupLocalCache.GetGroupMemberIDs(ctx, groupID)
 		if err != nil {
 			return err
 		}
-
 		switch msg.ContentType {
 		case constant.MemberQuitNotification:
 			var tips sdkws.MemberQuitTips
@@ -162,7 +155,7 @@ func (p *Pusher) Push2SuperGroup(ctx context.Context, groupID string, msg *sdkws
 				return err
 			}
 			defer func(groupID string, userIDs []string) {
-				if err = p.DeleteMemberAndSetConversationSeq(ctx, groupID, userIDs); err != nil {
+				if err := p.DeleteMemberAndSetConversationSeq(ctx, groupID, userIDs); err != nil {
 					log.ZError(ctx, "MemberQuitNotification DeleteMemberAndSetConversationSeq", err, "groupID", groupID, "userIDs", userIDs)
 				}
 			}(groupID, []string{tips.QuitUser.UserID})
@@ -174,7 +167,7 @@ func (p *Pusher) Push2SuperGroup(ctx context.Context, groupID string, msg *sdkws
 			}
 			kickedUsers := utils.Slice(tips.KickedUserList, func(e *sdkws.GroupMemberFullInfo) string { return e.UserID })
 			defer func(groupID string, userIDs []string) {
-				if err = p.DeleteMemberAndSetConversationSeq(ctx, groupID, userIDs); err != nil {
+				if err := p.DeleteMemberAndSetConversationSeq(ctx, groupID, userIDs); err != nil {
 					log.ZError(ctx, "MemberKickedNotification DeleteMemberAndSetConversationSeq", err, "groupID", groupID, "userIDs", userIDs)
 				}
 			}(groupID, kickedUsers)
@@ -190,61 +183,48 @@ func (p *Pusher) Push2SuperGroup(ctx context.Context, groupID string, msg *sdkws
 					ctx = mcontext.WithOpUserIDContext(ctx, config.Config.Manager.UserID[0])
 				}
 				defer func(groupID string) {
-					if err = p.groupRpcClient.DismissGroup(ctx, groupID); err != nil {
+					if err := p.groupRpcClient.DismissGroup(ctx, groupID); err != nil {
 						log.ZError(ctx, "DismissGroup Notification clear members", err, "groupID", groupID)
 					}
 				}(groupID)
 			}
 		}
 	}
-
 	wsResults, err := p.GetConnsAndOnlinePush(ctx, msg, pushToUserIDs)
 	if err != nil {
 		return err
 	}
-
 	log.ZDebug(ctx, "get conn and online push success", "result", wsResults, "msg", msg)
+	p.successCount++
 	isOfflinePush := utils.GetSwitchFromOptions(msg.Options, constant.IsOfflinePush)
 	if isOfflinePush {
-		var (
-			onlineSuccessUserIDs      = []string{msg.SendID}
-			webAndPcBackgroundUserIDs []string
-		)
-
+		var onlineSuccessUserIDs []string
+		var WebAndPcBackgroundUserIDs []string
+		onlineSuccessUserIDs = append(onlineSuccessUserIDs, msg.SendID)
 		for _, v := range wsResults {
 			if v.OnlinePush && v.UserID != msg.SendID {
 				onlineSuccessUserIDs = append(onlineSuccessUserIDs, v.UserID)
 			}
-
-			if v.OnlinePush {
-				continue
-			}
-
-			if len(v.Resp) == 0 {
-				continue
-			}
-
-			for _, singleResult := range v.Resp {
-				if singleResult.ResultCode != -2 {
-					continue
-				}
-
-				isPC := constant.PlatformIDToName(int(singleResult.RecvPlatFormID)) == constant.TerminalPC
-				isWebID := singleResult.RecvPlatFormID == constant.WebPlatformID
-
-				if isPC || isWebID {
-					webAndPcBackgroundUserIDs = append(webAndPcBackgroundUserIDs, v.UserID)
+			if !v.OnlinePush {
+				if len(v.Resp) != 0 {
+					for _, singleResult := range v.Resp {
+						if singleResult.ResultCode == -2 {
+							if constant.PlatformIDToName(int(singleResult.RecvPlatFormID)) == constant.TerminalPC ||
+								singleResult.RecvPlatFormID == constant.WebPlatformID {
+								WebAndPcBackgroundUserIDs = append(WebAndPcBackgroundUserIDs, v.UserID)
+							}
+						}
+					}
 				}
 			}
 		}
-
 		needOfflinePushUserIDs := utils.DifferenceString(onlineSuccessUserIDs, pushToUserIDs)
 		if msg.ContentType != constant.SignalingNotification {
 			notNotificationUserIDs, err := p.conversationLocalCache.GetRecvMsgNotNotifyUserIDs(ctx, groupID)
 			if err != nil {
+				// log.ZError(ctx, "GetRecvMsgNotNotifyUserIDs failed", err, "groupID", groupID)
 				return err
 			}
-
 			needOfflinePushUserIDs = utils.SliceSub(needOfflinePushUserIDs, notNotificationUserIDs)
 		}
 		// Use offline push messaging
@@ -254,7 +234,6 @@ func (p *Pusher) Push2SuperGroup(ctx context.Context, groupID string, msg *sdkws
 			if err != nil {
 				return err
 			}
-
 			if len(offlinePushUserIDs) > 0 {
 				needOfflinePushUserIDs = offlinePushUserIDs
 			}
@@ -271,8 +250,8 @@ func (p *Pusher) Push2SuperGroup(ctx context.Context, groupID string, msg *sdkws
 					log.ZError(ctx, "offlinePushMsg failed", err, "groupID", groupID, "msg", msg)
 					return err
 				}
-				if _, err := p.GetConnsAndOnlinePush(ctx, msg, utils.IntersectString(resp.UserIDs, webAndPcBackgroundUserIDs)); err != nil {
-					log.ZError(ctx, "offlinePushMsg failed", err, "groupID", groupID, "msg", msg, "userIDs", utils.IntersectString(needOfflinePushUserIDs, webAndPcBackgroundUserIDs))
+				if _, err := p.GetConnsAndOnlinePush(ctx, msg, utils.IntersectString(resp.UserIDs, WebAndPcBackgroundUserIDs)); err != nil {
+					log.ZError(ctx, "offlinePushMsg failed", err, "groupID", groupID, "msg", msg, "userIDs", utils.IntersectString(needOfflinePushUserIDs, WebAndPcBackgroundUserIDs))
 					return err
 				}
 			}
@@ -287,44 +266,18 @@ func (p *Pusher) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.MsgData,
 	if err != nil {
 		return nil, err
 	}
-
-	var (
-		mu         sync.Mutex
-		wg         = errgroup.Group{}
-		input      = &msggateway.OnlineBatchPushOneMsgReq{MsgData: msg, PushToUserIDs: pushToUserIDs}
-		maxWorkers = config.Config.Push.MaxConcurrentWorkers
-	)
-
-	if maxWorkers < 3 {
-		maxWorkers = 3
-	}
-
-	wg.SetLimit(maxWorkers)
-
 	// Online push message
-	for _, conn := range conns {
-		conn := conn // loop var safe
-		wg.Go(func() error {
-			msgClient := msggateway.NewMsgGatewayClient(conn)
-			reply, err := msgClient.SuperGroupOnlineBatchPushOneMsg(ctx, input)
-			if err != nil {
-				return nil
-			}
-
-			log.ZDebug(ctx, "push result", "reply", reply)
-			if reply != nil && reply.SinglePushResult != nil {
-				mu.Lock()
-				wsResults = append(wsResults, reply.SinglePushResult...)
-				mu.Unlock()
-			}
-
-			return nil
-		})
+	for _, v := range conns {
+		msgClient := msggateway.NewMsgGatewayClient(v)
+		reply, err := msgClient.SuperGroupOnlineBatchPushOneMsg(ctx, &msggateway.OnlineBatchPushOneMsgReq{MsgData: msg, PushToUserIDs: pushToUserIDs})
+		if err != nil {
+			continue
+		}
+		log.ZDebug(ctx, "push result", "reply", reply)
+		if reply != nil && reply.SinglePushResult != nil {
+			wsResults = append(wsResults, reply.SinglePushResult...)
+		}
 	}
-
-	_ = wg.Wait()
-
-	// always return nil
 	return wsResults, nil
 }
 
@@ -335,7 +288,7 @@ func (p *Pusher) offlinePushMsg(ctx context.Context, conversationID string, msg
 	}
 	err = p.offlinePusher.Push(ctx, offlinePushUserIDs, title, content, opts)
 	if err != nil {
-		prommetrics.MsgOfflinePushFailedCounter.Inc()
+		prom_metrics.MsgOfflinePushFailedCounter.Inc()
 		return err
 	}
 	return nil
@@ -366,18 +319,15 @@ func (p *Pusher) getOfflinePushInfos(conversationID string, msg *sdkws.MsgData)
 		err = errNoOfflinePusher
 		return
 	}
-
-	type atContent struct {
+	type AtContent struct {
 		Text       string   `json:"text"`
 		AtUserList []string `json:"atUserList"`
 		IsAtSelf   bool     `json:"isAtSelf"`
 	}
-
 	opts, err = p.GetOfflinePushOpts(msg)
 	if err != nil {
 		return
 	}
-
 	if msg.OfflinePushInfo != nil {
 		title = msg.OfflinePushInfo.Title
 		content = msg.OfflinePushInfo.Desc
@@ -395,9 +345,9 @@ func (p *Pusher) getOfflinePushInfos(conversationID string, msg *sdkws.MsgData)
 		case constant.File:
 			title = constant.ContentType2PushContent[int64(msg.ContentType)]
 		case constant.AtText:
-			ac := atContent{}
-			_ = utils.JsonStringToStruct(string(msg.Content), &ac)
-			if utils.IsContain(conversationID, ac.AtUserList) {
+			a := AtContent{}
+			_ = utils.JsonStringToStruct(string(msg.Content), &a)
+			if utils.IsContain(conversationID, a.AtUserList) {
 				title = constant.ContentType2PushContent[constant.AtText] + constant.ContentType2PushContent[constant.Common]
 			} else {
 				title = constant.ContentType2PushContent[constant.GroupMsg]

internal/rpc/auth/auth.go

@@ -16,6 +16,7 @@ package auth
 
 import (
 	"context"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/prom_metrics"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
 
@@ -34,7 +35,6 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/controller"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
 )
 
@@ -74,7 +74,7 @@ func (s *authServer) UserToken(ctx context.Context, req *pbauth.UserTokenReq) (*
 	if err != nil {
 		return nil, err
 	}
-	prommetrics.UserLoginCounter.Inc()
+	prom_metrics.UserLoginCounter.Inc()
 	resp.Token = token
 	resp.ExpireTimeSeconds = config.Config.TokenPolicy.Expire * 24 * 60 * 60
 	return &resp, nil

internal/rpc/group/group.go

@@ -690,11 +690,7 @@ func (s *groupServer) GetGroupApplicationList(ctx context.Context, req *pbgroup.
 		return e.GroupID
 	})
 	resp.GroupRequests = utils.Slice(groupRequests, func(e *relationtb.GroupRequestModel) *sdkws.GroupRequest {
-		var ownerUserID string
-		if owner, ok := ownerMap[e.GroupID]; ok {
-			ownerUserID = owner.UserID
-		}
-		return convert.Db2PbGroupRequest(e, userMap[e.UserID], convert.Db2PbGroupInfo(groupMap[e.GroupID], ownerUserID, groupMemberNumMap[e.GroupID]))
+		return convert.Db2PbGroupRequest(e, userMap[e.UserID], convert.Db2PbGroupInfo(groupMap[e.GroupID], ownerMap[e.GroupID].UserID, groupMemberNumMap[e.GroupID]))
 	})
 	return resp, nil
 }
@@ -1060,20 +1056,16 @@ func (s *groupServer) GetGroups(ctx context.Context, req *pbgroup.GetGroupsReq)
 	ownerMemberMap := utils.SliceToMap(ownerMembers, func(e *relationtb.GroupMemberModel) string {
 		return e.GroupID
 	})
+	if ids := utils.Single(groupIDs, utils.Keys(ownerMemberMap)); len(ids) > 0 {
+		return nil, errs.ErrDatabase.Wrap("group not owner " + strings.Join(ids, ","))
+	}
 	groupMemberNumMap, err := s.GroupDatabase.MapGroupMemberNum(ctx, groupIDs)
 	if err != nil {
 		return nil, err
 	}
 	resp.Groups = utils.Slice(groups, func(group *relationtb.GroupModel) *pbgroup.CMSGroup {
-		var (
-			userID   string
-			username string
-		)
-		if member, ok := ownerMemberMap[group.GroupID]; ok {
-			userID = member.UserID
-			username = member.Nickname
-		}
-		return convert.Db2PbCMSGroup(group, userID, username, groupMemberNumMap[group.GroupID])
+		member := ownerMemberMap[group.GroupID]
+		return convert.Db2PbCMSGroup(group, member.UserID, member.Nickname, uint32(groupMemberNumMap[group.GroupID]))
 	})
 	return resp, nil
 }
@@ -1127,13 +1119,16 @@ func (s *groupServer) GetUserReqApplicationList(ctx context.Context, req *pbgrou
 	groupIDs := utils.Distinct(utils.Slice(requests, func(e *relationtb.GroupRequestModel) string {
 		return e.GroupID
 	}))
-	groups, err := s.GroupDatabase.FindGroup(ctx, groupIDs)
+	groups, err := s.GroupDatabase.FindNotDismissedGroup(ctx, groupIDs)
 	if err != nil {
 		return nil, err
 	}
 	groupMap := utils.SliceToMap(groups, func(e *relationtb.GroupModel) string {
 		return e.GroupID
 	})
+	if ids := utils.Single(groupIDs, utils.Keys(groupMap)); len(ids) > 0 {
+		return nil, errs.ErrGroupIDNotFound.Wrap(strings.Join(ids, ","))
+	}
 	owners, err := s.FindGroupMember(ctx, groupIDs, nil, []int32{constant.GroupOwner})
 	if err != nil {
 		return nil, err
@@ -1141,16 +1136,15 @@ func (s *groupServer) GetUserReqApplicationList(ctx context.Context, req *pbgrou
 	ownerMap := utils.SliceToMap(owners, func(e *relationtb.GroupMemberModel) string {
 		return e.GroupID
 	})
+	if ids := utils.Single(groupIDs, utils.Keys(ownerMap)); len(ids) > 0 {
+		return nil, errs.ErrData.Wrap("group no owner", strings.Join(ids, ","))
+	}
 	groupMemberNum, err := s.GroupDatabase.MapGroupMemberNum(ctx, groupIDs)
 	if err != nil {
 		return nil, err
 	}
 	resp.GroupRequests = utils.Slice(requests, func(e *relationtb.GroupRequestModel) *sdkws.GroupRequest {
-		var ownerUserID string
-		if owner, ok := ownerMap[e.GroupID]; ok {
-			ownerUserID = owner.UserID
-		}
-		return convert.Db2PbGroupRequest(e, user, convert.Db2PbGroupInfo(groupMap[e.GroupID], ownerUserID, groupMemberNum[e.GroupID]))
+		return convert.Db2PbGroupRequest(e, user, convert.Db2PbGroupInfo(groupMap[e.GroupID], ownerMap[e.GroupID].UserID, uint32(groupMemberNum[e.GroupID])))
 	})
 	return resp, nil
 }
@@ -1569,16 +1563,15 @@ func (s *groupServer) GetGroupUsersReqApplicationList(ctx context.Context, req *
 	ownerMap := utils.SliceToMap(owners, func(e *relationtb.GroupMemberModel) string {
 		return e.GroupID
 	})
+	if ids := utils.Single(groupIDs, utils.Keys(ownerMap)); len(ids) > 0 {
+		return nil, errs.ErrData.Wrap("group no owner", strings.Join(ids, ","))
+	}
 	groupMemberNum, err := s.GroupDatabase.MapGroupMemberNum(ctx, groupIDs)
 	if err != nil {
 		return nil, err
 	}
 	resp.GroupRequests = utils.Slice(requests, func(e *relationtb.GroupRequestModel) *sdkws.GroupRequest {
-		var ownerUserID string
-		if owner, ok := ownerMap[e.GroupID]; ok {
-			ownerUserID = owner.UserID
-		}
-		return convert.Db2PbGroupRequest(e, nil, convert.Db2PbGroupInfo(groupMap[e.GroupID], ownerUserID, groupMemberNum[e.GroupID]))
+		return convert.Db2PbGroupRequest(e, nil, convert.Db2PbGroupInfo(groupMap[e.GroupID], ownerMap[e.GroupID].UserID, uint32(groupMemberNum[e.GroupID])))
 	})
 	resp.Total = total
 	return resp, nil

internal/rpc/msg/send.go

@@ -16,8 +16,8 @@ package msg
 
 import (
 	"context"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/prom_metrics"
 
-	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
 
 	"github.com/OpenIMSDK/protocol/constant"
@@ -59,7 +59,7 @@ func (m *msgServer) sendMsgSuperGroupChat(
 	req *pbmsg.SendMsgReq,
 ) (resp *pbmsg.SendMsgResp, err error) {
 	if err = m.messageVerification(ctx, req); err != nil {
-		prommetrics.GroupChatMsgProcessFailedCounter.Inc()
+		prom_metrics.GroupChatMsgProcessFailedCounter.Inc()
 		return nil, err
 	}
 	if err = callbackBeforeSendGroupMsg(ctx, req); err != nil {
@@ -78,7 +78,7 @@ func (m *msgServer) sendMsgSuperGroupChat(
 	if err = callbackAfterSendGroupMsg(ctx, req); err != nil {
 		log.ZWarn(ctx, "CallbackAfterSendGroupMsg", err)
 	}
-	prommetrics.GroupChatMsgProcessSuccessCounter.Inc()
+	prom_metrics.GroupChatMsgProcessSuccessCounter.Inc()
 	resp = &pbmsg.SendMsgResp{}
 	resp.SendTime = req.MsgData.SendTime
 	resp.ServerMsgID = req.MsgData.ServerMsgID
@@ -161,7 +161,7 @@ func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *pbmsg.SendMsgReq
 		}
 	}
 	if !isSend {
-		prommetrics.SingleChatMsgProcessFailedCounter.Inc()
+		prom_metrics.SingleChatMsgProcessFailedCounter.Inc()
 		return nil, nil
 	} else {
 		if err = callbackBeforeSendSingleMsg(ctx, req); err != nil {
@@ -171,7 +171,7 @@ func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *pbmsg.SendMsgReq
 			return nil, err
 		}
 		if err := m.MsgDatabase.MsgToMQ(ctx, utils.GenConversationUniqueKeyForSingle(req.MsgData.SendID, req.MsgData.RecvID), req.MsgData); err != nil {
-			prommetrics.SingleChatMsgProcessFailedCounter.Inc()
+			prom_metrics.SingleChatMsgProcessFailedCounter.Inc()
 			return nil, err
 		}
 		err = callbackAfterSendSingleMsg(ctx, req)
@@ -183,7 +183,7 @@ func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *pbmsg.SendMsgReq
 			ClientMsgID: req.MsgData.ClientMsgID,
 			SendTime:    req.MsgData.SendTime,
 		}
-		prommetrics.SingleChatMsgProcessSuccessCounter.Inc()
+		prom_metrics.SingleChatMsgProcessSuccessCounter.Inc()
 		return resp, nil
 	}
 }

internal/tools/cron_task.go

@@ -17,18 +17,13 @@ package tools
 import (
 	"context"
 	"fmt"
-	"os"
-	"os/signal"
-	"syscall"
-	"time"
+	"sync"
 
-	"github.com/redis/go-redis/v9"
 	"github.com/robfig/cron/v3"
 
 	"github.com/OpenIMSDK/tools/log"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
 )
 
 func StartTask() error {
@@ -37,75 +32,23 @@ func StartTask() error {
 	if err != nil {
 		return err
 	}
-
-	msgTool.convertTools()
-
-	rdb, err := cache.NewRedis()
-	if err != nil {
-		return err
-	}
-
-	// register cron tasks
-	var crontab = cron.New()
+	msgTool.ConvertTools()
+	c := cron.New()
+	var wg sync.WaitGroup
+	wg.Add(1)
 	log.ZInfo(context.Background(), "start chatRecordsClearTime cron task", "cron config", config.Config.ChatRecordsClearTime)
-	_, err = crontab.AddFunc(config.Config.ChatRecordsClearTime, cronWrapFunc(rdb, "cron_clear_msg_and_fix_seq", msgTool.AllConversationClearMsgAndFixSeq))
+	_, err = c.AddFunc(config.Config.ChatRecordsClearTime, msgTool.AllConversationClearMsgAndFixSeq)
 	if err != nil {
 		log.ZError(context.Background(), "start allConversationClearMsgAndFixSeq cron failed", err)
 		panic(err)
 	}
-
 	log.ZInfo(context.Background(), "start msgDestruct cron task", "cron config", config.Config.MsgDestructTime)
-	_, err = crontab.AddFunc(config.Config.MsgDestructTime, cronWrapFunc(rdb, "cron_conversations_destruct_msgs", msgTool.ConversationsDestructMsgs))
+	_, err = c.AddFunc(config.Config.MsgDestructTime, msgTool.ConversationsDestructMsgs)
 	if err != nil {
 		log.ZError(context.Background(), "start conversationsDestructMsgs cron failed", err)
 		panic(err)
 	}
-
-	// start crontab
-	crontab.Start()
-
-	sigs := make(chan os.Signal, 1)
-	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
-	<-sigs
-
-	// stop crontab, Wait for the running task to exit.
-	ctx := crontab.Stop()
-
-	select {
-	case <-ctx.Done():
-		// graceful exit
-
-	case <-time.After(15 * time.Second):
-		// forced exit on timeout
-	}
-
+	c.Start()
+	wg.Wait()
 	return nil
 }
-
-// netlock redis lock.
-func netlock(rdb redis.UniversalClient, key string, ttl time.Duration) bool {
-	value := "used"
-	ok, err := rdb.SetNX(context.Background(), key, value, ttl).Result() // nolint
-	if err != nil {
-		// when err is about redis server, return true.
-		return false
-	}
-
-	return ok
-}
-
-func cronWrapFunc(rdb redis.UniversalClient, key string, fn func()) func() {
-	enableCronLocker := config.Config.EnableCronLocker
-	return func() {
-		// if don't enable cron-locker, call fn directly.
-		if !enableCronLocker {
-			fn()
-			return
-		}
-
-		// when acquire redis lock, call fn().
-		if netlock(rdb, key, 5*time.Second) {
-			fn()
-		}
-	}
-}

internal/tools/cron_task_test.go

@@ -1,82 +0,0 @@
-package tools
-
-import (
-	"fmt"
-	"math/rand"
-	"sync"
-	"testing"
-	"time"
-
-	"github.com/redis/go-redis/v9"
-	"github.com/robfig/cron/v3"
-	"github.com/stretchr/testify/assert"
-
-	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
-)
-
-func TestDisLock(t *testing.T) {
-	rdb := redis.NewClient(&redis.Options{})
-	defer rdb.Close()
-
-	assert.Equal(t, true, netlock(rdb, "cron-1", 1*time.Second))
-
-	// if exists, get false
-	assert.Equal(t, false, netlock(rdb, "cron-1", 1*time.Second))
-
-	time.Sleep(2 * time.Second)
-
-	// wait for key on timeout, get true
-	assert.Equal(t, true, netlock(rdb, "cron-1", 2*time.Second))
-
-	// set different key
-	assert.Equal(t, true, netlock(rdb, "cron-2", 2*time.Second))
-}
-
-func TestCronWrapFunc(t *testing.T) {
-	rdb := redis.NewClient(&redis.Options{})
-	defer rdb.Close()
-
-	once := sync.Once{}
-	done := make(chan struct{}, 1)
-	cb := func() {
-		once.Do(func() {
-			close(done)
-		})
-	}
-
-	start := time.Now()
-	key := fmt.Sprintf("cron-%v", rand.Int31())
-	crontab := cron.New(cron.WithSeconds())
-	crontab.AddFunc("*/1 * * * * *", cronWrapFunc(rdb, key, cb))
-	crontab.Start()
-	<-done
-
-	dur := time.Since(start)
-	assert.LessOrEqual(t, dur.Seconds(), float64(2*time.Second))
-	crontab.Stop()
-}
-
-func TestCronWrapFuncWithNetlock(t *testing.T) {
-	config.Config.EnableCronLocker = true
-	rdb := redis.NewClient(&redis.Options{})
-	defer rdb.Close()
-
-	done := make(chan string, 10)
-
-	crontab := cron.New(cron.WithSeconds())
-
-	key := fmt.Sprintf("cron-%v", rand.Int31())
-	crontab.AddFunc("*/1 * * * * *", cronWrapFunc(rdb, key, func() {
-		done <- "host1"
-	}))
-	crontab.AddFunc("*/1 * * * * *", cronWrapFunc(rdb, key, func() {
-		done <- "host2"
-	}))
-	crontab.Start()
-
-	time.Sleep(12 * time.Second)
-	// the ttl of netlock is 5s, so expected value is 2.
-	assert.Equal(t, len(done), 2)
-
-	crontab.Stop()
-}

internal/tools/msg.go

@@ -23,7 +23,7 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 
-	kdisc "github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/discovery_register"
 
 	"math/rand"
 
@@ -76,7 +76,7 @@ func InitMsgTool() (*MsgTool, error) {
 	if err != nil {
 		return nil, err
 	}
-	discov, err := kdisc.NewDiscoveryRegister(config.Config.Envs.Discovery)
+	discov, err := discovery_register.NewDiscoveryRegister(config.Config.Envs.Discovery)
 	/*
 		discov, err := zookeeper.NewClient(config.Config.Zookeeper.ZkAddr, config.Config.Zookeeper.Schema,
 			zookeeper.WithFreq(time.Hour), zookeeper.WithRoundRobin(), zookeeper.WithUserNameAndPassword(config.Config.Zookeeper.Username,

internal/tools/msg_doc_convert.go

@@ -22,7 +22,7 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
 )
 
-func (c *MsgTool) convertTools() {
+func (c *MsgTool) ConvertTools() {
 	ctx := mcontext.NewCtx("convert")
 	conversationIDs, err := c.conversationDatabase.GetAllConversationIDs(ctx)
 	if err != nil {

pkg/apistruct/manage.go

@@ -22,37 +22,37 @@ import (
 type SendMsg struct {
 	// SendID uniquely identifies the sender.
 	SendID string `json:"sendID" binding:"required"`
-
+	
 	// GroupID is the identifier for the group, required if SessionType is 2 or 3.
 	GroupID string `json:"groupID" binding:"required_if=SessionType 2|required_if=SessionType 3"`
-
+	
 	// SenderNickname is the nickname of the sender.
 	SenderNickname string `json:"senderNickname"`
-
+	
 	// SenderFaceURL is the URL to the sender's avatar.
 	SenderFaceURL string `json:"senderFaceURL"`
-
+	
 	// SenderPlatformID is an integer identifier for the sender's platform.
 	SenderPlatformID int32 `json:"senderPlatformID"`
-
+	
 	// Content is the actual content of the message, required and excluded from Swagger documentation.
 	Content map[string]interface{} `json:"content" binding:"required" swaggerignore:"true"`
-
+	
 	// ContentType is an integer that represents the type of the content.
 	ContentType int32 `json:"contentType" binding:"required"`
-
+	
 	// SessionType is an integer that represents the type of session for the message.
 	SessionType int32 `json:"sessionType" binding:"required"`
-
+	
 	// IsOnlineOnly specifies if the message is only sent when the receiver is online.
 	IsOnlineOnly bool `json:"isOnlineOnly"`
-
+	
 	// NotOfflinePush specifies if the message should not trigger offline push notifications.
 	NotOfflinePush bool `json:"notOfflinePush"`
-
+	
 	// SendTime is a timestamp indicating when the message was sent.
 	SendTime int64 `json:"sendTime"`
-
+	
 	// OfflinePushInfo contains information for offline push notifications.
 	OfflinePushInfo *sdkws.OfflinePushInfo `json:"offlinePushInfo"`
 }
@@ -67,10 +67,10 @@ type SendMsgReq struct {
 // BatchSendMsgReq defines the structure for sending a message to multiple recipients.
 type BatchSendMsgReq struct {
 	SendMsg
-
+	
 	// IsSendAll indicates whether the message should be sent to all users.
 	IsSendAll bool `json:"isSendAll"`
-
+	
 	// RecvIDs is a slice of receiver identifiers to whom the message will be sent, required field.
 	RecvIDs []string `json:"recvIDs" binding:"required"`
 }
@@ -79,7 +79,7 @@ type BatchSendMsgReq struct {
 type BatchSendMsgResp struct {
 	// Results is a slice of SingleReturnResult, representing the outcome of each message sent.
 	Results []*SingleReturnResult `json:"results"`
-
+	
 	// FailedIDs is a slice of user IDs for whom the message send failed.
 	FailedIDs []string `json:"failedUserIDs"`
 }
@@ -88,13 +88,13 @@ type BatchSendMsgResp struct {
 type SingleReturnResult struct {
 	// ServerMsgID is the message identifier on the server-side.
 	ServerMsgID string `json:"serverMsgID"`
-
+	
 	// ClientMsgID is the message identifier on the client-side.
 	ClientMsgID string `json:"clientMsgID"`
-
+	
 	// SendTime is the timestamp of when the message was sent.
 	SendTime int64 `json:"sendTime"`
-
+	
 	// RecvID uniquely identifies the receiver of the message.
 	RecvID string `json:"recvID"`
 }

pkg/common/cmd/msg_transfer.go

@@ -16,11 +16,9 @@ package cmd
 
 import (
 	"fmt"
-
 	"github.com/OpenIMSDK/protocol/constant"
-	"github.com/spf13/cobra"
-
 	config2 "github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"github.com/spf13/cobra"
 
 	"github.com/openimsdk/open-im-server/v3/internal/msgtransfer"
 )

pkg/common/cmd/root.go

@@ -20,7 +20,6 @@ import (
 	config2 "github.com/openimsdk/open-im-server/v3/pkg/common/config"
 
 	"github.com/spf13/cobra"
-	_ "go.uber.org/automaxprocs"
 
 	"github.com/OpenIMSDK/protocol/constant"
 	"github.com/OpenIMSDK/tools/log"

pkg/common/config/config.go

@@ -78,20 +78,17 @@ type configStruct struct {
 	} `yaml:"mongo"`
 
 	Redis struct {
-		ClusterMode    bool     `yaml:"clusterMode"`
-		Address        []string `yaml:"address"`
-		Username       string   `yaml:"username"`
-		Password       string   `yaml:"password"`
-		EnablePipeline bool     `yaml:"enablePipeline"`
+		ClusterMode bool     `yaml:"clusterMode"`
+		Address     []string `yaml:"address"`
+		Username    string   `yaml:"username"`
+		Password    string   `yaml:"password"`
 	} `yaml:"redis"`
 
 	Kafka struct {
-		Username     string   `yaml:"username"`
-		Password     string   `yaml:"password"`
-		ProducerAck  string   `yaml:"producerAck"`
-		CompressType string   `yaml:"compressType"`
-		Addr         []string `yaml:"addr"`
-		TLS          *struct {
+		Username string   `yaml:"username"`
+		Password string   `yaml:"password"`
+		Addr     []string `yaml:"addr"`
+		TLS      *struct {
 			CACrt              string `yaml:"caCrt"`
 			ClientCrt          string `yaml:"clientCrt"`
 			ClientKey          string `yaml:"clientKey"`
@@ -196,13 +193,11 @@ type configStruct struct {
 		WebsocketMaxConnNum      int   `yaml:"websocketMaxConnNum"`
 		WebsocketMaxMsgLen       int   `yaml:"websocketMaxMsgLen"`
 		WebsocketTimeout         int   `yaml:"websocketTimeout"`
-		WebsocketWriteBufferSize int   `yaml:"websocketWriteBufferSize"`
 	} `yaml:"longConnSvr"`
 
 	Push struct {
-		MaxConcurrentWorkers int    `yaml:"maxConcurrentWorkers"`
-		Enable               string `yaml:"enable"`
-		GeTui                struct {
+		Enable string `yaml:"enable"`
+		GeTui  struct {
 			PushUrl      string `yaml:"pushUrl"`
 			AppKey       string `yaml:"appKey"`
 			Intent       string `yaml:"intent"`
@@ -234,7 +229,6 @@ type configStruct struct {
 	ChatRecordsClearTime              string `yaml:"chatRecordsClearTime"`
 	MsgDestructTime                   string `yaml:"msgDestructTime"`
 	Secret                            string `yaml:"secret"`
-	EnableCronLocker                  bool   `yaml:"enableCronLocker"`
 	TokenPolicy                       struct {
 		Expire int64 `yaml:"expire"`
 	} `yaml:"tokenPolicy"`

pkg/common/db/cache/init_redis.go

@@ -28,21 +28,12 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 )
 
-var (
-	// singleton pattern.
-	redisClient redis.UniversalClient
-)
-
 const (
 	maxRetry = 10 // number of retries
 )
 
 // NewRedis Initialize redis connection.
 func NewRedis() (redis.UniversalClient, error) {
-	if redisClient != nil {
-		return redisClient, nil
-	}
-
 	if len(config.Config.Redis.Address) == 0 {
 		return nil, errors.New("redis address is empty")
 	}
@@ -75,6 +66,5 @@ func NewRedis() (redis.UniversalClient, error) {
 		return nil, fmt.Errorf("redis ping %w", err)
 	}
 
-	redisClient = rdb
 	return rdb, err
 }

pkg/common/db/cache/msg.go

@@ -20,7 +20,9 @@ import (
 	"strconv"
 	"time"
 
-	"golang.org/x/sync/errgroup"
+	"github.com/dtm-labs/rockscache"
+
+	unrelationtb "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/unrelation"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
 
@@ -60,8 +62,6 @@ const (
 	uidPidToken             = "UID_PID_TOKEN_STATUS:"
 )
 
-var concurrentLimit = 3
-
 type SeqCache interface {
 	SetMaxSeq(ctx context.Context, conversationID string, maxSeq int64) error
 	GetMaxSeqs(ctx context.Context, conversationIDs []string) (map[string]int64, error)
@@ -133,7 +133,10 @@ func NewMsgCacheModel(client redis.UniversalClient) MsgModel {
 
 type msgCache struct {
 	metaCache
-	rdb redis.UniversalClient
+	rdb            redis.UniversalClient
+	expireTime     time.Duration
+	rcClient       *rockscache.Client
+	msgDocDatabase unrelationtb.MsgDocModelInterface
 }
 
 func (c *msgCache) getMaxSeqKey(conversationID string) string {
@@ -170,6 +173,29 @@ func (c *msgCache) getSeqs(ctx context.Context, items []string, getkey func(s st
 	}
 
 	return m, nil
+
+	//pipe := c.rdb.Pipeline()
+	//for _, v := range items {
+	//	if err := pipe.Get(ctx, getkey(v)).Err(); err != nil && err != redis.Nil {
+	//		return nil, errs.Wrap(err)
+	//	}
+	//}
+	//result, err := pipe.Exec(ctx)
+	//if err != nil && err != redis.Nil {
+	//	return nil, errs.Wrap(err)
+	//}
+	//m = make(map[string]int64, len(items))
+	//for i, v := range result {
+	//	seq := v.(*redis.StringCmd)
+	//	if seq.Err() != nil && seq.Err() != redis.Nil {
+	//		return nil, errs.Wrap(v.Err())
+	//	}
+	//	val := utils.StringToInt64(seq.Val())
+	//	if val != 0 {
+	//		m[items[i]] = val
+	//	}
+	//}
+	//return m, nil
 }
 
 func (c *msgCache) SetMaxSeq(ctx context.Context, conversationID string, maxSeq int64) error {
@@ -195,6 +221,15 @@ func (c *msgCache) setSeqs(ctx context.Context, seqs map[string]int64, getkey fu
 		}
 	}
 	return nil
+	//pipe := c.rdb.Pipeline()
+	//for k, seq := range seqs {
+	//	err := pipe.Set(ctx, getkey(k), seq, 0).Err()
+	//	if err != nil {
+	//		return errs.Wrap(err)
+	//	}
+	//}
+	//_, err := pipe.Exec(ctx)
+	//return err
 }
 
 func (c *msgCache) SetMinSeqs(ctx context.Context, seqs map[string]int64) error {
@@ -310,165 +345,85 @@ func (c *msgCache) allMessageCacheKey(conversationID string) string {
 }
 
 func (c *msgCache) GetMessagesBySeq(ctx context.Context, conversationID string, seqs []int64) (seqMsgs []*sdkws.MsgData, failedSeqs []int64, err error) {
-	if config.Config.Redis.EnablePipeline {
-		return c.PipeGetMessagesBySeq(ctx, conversationID, seqs)
-	}
-
-	return c.ParallelGetMessagesBySeq(ctx, conversationID, seqs)
-}
-
-func (c *msgCache) PipeGetMessagesBySeq(ctx context.Context, conversationID string, seqs []int64) (seqMsgs []*sdkws.MsgData, failedSeqs []int64, err error) {
-	pipe := c.rdb.Pipeline()
-
-	results := []*redis.StringCmd{}
 	for _, seq := range seqs {
-		results = append(results, pipe.Get(ctx, c.getMessageCacheKey(conversationID, seq)))
-	}
-
-	_, err = pipe.Exec(ctx)
-	if err != nil && err != redis.Nil {
-		return seqMsgs, failedSeqs, errs.Wrap(err, "pipe.get")
-	}
-
-	for idx, res := range results {
-		seq := seqs[idx]
-		if res.Err() != nil {
-			log.ZError(ctx, "GetMessagesBySeq failed", err, "conversationID", conversationID, "seq", seq, "err", res.Err())
+		res, err := c.rdb.Get(ctx, c.getMessageCacheKey(conversationID, seq)).Result()
+		if err != nil {
+			log.ZError(ctx, "GetMessagesBySeq failed", err, "conversationID", conversationID, "seq", seq)
 			failedSeqs = append(failedSeqs, seq)
 			continue
 		}
-
 		msg := sdkws.MsgData{}
-		if err = msgprocessor.String2Pb(res.Val(), &msg); err != nil {
+		if err = msgprocessor.String2Pb(res, &msg); err != nil {
 			log.ZError(ctx, "GetMessagesBySeq Unmarshal failed", err, "res", res, "conversationID", conversationID, "seq", seq)
 			failedSeqs = append(failedSeqs, seq)
 			continue
 		}
-
 		if msg.Status == constant.MsgDeleted {
 			failedSeqs = append(failedSeqs, seq)
 			continue
 		}
-
 		seqMsgs = append(seqMsgs, &msg)
 	}
 
 	return
-}
-
-func (c *msgCache) ParallelGetMessagesBySeq(ctx context.Context, conversationID string, seqs []int64) (seqMsgs []*sdkws.MsgData, failedSeqs []int64, err error) {
-	type entry struct {
-		err error
-		msg *sdkws.MsgData
-	}
-
-	wg := errgroup.Group{}
-	wg.SetLimit(concurrentLimit)
-
-	results := make([]entry, len(seqs)) // set slice len/cap to length of seqs.
-	for idx, seq := range seqs {
-		// closure safe var
-		idx := idx
-		seq := seq
-
-		wg.Go(func() error {
-			res, err := c.rdb.Get(ctx, c.getMessageCacheKey(conversationID, seq)).Result()
-			if err != nil {
-				log.ZError(ctx, "GetMessagesBySeq failed", err, "conversationID", conversationID, "seq", seq)
-				results[idx] = entry{err: err}
-				return nil
-			}
-
-			msg := sdkws.MsgData{}
-			if err = msgprocessor.String2Pb(res, &msg); err != nil {
-				log.ZError(ctx, "GetMessagesBySeq Unmarshal failed", err, "res", res, "conversationID", conversationID, "seq", seq)
-				results[idx] = entry{err: err}
-				return nil
-			}
-
-			if msg.Status == constant.MsgDeleted {
-				results[idx] = entry{err: err}
-				return nil
-			}
-
-			results[idx] = entry{msg: &msg}
-			return nil
-		})
-	}
-
-	_ = wg.Wait()
-
-	for idx, res := range results {
-		if res.err != nil {
-			failedSeqs = append(failedSeqs, seqs[idx])
-			continue
-		}
-
-		seqMsgs = append(seqMsgs, res.msg)
-	}
-
-	return
+	//pipe := c.rdb.Pipeline()
+	//for _, v := range seqs {
+	//	// MESSAGE_CACHE:169.254.225.224_reliability1653387820_0_1
+	//	key := c.getMessageCacheKey(conversationID, v)
+	//	if err := pipe.Get(ctx, key).Err(); err != nil && err != redis.Nil {
+	//		return nil, nil, err
+	//	}
+	//}
+	//result, err := pipe.Exec(ctx)
+	//for i, v := range result {
+	//	cmd := v.(*redis.StringCmd)
+	//	if cmd.Err() != nil {
+	//		failedSeqs = append(failedSeqs, seqs[i])
+	//	} else {
+	//		msg := sdkws.MsgData{}
+	//		err = msgprocessor.String2Pb(cmd.Val(), &msg)
+	//		if err == nil {
+	//			if msg.Status != constant.MsgDeleted {
+	//				seqMsgs = append(seqMsgs, &msg)
+	//				continue
+	//			}
+	//		} else {
+	//			log.ZWarn(ctx, "UnmarshalString failed", err, "conversationID", conversationID, "seq", seqs[i], "msg", cmd.Val())
+	//		}
+	//		failedSeqs = append(failedSeqs, seqs[i])
+	//	}
+	//}
+	//return seqMsgs, failedSeqs, err
 }
 
 func (c *msgCache) SetMessageToCache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (int, error) {
-	if config.Config.Redis.EnablePipeline {
-		return c.PipeSetMessageToCache(ctx, conversationID, msgs)
-	}
-	return c.ParallelSetMessageToCache(ctx, conversationID, msgs)
-}
-
-func (c *msgCache) PipeSetMessageToCache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (int, error) {
-	pipe := c.rdb.Pipeline()
 	for _, msg := range msgs {
 		s, err := msgprocessor.Pb2String(msg)
 		if err != nil {
-			return 0, errs.Wrap(err, "pb.marshal")
+			return 0, errs.Wrap(err)
 		}
-
 		key := c.getMessageCacheKey(conversationID, msg.Seq)
-		_ = pipe.Set(ctx, key, s, time.Duration(config.Config.MsgCacheTimeout)*time.Second)
-	}
-
-	results, err := pipe.Exec(ctx)
-	if err != nil {
-		return 0, errs.Wrap(err, "pipe.set")
-	}
-
-	for _, res := range results {
-		if res.Err() != nil {
+		if err := c.rdb.Set(ctx, key, s, time.Duration(config.Config.MsgCacheTimeout)*time.Second).Err(); err != nil {
 			return 0, errs.Wrap(err)
 		}
 	}
-
-	return len(msgs), nil
-}
-
-func (c *msgCache) ParallelSetMessageToCache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (int, error) {
-	wg := errgroup.Group{}
-	wg.SetLimit(concurrentLimit)
-
-	for _, msg := range msgs {
-		msg := msg // closure safe var
-		wg.Go(func() error {
-			s, err := msgprocessor.Pb2String(msg)
-			if err != nil {
-				return errs.Wrap(err)
-			}
-
-			key := c.getMessageCacheKey(conversationID, msg.Seq)
-			if err := c.rdb.Set(ctx, key, s, time.Duration(config.Config.MsgCacheTimeout)*time.Second).Err(); err != nil {
-				return errs.Wrap(err)
-			}
-			return nil
-		})
-	}
-
-	err := wg.Wait()
-	if err != nil {
-		return 0, err
-	}
-
 	return len(msgs), nil
+	//pipe := c.rdb.Pipeline()
+	//var failedMsgs []*sdkws.MsgData
+	//for _, msg := range msgs {
+	//	key := c.getMessageCacheKey(conversationID, msg.Seq)
+	//	s, err := msgprocessor.Pb2String(msg)
+	//	if err != nil {
+	//		return 0, errs.Wrap(err)
+	//	}
+	//	err = pipe.Set(ctx, key, s, time.Duration(config.Config.MsgCacheTimeout)*time.Second).Err()
+	//	if err != nil {
+	//		failedMsgs = append(failedMsgs, msg)
+	//		log.ZWarn(ctx, "set msg 2 cache failed", err, "msg", failedMsgs)
+	//	}
+	//}
+	//_, err := pipe.Exec(ctx)
+	//return len(failedMsgs), err
 }
 
 func (c *msgCache) getMessageDelUserListKey(conversationID string, seq int64) string {
@@ -599,49 +554,20 @@ func (c *msgCache) DelUserDeleteMsgsList(ctx context.Context, conversationID str
 }
 
 func (c *msgCache) DeleteMessages(ctx context.Context, conversationID string, seqs []int64) error {
-	if config.Config.Redis.EnablePipeline {
-		return c.PipeDeleteMessages(ctx, conversationID, seqs)
-	}
-
-	return c.ParallelDeleteMessages(ctx, conversationID, seqs)
-}
-
-func (c *msgCache) ParallelDeleteMessages(ctx context.Context, conversationID string, seqs []int64) error {
-	wg := errgroup.Group{}
-	wg.SetLimit(concurrentLimit)
-
 	for _, seq := range seqs {
-		seq := seq
-		wg.Go(func() error {
-			err := c.rdb.Del(ctx, c.getMessageCacheKey(conversationID, seq)).Err()
-			if err != nil {
-				return errs.Wrap(err)
-			}
-			return nil
-		})
-	}
-
-	return wg.Wait()
-}
-
-func (c *msgCache) PipeDeleteMessages(ctx context.Context, conversationID string, seqs []int64) error {
-	pipe := c.rdb.Pipeline()
-	for _, seq := range seqs {
-		_ = pipe.Del(ctx, c.getMessageCacheKey(conversationID, seq))
-	}
-
-	results, err := pipe.Exec(ctx)
-	if err != nil {
-		return errs.Wrap(err, "pipe.del")
-	}
-
-	for _, res := range results {
-		if res.Err() != nil {
+		if err := c.rdb.Del(ctx, c.getMessageCacheKey(conversationID, seq)).Err(); err != nil {
 			return errs.Wrap(err)
 		}
 	}
-
 	return nil
+	//pipe := c.rdb.Pipeline()
+	//for _, seq := range seqs {
+	//	if err := pipe.Del(ctx, c.getMessageCacheKey(conversationID, seq)).Err(); err != nil {
+	//		return errs.Wrap(err)
+	//	}
+	//}
+	//_, err := pipe.Exec(ctx)
+	//return errs.Wrap(err)
 }
 
 func (c *msgCache) CleanUpOneConversationAllMsg(ctx context.Context, conversationID string) error {
@@ -658,6 +584,14 @@ func (c *msgCache) CleanUpOneConversationAllMsg(ctx context.Context, conversatio
 		}
 	}
 	return nil
+	//pipe := c.rdb.Pipeline()
+	//for _, v := range vals {
+	//	if err := pipe.Del(ctx, v).Err(); err != nil {
+	//		return errs.Wrap(err)
+	//	}
+	//}
+	//_, err = pipe.Exec(ctx)
+	//return errs.Wrap(err)
 }
 
 func (c *msgCache) DelMsgFromCache(ctx context.Context, userID string, seqs []int64) error {

pkg/common/db/cache/msg_test.go

@@ -1,387 +0,0 @@
-package cache
-
-import (
-	"context"
-	"fmt"
-	"math/rand"
-	"testing"
-
-	"github.com/OpenIMSDK/protocol/sdkws"
-	"github.com/redis/go-redis/v9"
-	"github.com/stretchr/testify/assert"
-)
-
-func TestParallelSetMessageToCache(t *testing.T) {
-	var (
-		cid      = fmt.Sprintf("cid-%v", rand.Int63())
-		seqFirst = rand.Int63()
-		msgs     = []*sdkws.MsgData{}
-	)
-
-	for i := 0; i < 100; i++ {
-		msgs = append(msgs, &sdkws.MsgData{
-			Seq: seqFirst + int64(i),
-		})
-	}
-
-	testParallelSetMessageToCache(t, cid, msgs)
-}
-
-func testParallelSetMessageToCache(t *testing.T, cid string, msgs []*sdkws.MsgData) {
-	rdb := redis.NewClient(&redis.Options{})
-	defer rdb.Close()
-
-	cacher := msgCache{rdb: rdb}
-
-	ret, err := cacher.ParallelSetMessageToCache(context.Background(), cid, msgs)
-	assert.Nil(t, err)
-	assert.Equal(t, len(msgs), ret)
-
-	// validate
-	for _, msg := range msgs {
-		key := cacher.getMessageCacheKey(cid, msg.Seq)
-		val, err := rdb.Exists(context.Background(), key).Result()
-		assert.Nil(t, err)
-		assert.EqualValues(t, 1, val)
-	}
-}
-
-func TestPipeSetMessageToCache(t *testing.T) {
-	var (
-		cid      = fmt.Sprintf("cid-%v", rand.Int63())
-		seqFirst = rand.Int63()
-		msgs     = []*sdkws.MsgData{}
-	)
-
-	for i := 0; i < 100; i++ {
-		msgs = append(msgs, &sdkws.MsgData{
-			Seq: seqFirst + int64(i),
-		})
-	}
-
-	testPipeSetMessageToCache(t, cid, msgs)
-}
-
-func testPipeSetMessageToCache(t *testing.T, cid string, msgs []*sdkws.MsgData) {
-	rdb := redis.NewClient(&redis.Options{})
-	defer rdb.Close()
-
-	cacher := msgCache{rdb: rdb}
-
-	ret, err := cacher.PipeSetMessageToCache(context.Background(), cid, msgs)
-	assert.Nil(t, err)
-	assert.Equal(t, len(msgs), ret)
-
-	// validate
-	for _, msg := range msgs {
-		key := cacher.getMessageCacheKey(cid, msg.Seq)
-		val, err := rdb.Exists(context.Background(), key).Result()
-		assert.Nil(t, err)
-		assert.EqualValues(t, 1, val)
-	}
-}
-
-func TestGetMessagesBySeq(t *testing.T) {
-	var (
-		cid      = fmt.Sprintf("cid-%v", rand.Int63())
-		seqFirst = rand.Int63()
-		msgs     = []*sdkws.MsgData{}
-	)
-
-	seqs := []int64{}
-	for i := 0; i < 100; i++ {
-		msgs = append(msgs, &sdkws.MsgData{
-			Seq:    seqFirst + int64(i),
-			SendID: fmt.Sprintf("fake-sendid-%v", i),
-		})
-		seqs = append(seqs, seqFirst+int64(i))
-	}
-
-	// set data to cache
-	testPipeSetMessageToCache(t, cid, msgs)
-
-	// get data from cache with parallet mode
-	testParallelGetMessagesBySeq(t, cid, seqs, msgs)
-
-	// get data from cache with pipeline mode
-	testPipeGetMessagesBySeq(t, cid, seqs, msgs)
-}
-
-func testParallelGetMessagesBySeq(t *testing.T, cid string, seqs []int64, inputMsgs []*sdkws.MsgData) {
-	rdb := redis.NewClient(&redis.Options{})
-	defer rdb.Close()
-
-	cacher := msgCache{rdb: rdb}
-
-	respMsgs, failedSeqs, err := cacher.ParallelGetMessagesBySeq(context.Background(), cid, seqs)
-	assert.Nil(t, err)
-	assert.Equal(t, 0, len(failedSeqs))
-	assert.Equal(t, len(respMsgs), len(seqs))
-
-	// validate
-	for idx, msg := range respMsgs {
-		assert.Equal(t, msg.Seq, inputMsgs[idx].Seq)
-		assert.Equal(t, msg.SendID, inputMsgs[idx].SendID)
-	}
-}
-
-func testPipeGetMessagesBySeq(t *testing.T, cid string, seqs []int64, inputMsgs []*sdkws.MsgData) {
-	rdb := redis.NewClient(&redis.Options{})
-	defer rdb.Close()
-
-	cacher := msgCache{rdb: rdb}
-
-	respMsgs, failedSeqs, err := cacher.PipeGetMessagesBySeq(context.Background(), cid, seqs)
-	assert.Nil(t, err)
-	assert.Equal(t, 0, len(failedSeqs))
-	assert.Equal(t, len(respMsgs), len(seqs))
-
-	// validate
-	for idx, msg := range respMsgs {
-		assert.Equal(t, msg.Seq, inputMsgs[idx].Seq)
-		assert.Equal(t, msg.SendID, inputMsgs[idx].SendID)
-	}
-}
-
-func TestGetMessagesBySeqWithEmptySeqs(t *testing.T) {
-	var (
-		cid            = fmt.Sprintf("cid-%v", rand.Int63())
-		seqFirst int64 = 0
-		msgs           = []*sdkws.MsgData{}
-	)
-
-	seqs := []int64{}
-	for i := 0; i < 100; i++ {
-		msgs = append(msgs, &sdkws.MsgData{
-			Seq:    seqFirst + int64(i),
-			SendID: fmt.Sprintf("fake-sendid-%v", i),
-		})
-		seqs = append(seqs, seqFirst+int64(i))
-	}
-
-	// don't set cache, only get data from cache.
-
-	// get data from cache with parallet mode
-	testParallelGetMessagesBySeqWithEmptry(t, cid, seqs, msgs)
-
-	// get data from cache with pipeline mode
-	testPipeGetMessagesBySeqWithEmptry(t, cid, seqs, msgs)
-}
-
-func testParallelGetMessagesBySeqWithEmptry(t *testing.T, cid string, seqs []int64, inputMsgs []*sdkws.MsgData) {
-	rdb := redis.NewClient(&redis.Options{})
-	defer rdb.Close()
-
-	cacher := msgCache{rdb: rdb}
-
-	respMsgs, failedSeqs, err := cacher.ParallelGetMessagesBySeq(context.Background(), cid, seqs)
-	assert.Nil(t, err)
-	assert.Equal(t, len(seqs), len(failedSeqs))
-	assert.Equal(t, 0, len(respMsgs))
-}
-
-func testPipeGetMessagesBySeqWithEmptry(t *testing.T, cid string, seqs []int64, inputMsgs []*sdkws.MsgData) {
-	rdb := redis.NewClient(&redis.Options{})
-	defer rdb.Close()
-
-	cacher := msgCache{rdb: rdb}
-
-	respMsgs, failedSeqs, err := cacher.PipeGetMessagesBySeq(context.Background(), cid, seqs)
-	assert.Equal(t, err, redis.Nil)
-	assert.Equal(t, len(seqs), len(failedSeqs))
-	assert.Equal(t, 0, len(respMsgs))
-}
-
-func TestGetMessagesBySeqWithLostHalfSeqs(t *testing.T) {
-	var (
-		cid            = fmt.Sprintf("cid-%v", rand.Int63())
-		seqFirst int64 = 0
-		msgs           = []*sdkws.MsgData{}
-	)
-
-	seqs := []int64{}
-	for i := 0; i < 100; i++ {
-		msgs = append(msgs, &sdkws.MsgData{
-			Seq:    seqFirst + int64(i),
-			SendID: fmt.Sprintf("fake-sendid-%v", i),
-		})
-		seqs = append(seqs, seqFirst+int64(i))
-	}
-
-	// Only set half the number of messages.
-	testParallelSetMessageToCache(t, cid, msgs[:50])
-
-	// get data from cache with parallet mode
-	testParallelGetMessagesBySeqWithLostHalfSeqs(t, cid, seqs, msgs)
-
-	// get data from cache with pipeline mode
-	testPipeGetMessagesBySeqWithLostHalfSeqs(t, cid, seqs, msgs)
-}
-
-func testParallelGetMessagesBySeqWithLostHalfSeqs(t *testing.T, cid string, seqs []int64, inputMsgs []*sdkws.MsgData) {
-	rdb := redis.NewClient(&redis.Options{})
-	defer rdb.Close()
-
-	cacher := msgCache{rdb: rdb}
-
-	respMsgs, failedSeqs, err := cacher.ParallelGetMessagesBySeq(context.Background(), cid, seqs)
-	assert.Nil(t, err)
-	assert.Equal(t, len(seqs)/2, len(failedSeqs))
-	assert.Equal(t, len(seqs)/2, len(respMsgs))
-
-	for idx, msg := range respMsgs {
-		assert.Equal(t, msg.Seq, seqs[idx])
-	}
-}
-
-func testPipeGetMessagesBySeqWithLostHalfSeqs(t *testing.T, cid string, seqs []int64, inputMsgs []*sdkws.MsgData) {
-	rdb := redis.NewClient(&redis.Options{})
-	defer rdb.Close()
-
-	cacher := msgCache{rdb: rdb}
-
-	respMsgs, failedSeqs, err := cacher.PipeGetMessagesBySeq(context.Background(), cid, seqs)
-	assert.Nil(t, err)
-	assert.Equal(t, len(seqs)/2, len(failedSeqs))
-	assert.Equal(t, len(seqs)/2, len(respMsgs))
-
-	for idx, msg := range respMsgs {
-		assert.Equal(t, msg.Seq, seqs[idx])
-	}
-}
-
-func TestPipeDeleteMessages(t *testing.T) {
-	var (
-		cid      = fmt.Sprintf("cid-%v", rand.Int63())
-		seqFirst = rand.Int63()
-		msgs     = []*sdkws.MsgData{}
-	)
-
-	var seqs []int64
-	for i := 0; i < 100; i++ {
-		msgs = append(msgs, &sdkws.MsgData{
-			Seq: seqFirst + int64(i),
-		})
-		seqs = append(seqs, msgs[i].Seq)
-	}
-
-	testPipeSetMessageToCache(t, cid, msgs)
-	testPipeDeleteMessagesOK(t, cid, seqs, msgs)
-
-	// set again
-	testPipeSetMessageToCache(t, cid, msgs)
-	testPipeDeleteMessagesMix(t, cid, seqs[:90], msgs)
-}
-
-func testPipeDeleteMessagesOK(t *testing.T, cid string, seqs []int64, inputMsgs []*sdkws.MsgData) {
-	rdb := redis.NewClient(&redis.Options{})
-	defer rdb.Close()
-
-	cacher := msgCache{rdb: rdb}
-
-	err := cacher.PipeDeleteMessages(context.Background(), cid, seqs)
-	assert.Nil(t, err)
-
-	// validate
-	for _, msg := range inputMsgs {
-		key := cacher.getMessageCacheKey(cid, msg.Seq)
-		val := rdb.Exists(context.Background(), key).Val()
-		assert.EqualValues(t, 0, val)
-	}
-}
-
-func testPipeDeleteMessagesMix(t *testing.T, cid string, seqs []int64, inputMsgs []*sdkws.MsgData) {
-	rdb := redis.NewClient(&redis.Options{})
-	defer rdb.Close()
-
-	cacher := msgCache{rdb: rdb}
-
-	err := cacher.PipeDeleteMessages(context.Background(), cid, seqs)
-	assert.Nil(t, err)
-
-	// validate
-	for idx, msg := range inputMsgs {
-		key := cacher.getMessageCacheKey(cid, msg.Seq)
-		val, err := rdb.Exists(context.Background(), key).Result()
-		assert.Nil(t, err)
-		if idx < 90 {
-			assert.EqualValues(t, 0, val) // not exists
-			continue
-		}
-
-		assert.EqualValues(t, 1, val) // exists
-	}
-}
-
-func TestParallelDeleteMessages(t *testing.T) {
-	var (
-		cid      = fmt.Sprintf("cid-%v", rand.Int63())
-		seqFirst = rand.Int63()
-		msgs     = []*sdkws.MsgData{}
-	)
-
-	var seqs []int64
-	for i := 0; i < 100; i++ {
-		msgs = append(msgs, &sdkws.MsgData{
-			Seq: seqFirst + int64(i),
-		})
-		seqs = append(seqs, msgs[i].Seq)
-	}
-
-	randSeqs := []int64{}
-	for i := seqFirst + 100; i < seqFirst+200; i++ {
-		randSeqs = append(randSeqs, i)
-	}
-
-	testParallelSetMessageToCache(t, cid, msgs)
-	testParallelDeleteMessagesOK(t, cid, seqs, msgs)
-
-	// set again
-	testParallelSetMessageToCache(t, cid, msgs)
-	testParallelDeleteMessagesMix(t, cid, seqs[:90], msgs, 90)
-	testParallelDeleteMessagesOK(t, cid, seqs[90:], msgs[:90])
-
-	// set again
-	testParallelSetMessageToCache(t, cid, msgs)
-	testParallelDeleteMessagesMix(t, cid, randSeqs, msgs, 0)
-}
-
-func testParallelDeleteMessagesOK(t *testing.T, cid string, seqs []int64, inputMsgs []*sdkws.MsgData) {
-	rdb := redis.NewClient(&redis.Options{})
-	defer rdb.Close()
-
-	cacher := msgCache{rdb: rdb}
-
-	err := cacher.PipeDeleteMessages(context.Background(), cid, seqs)
-	assert.Nil(t, err)
-
-	// validate
-	for _, msg := range inputMsgs {
-		key := cacher.getMessageCacheKey(cid, msg.Seq)
-		val := rdb.Exists(context.Background(), key).Val()
-		assert.EqualValues(t, 0, val)
-	}
-}
-
-func testParallelDeleteMessagesMix(t *testing.T, cid string, seqs []int64, inputMsgs []*sdkws.MsgData, lessValNonExists int) {
-	rdb := redis.NewClient(&redis.Options{})
-	defer rdb.Close()
-
-	cacher := msgCache{rdb: rdb}
-
-	err := cacher.PipeDeleteMessages(context.Background(), cid, seqs)
-	assert.Nil(t, err)
-
-	// validate
-	for idx, msg := range inputMsgs {
-		key := cacher.getMessageCacheKey(cid, msg.Seq)
-		val, err := rdb.Exists(context.Background(), key).Result()
-		assert.Nil(t, err)
-		if idx < lessValNonExists {
-			assert.EqualValues(t, 0, val) // not exists
-			continue
-		}
-
-		assert.EqualValues(t, 1, val) // exists
-	}
-}

pkg/common/db/controller/msg.go

@@ -16,27 +16,22 @@ package controller
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/prom_metrics"
 	"time"
 
-	"github.com/OpenIMSDK/protocol/constant"
-
-	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
-
 	"github.com/redis/go-redis/v9"
 
 	"github.com/OpenIMSDK/tools/errs"
 	"github.com/OpenIMSDK/tools/log"
 
-	"go.mongodb.org/mongo-driver/mongo"
-
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/convert"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
 	unrelationtb "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/unrelation"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/unrelation"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/kafka"
+	"go.mongodb.org/mongo-driver/mongo"
 
 	pbmsg "github.com/OpenIMSDK/protocol/msg"
 	"github.com/OpenIMSDK/protocol/sdkws"
@@ -381,20 +376,20 @@ func (db *commonMsgDatabase) BatchInsertChat2Cache(ctx context.Context, conversa
 	}
 	failedNum, err := db.cache.SetMessageToCache(ctx, conversationID, msgs)
 	if err != nil {
-		prommetrics.MsgInsertRedisFailedCounter.Add(float64(failedNum))
+		prom_metrics.MsgInsertRedisFailedCounter.Add(float64(failedNum))
 		log.ZError(ctx, "setMessageToCache error", err, "len", len(msgs), "conversationID", conversationID)
 	} else {
-		prommetrics.MsgInsertRedisSuccessCounter.Inc()
+		prom_metrics.MsgInsertRedisSuccessCounter.Inc()
 	}
 	err = db.cache.SetMaxSeq(ctx, conversationID, currentMaxSeq)
 	if err != nil {
 		log.ZError(ctx, "db.cache.SetMaxSeq error", err, "conversationID", conversationID)
-		prommetrics.SeqSetFailedCounter.Inc()
+		prom_metrics.SeqSetFailedCounter.Inc()
 	}
 	err2 := db.cache.SetHasReadSeqs(ctx, conversationID, userSeqMap)
 	if err != nil {
 		log.ZError(ctx, "SetHasReadSeqs error", err2, "userSeqMap", userSeqMap, "conversationID", conversationID)
-		prommetrics.SeqSetFailedCounter.Inc()
+		prom_metrics.SeqSetFailedCounter.Inc()
 	}
 	return lastMaxSeq, isNew, utils.Wrap(err, "")
 }
@@ -402,7 +397,7 @@ func (db *commonMsgDatabase) BatchInsertChat2Cache(ctx context.Context, conversa
 func (db *commonMsgDatabase) getMsgBySeqs(ctx context.Context, userID, conversationID string, seqs []int64) (totalMsgs []*sdkws.MsgData, err error) {
 	for docID, seqs := range db.msg.GetDocIDSeqsMap(conversationID, seqs) {
 		// log.ZDebug(ctx, "getMsgBySeqs", "docID", docID, "seqs", seqs)
-		msgs, err := db.findMsgInfoBySeq(ctx, userID, docID, conversationID, seqs)
+		msgs, err := db.findMsgInfoBySeq(ctx, userID, docID, seqs)
 		if err != nil {
 			return nil, err
 		}
@@ -413,70 +408,12 @@ func (db *commonMsgDatabase) getMsgBySeqs(ctx context.Context, userID, conversat
 	return totalMsgs, nil
 }
 
-func (db *commonMsgDatabase) handlerDBMsg(ctx context.Context, cache map[int64][]*unrelationtb.MsgInfoModel, userID, conversationID string, msg *unrelationtb.MsgInfoModel) {
-	if msg.IsRead {
-		msg.Msg.IsRead = true
-	}
-	if msg.Msg.ContentType != constant.Quote {
-		return
-	}
-	if msg.Msg.Content == "" {
-		return
-	}
-	var quoteMsg struct {
-		Text              string          `json:"text,omitempty"`
-		QuoteMessage      *sdkws.MsgData  `json:"quoteMessage,omitempty"`
-		MessageEntityList json.RawMessage `json:"messageEntityList,omitempty"`
-	}
-	if err := json.Unmarshal([]byte(msg.Msg.Content), &quoteMsg); err != nil {
-		log.ZError(ctx, "json.Unmarshal", err)
-		return
-	}
-	if quoteMsg.QuoteMessage == nil || quoteMsg.QuoteMessage.ContentType == constant.MsgRevokeNotification {
-		return
-	}
-	var msgs []*unrelationtb.MsgInfoModel
-	if v, ok := cache[quoteMsg.QuoteMessage.Seq]; ok {
-		msgs = v
-	} else {
-		if quoteMsg.QuoteMessage.Seq > 0 {
-			ms, err := db.msgDocDatabase.GetMsgBySeqIndexIn1Doc(ctx, userID, db.msg.GetDocID(conversationID, quoteMsg.QuoteMessage.Seq), []int64{quoteMsg.QuoteMessage.Seq})
-			if err != nil {
-				log.ZError(ctx, "GetMsgBySeqIndexIn1Doc", err, "conversationID", conversationID, "seq", quoteMsg.QuoteMessage.Seq)
-				return
-			}
-			msgs = ms
-			cache[quoteMsg.QuoteMessage.Seq] = ms
-		}
-	}
-	if len(msgs) != 0 && msgs[0].Msg.ContentType != constant.MsgRevokeNotification {
-		return
-	}
-	quoteMsg.QuoteMessage.ContentType = constant.MsgRevokeNotification
-	if len(msgs) > 0 {
-		quoteMsg.QuoteMessage.Content = []byte(msgs[0].Msg.Content)
-	} else {
-		quoteMsg.QuoteMessage.Content = []byte("{}")
-	}
-	data, err := json.Marshal(&quoteMsg)
-	if err != nil {
-		log.ZError(ctx, "json.Marshal", err)
-		return
-	}
-	msg.Msg.Content = string(data)
-	if _, err := db.msgDocDatabase.UpdateMsg(ctx, db.msg.GetDocID(conversationID, msg.Msg.Seq), db.msg.GetMsgIndex(msg.Msg.Seq), "msg", msg.Msg); err != nil {
-		log.ZError(ctx, "UpdateMsgContent", err)
-	}
-}
-
-func (db *commonMsgDatabase) findMsgInfoBySeq(ctx context.Context, userID, docID string, conversationID string, seqs []int64) (totalMsgs []*unrelationtb.MsgInfoModel, err error) {
+func (db *commonMsgDatabase) findMsgInfoBySeq(ctx context.Context, userID, docID string, seqs []int64) (totalMsgs []*unrelationtb.MsgInfoModel, err error) {
 	msgs, err := db.msgDocDatabase.GetMsgBySeqIndexIn1Doc(ctx, userID, docID, seqs)
-	if err != nil {
-		return nil, err
-	}
-	tempCache := make(map[int64][]*unrelationtb.MsgInfoModel)
 	for _, msg := range msgs {
-		db.handlerDBMsg(ctx, tempCache, userID, conversationID, msg)
+		if msg.IsRead {
+			msg.Msg.IsRead = true
+		}
 	}
 	return msgs, err
 }
@@ -485,7 +422,7 @@ func (db *commonMsgDatabase) getMsgBySeqsRange(ctx context.Context, userID strin
 	log.ZDebug(ctx, "getMsgBySeqsRange", "conversationID", conversationID, "allSeqs", allSeqs, "begin", begin, "end", end)
 	for docID, seqs := range db.msg.GetDocIDSeqsMap(conversationID, allSeqs) {
 		log.ZDebug(ctx, "getMsgBySeqsRange", "docID", docID, "seqs", seqs)
-		msgs, err := db.findMsgInfoBySeq(ctx, userID, docID, conversationID, seqs)
+		msgs, err := db.findMsgInfoBySeq(ctx, userID, docID, seqs)
 		if err != nil {
 			return nil, err
 		}

pkg/common/db/controller/msg_test.go

@@ -144,9 +144,9 @@ func Test_BatchInsertChat2DB(t *testing.T) {
 }
 
 func GetDB() *commonMsgDatabase {
-	config.Config.Mongo.Address = []string{"203.56.175.233:37017"}
+	config.Config.Mongo.Address = []string{"192.168.44.128:37017"}
 	// config.Config.Mongo.Timeout = 60
-	config.Config.Mongo.Database = "openIM_v3"
+	config.Config.Mongo.Database = "openIM"
 	// config.Config.Mongo.Source = "admin"
 	config.Config.Mongo.Username = "root"
 	config.Config.Mongo.Password = "openIM123"
@@ -232,17 +232,37 @@ func Test_FindBySeq(t *testing.T) {
 //	}
 //}
 
-func TestName(t *testing.T) {
-	db := GetDB()
-	var seqs []int64
-	for i := int64(1); i <= 4; i++ {
-		seqs = append(seqs, i)
-	}
-	msgs, err := db.getMsgBySeqsRange(context.Background(), "4931176757", "si_3866692501_4931176757", seqs, seqs[0], seqs[len(seqs)-1])
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	t.Log(msgs)
-
-}
+//func Test_Delete1(t *testing.T) {
+//	config.Config.Mongo.DBAddress = []string{"192.168.44.128:37017"}
+//	config.Config.Mongo.DBTimeout = 60
+//	config.Config.Mongo.DBDatabase = "openIM"
+//	config.Config.Mongo.DBSource = "admin"
+//	config.Config.Mongo.DBUserName = "root"
+//	config.Config.Mongo.DBPassword = "openIM123"
+//	config.Config.Mongo.DBMaxPoolSize = 100
+//	config.Config.Mongo.DBRetainChatRecords = 3650
+//	config.Config.Mongo.ChatRecordsClearTime = "0 2 * * 3"
+//
+//	mongo, err := unrelation.NewMongo()
+//	if err != nil {
+//		panic(err)
+//	}
+//	err = mongo.GetDatabase().Client().Ping(context.Background(), nil)
+//	if err != nil {
+//		panic(err)
+//	}
+//
+//	c := mongo.GetClient().Database("openIM").Collection("msg")
+//
+//	var o unrelationtb.MsgDocModel
+//
+//	err = c.FindOne(context.Background(), bson.M{"doc_id": "test:0"}).Decode(&o)
+//	if err != nil {
+//		panic(err)
+//	}
+//
+//	for i, model := range o.Msg {
+//		fmt.Println(i, model == nil)
+//	}
+//
+//}

pkg/common/db/s3/cont/controller.go

@@ -175,6 +175,7 @@ func (c *Controller) CompleteUpload(ctx context.Context, uploadID string, partHa
 		return nil, err
 	}
 	if md5Sum := md5.Sum([]byte(strings.Join(partHashs, partSeparator))); hex.EncodeToString(md5Sum[:]) != upload.Hash {
+		fmt.Println("CompleteUpload sum:", hex.EncodeToString(md5Sum[:]), "upload hash:", upload.Hash)
 		return nil, errors.New("md5 mismatching")
 	}
 	if info, err := c.StatObject(ctx, c.HashPath(upload.Hash)); err == nil {

pkg/common/discovery_register/k8s_discovery_register.go

@@ -1,4 +1,4 @@
-package discoveryregister
+package discovery_register
 
 import (
 	"context"

pkg/common/discoveryregister/discoveryregister_test.go

@@ -1,407 +0,0 @@
-package discoveryregister
-
-import (
-	"context"
-	"reflect"
-	"testing"
-
-	"github.com/OpenIMSDK/tools/discoveryregistry"
-	"google.golang.org/grpc"
-)
-
-func TestNewDiscoveryRegister(t *testing.T) {
-	type args struct {
-		envType string
-	}
-	tests := []struct {
-		name    string
-		args    args
-		want    discoveryregistry.SvcDiscoveryRegistry
-		wantErr bool
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			got, err := NewDiscoveryRegister(tt.args.envType)
-			if (err != nil) != tt.wantErr {
-				t.Errorf("NewDiscoveryRegister() error = %v, wantErr %v", err, tt.wantErr)
-				return
-			}
-			if !reflect.DeepEqual(got, tt.want) {
-				t.Errorf("NewDiscoveryRegister() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-func TestNewK8sDiscoveryRegister(t *testing.T) {
-	tests := []struct {
-		name    string
-		want    discoveryregistry.SvcDiscoveryRegistry
-		wantErr bool
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			got, err := NewK8sDiscoveryRegister()
-			if (err != nil) != tt.wantErr {
-				t.Errorf("NewK8sDiscoveryRegister() error = %v, wantErr %v", err, tt.wantErr)
-				return
-			}
-			if !reflect.DeepEqual(got, tt.want) {
-				t.Errorf("NewK8sDiscoveryRegister() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-func TestK8sDR_Register(t *testing.T) {
-	type fields struct {
-		options         []grpc.DialOption
-		rpcRegisterAddr string
-	}
-	type args struct {
-		serviceName string
-		host        string
-		port        int
-		opts        []grpc.DialOption
-	}
-	tests := []struct {
-		name    string
-		fields  fields
-		args    args
-		wantErr bool
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			cli := &K8sDR{
-				options:         tt.fields.options,
-				rpcRegisterAddr: tt.fields.rpcRegisterAddr,
-			}
-			if err := cli.Register(tt.args.serviceName, tt.args.host, tt.args.port, tt.args.opts...); (err != nil) != tt.wantErr {
-				t.Errorf("K8sDR.Register() error = %v, wantErr %v", err, tt.wantErr)
-			}
-		})
-	}
-}
-
-func TestK8sDR_UnRegister(t *testing.T) {
-	type fields struct {
-		options         []grpc.DialOption
-		rpcRegisterAddr string
-	}
-	tests := []struct {
-		name    string
-		fields  fields
-		wantErr bool
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			cli := &K8sDR{
-				options:         tt.fields.options,
-				rpcRegisterAddr: tt.fields.rpcRegisterAddr,
-			}
-			if err := cli.UnRegister(); (err != nil) != tt.wantErr {
-				t.Errorf("K8sDR.UnRegister() error = %v, wantErr %v", err, tt.wantErr)
-			}
-		})
-	}
-}
-
-func TestK8sDR_CreateRpcRootNodes(t *testing.T) {
-	type fields struct {
-		options         []grpc.DialOption
-		rpcRegisterAddr string
-	}
-	type args struct {
-		serviceNames []string
-	}
-	tests := []struct {
-		name    string
-		fields  fields
-		args    args
-		wantErr bool
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			cli := &K8sDR{
-				options:         tt.fields.options,
-				rpcRegisterAddr: tt.fields.rpcRegisterAddr,
-			}
-			if err := cli.CreateRpcRootNodes(tt.args.serviceNames); (err != nil) != tt.wantErr {
-				t.Errorf("K8sDR.CreateRpcRootNodes() error = %v, wantErr %v", err, tt.wantErr)
-			}
-		})
-	}
-}
-
-func TestK8sDR_RegisterConf2Registry(t *testing.T) {
-	type fields struct {
-		options         []grpc.DialOption
-		rpcRegisterAddr string
-	}
-	type args struct {
-		key  string
-		conf []byte
-	}
-	tests := []struct {
-		name    string
-		fields  fields
-		args    args
-		wantErr bool
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			cli := &K8sDR{
-				options:         tt.fields.options,
-				rpcRegisterAddr: tt.fields.rpcRegisterAddr,
-			}
-			if err := cli.RegisterConf2Registry(tt.args.key, tt.args.conf); (err != nil) != tt.wantErr {
-				t.Errorf("K8sDR.RegisterConf2Registry() error = %v, wantErr %v", err, tt.wantErr)
-			}
-		})
-	}
-}
-
-func TestK8sDR_GetConfFromRegistry(t *testing.T) {
-	type fields struct {
-		options         []grpc.DialOption
-		rpcRegisterAddr string
-	}
-	type args struct {
-		key string
-	}
-	tests := []struct {
-		name    string
-		fields  fields
-		args    args
-		want    []byte
-		wantErr bool
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			cli := &K8sDR{
-				options:         tt.fields.options,
-				rpcRegisterAddr: tt.fields.rpcRegisterAddr,
-			}
-			got, err := cli.GetConfFromRegistry(tt.args.key)
-			if (err != nil) != tt.wantErr {
-				t.Errorf("K8sDR.GetConfFromRegistry() error = %v, wantErr %v", err, tt.wantErr)
-				return
-			}
-			if !reflect.DeepEqual(got, tt.want) {
-				t.Errorf("K8sDR.GetConfFromRegistry() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-func TestK8sDR_GetConns(t *testing.T) {
-	type fields struct {
-		options         []grpc.DialOption
-		rpcRegisterAddr string
-	}
-	type args struct {
-		ctx         context.Context
-		serviceName string
-		opts        []grpc.DialOption
-	}
-	tests := []struct {
-		name    string
-		fields  fields
-		args    args
-		want    []*grpc.ClientConn
-		wantErr bool
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			cli := &K8sDR{
-				options:         tt.fields.options,
-				rpcRegisterAddr: tt.fields.rpcRegisterAddr,
-			}
-			got, err := cli.GetConns(tt.args.ctx, tt.args.serviceName, tt.args.opts...)
-			if (err != nil) != tt.wantErr {
-				t.Errorf("K8sDR.GetConns() error = %v, wantErr %v", err, tt.wantErr)
-				return
-			}
-			if !reflect.DeepEqual(got, tt.want) {
-				t.Errorf("K8sDR.GetConns() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-func TestK8sDR_GetConn(t *testing.T) {
-	type fields struct {
-		options         []grpc.DialOption
-		rpcRegisterAddr string
-	}
-	type args struct {
-		ctx         context.Context
-		serviceName string
-		opts        []grpc.DialOption
-	}
-	tests := []struct {
-		name    string
-		fields  fields
-		args    args
-		want    *grpc.ClientConn
-		wantErr bool
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			cli := &K8sDR{
-				options:         tt.fields.options,
-				rpcRegisterAddr: tt.fields.rpcRegisterAddr,
-			}
-			got, err := cli.GetConn(tt.args.ctx, tt.args.serviceName, tt.args.opts...)
-			if (err != nil) != tt.wantErr {
-				t.Errorf("K8sDR.GetConn() error = %v, wantErr %v", err, tt.wantErr)
-				return
-			}
-			if !reflect.DeepEqual(got, tt.want) {
-				t.Errorf("K8sDR.GetConn() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-func TestK8sDR_GetSelfConnTarget(t *testing.T) {
-	type fields struct {
-		options         []grpc.DialOption
-		rpcRegisterAddr string
-	}
-	tests := []struct {
-		name   string
-		fields fields
-		want   string
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			cli := &K8sDR{
-				options:         tt.fields.options,
-				rpcRegisterAddr: tt.fields.rpcRegisterAddr,
-			}
-			if got := cli.GetSelfConnTarget(); got != tt.want {
-				t.Errorf("K8sDR.GetSelfConnTarget() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-func TestK8sDR_AddOption(t *testing.T) {
-	type fields struct {
-		options         []grpc.DialOption
-		rpcRegisterAddr string
-	}
-	type args struct {
-		opts []grpc.DialOption
-	}
-	tests := []struct {
-		name   string
-		fields fields
-		args   args
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			cli := &K8sDR{
-				options:         tt.fields.options,
-				rpcRegisterAddr: tt.fields.rpcRegisterAddr,
-			}
-			cli.AddOption(tt.args.opts...)
-		})
-	}
-}
-
-func TestK8sDR_CloseConn(t *testing.T) {
-	type fields struct {
-		options         []grpc.DialOption
-		rpcRegisterAddr string
-	}
-	type args struct {
-		conn *grpc.ClientConn
-	}
-	tests := []struct {
-		name   string
-		fields fields
-		args   args
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			cli := &K8sDR{
-				options:         tt.fields.options,
-				rpcRegisterAddr: tt.fields.rpcRegisterAddr,
-			}
-			cli.CloseConn(tt.args.conn)
-		})
-	}
-}
-
-func TestK8sDR_GetClientLocalConns(t *testing.T) {
-	type fields struct {
-		options         []grpc.DialOption
-		rpcRegisterAddr string
-	}
-	tests := []struct {
-		name   string
-		fields fields
-		want   map[string][]*grpc.ClientConn
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			cli := &K8sDR{
-				options:         tt.fields.options,
-				rpcRegisterAddr: tt.fields.rpcRegisterAddr,
-			}
-			if got := cli.GetClientLocalConns(); !reflect.DeepEqual(got, tt.want) {
-				t.Errorf("K8sDR.GetClientLocalConns() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-func TestK8sDR_Close(t *testing.T) {
-	type fields struct {
-		options         []grpc.DialOption
-		rpcRegisterAddr string
-	}
-	tests := []struct {
-		name   string
-		fields fields
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			cli := &K8sDR{
-				options:         tt.fields.options,
-				rpcRegisterAddr: tt.fields.rpcRegisterAddr,
-			}
-			cli.Close()
-		})
-	}
-}

pkg/common/ginPrometheus/ginPrometheus.go

@@ -1,9 +1,9 @@
-package ginprometheus
+package ginPrometheus
 
 import (
 	"bytes"
 	"fmt"
-	"io"
+	"io/ioutil"
 	"net/http"
 	"os"
 	"strconv"
@@ -17,42 +17,40 @@ import (
 var defaultMetricPath = "/metrics"
 
 // counter, counter_vec, gauge, gauge_vec,
-// histogram, histogram_vec, summary, summary_vec.
-var (
-	reqCounter = &Metric{
-		ID:          "reqCnt",
-		Name:        "requests_total",
-		Description: "How many HTTP requests processed, partitioned by status code and HTTP method.",
-		Type:        "counter_vec",
-		Args:        []string{"code", "method", "handler", "host", "url"}}
+// histogram, histogram_vec, summary, summary_vec
+var reqCnt = &Metric{
+	ID:          "reqCnt",
+	Name:        "requests_total",
+	Description: "How many HTTP requests processed, partitioned by status code and HTTP method.",
+	Type:        "counter_vec",
+	Args:        []string{"code", "method", "handler", "host", "url"}}
 
-	reqDuration = &Metric{
-		ID:          "reqDur",
-		Name:        "request_duration_seconds",
-		Description: "The HTTP request latencies in seconds.",
-		Type:        "histogram_vec",
-		Args:        []string{"code", "method", "url"},
-	}
+var reqDur = &Metric{
+	ID:          "reqDur",
+	Name:        "request_duration_seconds",
+	Description: "The HTTP request latencies in seconds.",
+	Type:        "histogram_vec",
+	Args:        []string{"code", "method", "url"},
+}
 
-	resSize = &Metric{
-		ID:          "resSz",
-		Name:        "response_size_bytes",
-		Description: "The HTTP response sizes in bytes.",
-		Type:        "summary"}
+var resSz = &Metric{
+	ID:          "resSz",
+	Name:        "response_size_bytes",
+	Description: "The HTTP response sizes in bytes.",
+	Type:        "summary"}
 
-	reqSize = &Metric{
-		ID:          "reqSz",
-		Name:        "request_size_bytes",
-		Description: "The HTTP request sizes in bytes.",
-		Type:        "summary"}
+var reqSz = &Metric{
+	ID:          "reqSz",
+	Name:        "request_size_bytes",
+	Description: "The HTTP request sizes in bytes.",
+	Type:        "summary"}
 
-	standardMetrics = []*Metric{
-		reqCounter,
-		reqDuration,
-		resSize,
-		reqSize,
-	}
-)
+var standardMetrics = []*Metric{
+	reqCnt,
+	reqDur,
+	resSz,
+	reqSz,
+}
 
 /*
 RequestCounterURLLabelMappingFn is a function which can be supplied to the middleware to control
@@ -76,7 +74,7 @@ which would map "/customer/alice" and "/customer/bob" to their template "/custom
 type RequestCounterURLLabelMappingFn func(c *gin.Context) string
 
 // Metric is a definition for the name, description, type, ID, and
-// prometheus.Collector type (i.e. CounterVec, Summary, etc) of each metric.
+// prometheus.Collector type (i.e. CounterVec, Summary, etc) of each metric
 type Metric struct {
 	MetricCollector prometheus.Collector
 	ID              string
@@ -86,7 +84,7 @@ type Metric struct {
 	Args            []string
 }
 
-// Prometheus contains the metrics gathered by the instance and its path.
+// Prometheus contains the metrics gathered by the instance and its path
 type Prometheus struct {
 	reqCnt        *prometheus.CounterVec
 	reqDur        *prometheus.HistogramVec
@@ -104,7 +102,7 @@ type Prometheus struct {
 	URLLabelFromContext string
 }
 
-// PrometheusPushGateway contains the configuration for pushing to a Prometheus pushgateway (optional).
+// PrometheusPushGateway contains the configuration for pushing to a Prometheus pushgateway (optional)
 type PrometheusPushGateway struct {
 
 	// Push interval in seconds
@@ -114,7 +112,7 @@ type PrometheusPushGateway struct {
 	// where JOBNAME can be any string of your choice
 	PushGatewayURL string
 
-	// Local metrics URL where metrics are fetched from, this could be omitted in the future
+	// Local metrics URL where metrics are fetched from, this could be ommited in the future
 	// if implemented using prometheus common/expfmt instead
 	MetricsURL string
 
@@ -122,11 +120,9 @@ type PrometheusPushGateway struct {
 	Job string
 }
 
-// NewPrometheus generates a new set of metrics with a certain subsystem name.
+// NewPrometheus generates a new set of metrics with a certain subsystem name
 func NewPrometheus(subsystem string, customMetricsList ...[]*Metric) *Prometheus {
-	if subsystem == "" {
-		subsystem = "app"
-	}
+	subsystem = "app"
 
 	var metricsList []*Metric
 
@@ -135,13 +131,16 @@ func NewPrometheus(subsystem string, customMetricsList ...[]*Metric) *Prometheus
 	} else if len(customMetricsList) == 1 {
 		metricsList = customMetricsList[0]
 	}
-	metricsList = append(metricsList, standardMetrics...)
+
+	for _, metric := range standardMetrics {
+		metricsList = append(metricsList, metric)
+	}
 
 	p := &Prometheus{
 		MetricsList: metricsList,
 		MetricsPath: defaultMetricPath,
 		ReqCntURLLabelMappingFn: func(c *gin.Context) string {
-			return c.FullPath() // e.g. /user/:id , /user/:id/info
+			return c.Request.URL.Path
 		},
 	}
 
@@ -151,7 +150,7 @@ func NewPrometheus(subsystem string, customMetricsList ...[]*Metric) *Prometheus
 }
 
 // SetPushGateway sends metrics to a remote pushgateway exposed on pushGatewayURL
-// every pushIntervalSeconds. Metrics are fetched from metricsURL.
+// every pushIntervalSeconds. Metrics are fetched from metricsURL
 func (p *Prometheus) SetPushGateway(pushGatewayURL, metricsURL string, pushIntervalSeconds time.Duration) {
 	p.Ppg.PushGatewayURL = pushGatewayURL
 	p.Ppg.MetricsURL = metricsURL
@@ -159,13 +158,13 @@ func (p *Prometheus) SetPushGateway(pushGatewayURL, metricsURL string, pushInter
 	p.startPushTicker()
 }
 
-// SetPushGatewayJob job name, defaults to "gin".
+// SetPushGatewayJob job name, defaults to "gin"
 func (p *Prometheus) SetPushGatewayJob(j string) {
 	p.Ppg.Job = j
 }
 
 // SetListenAddress for exposing metrics on address. If not set, it will be exposed at the
-// same address of the gin engine that is being used.
+// same address of the gin engine that is being used
 func (p *Prometheus) SetListenAddress(address string) {
 	p.listenAddress = address
 	if p.listenAddress != "" {
@@ -182,7 +181,7 @@ func (p *Prometheus) SetListenAddressWithRouter(listenAddress string, r *gin.Eng
 	}
 }
 
-// SetMetricsPath set metrics paths.
+// SetMetricsPath set metrics paths
 func (p *Prometheus) SetMetricsPath(e *gin.Engine) {
 
 	if p.listenAddress != "" {
@@ -193,7 +192,7 @@ func (p *Prometheus) SetMetricsPath(e *gin.Engine) {
 	}
 }
 
-// SetMetricsPathWithAuth set metrics paths with authentication.
+// SetMetricsPathWithAuth set metrics paths with authentication
 func (p *Prometheus) SetMetricsPathWithAuth(e *gin.Engine, accounts gin.Accounts) {
 
 	if p.listenAddress != "" {
@@ -206,43 +205,34 @@ func (p *Prometheus) SetMetricsPathWithAuth(e *gin.Engine, accounts gin.Accounts
 }
 
 func (p *Prometheus) runServer() {
-	go p.router.Run(p.listenAddress)
+	if p.listenAddress != "" {
+		go p.router.Run(p.listenAddress)
+	}
 }
 
 func (p *Prometheus) getMetrics() []byte {
-	response, err := http.Get(p.Ppg.MetricsURL)
-	if err != nil {
-		return nil
-	}
+	response, _ := http.Get(p.Ppg.MetricsURL)
 
 	defer response.Body.Close()
+	body, _ := ioutil.ReadAll(response.Body)
 
-	body, _ := io.ReadAll(response.Body)
 	return body
 }
 
-var hostname, _ = os.Hostname()
-
 func (p *Prometheus) getPushGatewayURL() string {
+	h, _ := os.Hostname()
 	if p.Ppg.Job == "" {
 		p.Ppg.Job = "gin"
 	}
-	return p.Ppg.PushGatewayURL + "/metrics/job/" + p.Ppg.Job + "/instance/" + hostname
+	return p.Ppg.PushGatewayURL + "/metrics/job/" + p.Ppg.Job + "/instance/" + h
 }
 
 func (p *Prometheus) sendMetricsToPushGateway(metrics []byte) {
 	req, err := http.NewRequest("POST", p.getPushGatewayURL(), bytes.NewBuffer(metrics))
-	if err != nil {
-		return
-	}
-
 	client := &http.Client{}
-	resp, err := client.Do(req)
-	if err != nil {
+	if _, err = client.Do(req); err != nil {
 		fmt.Println("Error sending to push gateway error:", err.Error())
 	}
-
-	resp.Body.Close()
 }
 
 func (p *Prometheus) startPushTicker() {
@@ -254,7 +244,7 @@ func (p *Prometheus) startPushTicker() {
 	}()
 }
 
-// NewMetric associates prometheus.Collector based on Metric.Type.
+// NewMetric associates prometheus.Collector based on Metric.Type
 func NewMetric(m *Metric, subsystem string) prometheus.Collector {
 	var metric prometheus.Collector
 	switch m.Type {
@@ -331,20 +321,20 @@ func NewMetric(m *Metric, subsystem string) prometheus.Collector {
 }
 
 func (p *Prometheus) registerMetrics(subsystem string) {
+
 	for _, metricDef := range p.MetricsList {
 		metric := NewMetric(metricDef, subsystem)
 		if err := prometheus.Register(metric); err != nil {
 			fmt.Println("could not be registered in Prometheus,metricDef.Name:", metricDef.Name, "   error:", err.Error())
 		}
-
 		switch metricDef {
-		case reqCounter:
+		case reqCnt:
 			p.reqCnt = metric.(*prometheus.CounterVec)
-		case reqDuration:
+		case reqDur:
 			p.reqDur = metric.(*prometheus.HistogramVec)
-		case resSize:
+		case resSz:
 			p.resSz = metric.(prometheus.Summary)
-		case reqSize:
+		case reqSz:
 			p.reqSz = metric.(prometheus.Summary)
 		}
 		metricDef.MetricCollector = metric
@@ -363,7 +353,7 @@ func (p *Prometheus) UseWithAuth(e *gin.Engine, accounts gin.Accounts) {
 	p.SetMetricsPathWithAuth(e, accounts)
 }
 
-// HandlerFunc defines handler function for middleware.
+// HandlerFunc defines handler function for middleware
 func (p *Prometheus) HandlerFunc() gin.HandlerFunc {
 	return func(c *gin.Context) {
 		if c.Request.URL.Path == p.MetricsPath {
@@ -403,7 +393,7 @@ func prometheusHandler() gin.HandlerFunc {
 }
 
 func computeApproximateRequestSize(r *http.Request) int {
-	var s int
+	s := 0
 	if r.URL != nil {
 		s = len(r.URL.Path)
 	}

pkg/common/http/http_client_test.go

@@ -1,154 +0,0 @@
-// Copyright © 2023 OpenIM. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package http
-
-import (
-	"context"
-	"reflect"
-	"testing"
-
-	"github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
-)
-
-func TestGet(t *testing.T) {
-	type args struct {
-		url string
-	}
-	tests := []struct {
-		name         string
-		args         args
-		wantResponse []byte
-		wantErr      bool
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			gotResponse, err := Get(tt.args.url)
-			if (err != nil) != tt.wantErr {
-				t.Errorf("Get() error = %v, wantErr %v", err, tt.wantErr)
-				return
-			}
-			if !reflect.DeepEqual(gotResponse, tt.wantResponse) {
-				t.Errorf("Get() = %v, want %v", gotResponse, tt.wantResponse)
-			}
-		})
-	}
-}
-
-func TestPost(t *testing.T) {
-	type args struct {
-		ctx     context.Context
-		url     string
-		header  map[string]string
-		data    interface{}
-		timeout int
-	}
-	tests := []struct {
-		name        string
-		args        args
-		wantContent []byte
-		wantErr     bool
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			gotContent, err := Post(tt.args.ctx, tt.args.url, tt.args.header, tt.args.data, tt.args.timeout)
-			if (err != nil) != tt.wantErr {
-				t.Errorf("Post() error = %v, wantErr %v", err, tt.wantErr)
-				return
-			}
-			if !reflect.DeepEqual(gotContent, tt.wantContent) {
-				t.Errorf("Post() = %v, want %v", gotContent, tt.wantContent)
-			}
-		})
-	}
-}
-
-func TestPostReturn(t *testing.T) {
-	type args struct {
-		ctx           context.Context
-		url           string
-		header        map[string]string
-		input         interface{}
-		output        interface{}
-		timeOutSecond int
-	}
-	tests := []struct {
-		name    string
-		args    args
-		wantErr bool
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if err := PostReturn(tt.args.ctx, tt.args.url, tt.args.header, tt.args.input, tt.args.output, tt.args.timeOutSecond); (err != nil) != tt.wantErr {
-				t.Errorf("PostReturn() error = %v, wantErr %v", err, tt.wantErr)
-			}
-		})
-	}
-}
-
-func Test_callBackPostReturn(t *testing.T) {
-	type args struct {
-		ctx            context.Context
-		url            string
-		command        string
-		input          interface{}
-		output         callbackstruct.CallbackResp
-		callbackConfig config.CallBackConfig
-	}
-	tests := []struct {
-		name    string
-		args    args
-		wantErr bool
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if err := callBackPostReturn(tt.args.ctx, tt.args.url, tt.args.command, tt.args.input, tt.args.output, tt.args.callbackConfig); (err != nil) != tt.wantErr {
-				t.Errorf("callBackPostReturn() error = %v, wantErr %v", err, tt.wantErr)
-			}
-		})
-	}
-}
-
-func TestCallBackPostReturn(t *testing.T) {
-	type args struct {
-		ctx            context.Context
-		url            string
-		req            callbackstruct.CallbackReq
-		resp           callbackstruct.CallbackResp
-		callbackConfig config.CallBackConfig
-	}
-	tests := []struct {
-		name    string
-		args    args
-		wantErr bool
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if err := CallBackPostReturn(tt.args.ctx, tt.args.url, tt.args.req, tt.args.resp, tt.args.callbackConfig); (err != nil) != tt.wantErr {
-				t.Errorf("CallBackPostReturn() error = %v, wantErr %v", err, tt.wantErr)
-			}
-		})
-	}
-}

pkg/common/kafka/producer.go

@@ -15,10 +15,8 @@
 package kafka
 
 import (
-	"bytes"
 	"context"
 	"errors"
-	"strings"
 	"time"
 
 	"github.com/OpenIMSDK/protocol/constant"
@@ -51,23 +49,8 @@ func NewKafkaProducer(addr []string, topic string) *Producer {
 	p.config = sarama.NewConfig()             // Instantiate a sarama Config
 	p.config.Producer.Return.Successes = true // Whether to enable the successes channel to be notified after the message is sent successfully
 	p.config.Producer.Return.Errors = true
+	p.config.Producer.RequiredAcks = sarama.WaitForAll        // Set producer Message Reply level 0 1 all
 	p.config.Producer.Partitioner = sarama.NewHashPartitioner // Set the hash-key automatic hash partition. When sending a message, you must specify the key value of the message. If there is no key, the partition will be selected randomly
-
-	var producerAck = sarama.WaitForAll // default: WaitForAll
-	switch strings.ToLower(config.Config.Kafka.ProducerAck) {
-	case "no_response":
-		producerAck = sarama.NoResponse
-	case "wait_for_local":
-		producerAck = sarama.WaitForLocal
-	case "wait_for_all":
-		producerAck = sarama.WaitForAll
-	}
-	p.config.Producer.RequiredAcks = producerAck
-
-	var compress = sarama.CompressionNone // default: no compress
-	_ = compress.UnmarshalText(bytes.ToLower([]byte(config.Config.Kafka.CompressType)))
-	p.config.Producer.Compression = compress
-
 	if config.Config.Kafka.Username != "" && config.Config.Kafka.Password != "" {
 		p.config.Net.SASL.Enable = true
 		p.config.Net.SASL.User = config.Config.Kafka.Username

pkg/common/locker/doc.go

@@ -0,0 +1,15 @@
+// Copyright © 2023 OpenIM. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package locker // import "github.com/openimsdk/open-im-server/v3/pkg/common/locker"

pkg/common/locker/message_locker.go

@@ -0,0 +1,72 @@
+// Copyright © 2023 OpenIM. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package locker
+
+import (
+	"context"
+	"time"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
+)
+
+const GlOBALLOCK = "GLOBAL_LOCK"
+
+type MessageLocker interface {
+	LockMessageTypeKey(ctx context.Context, clientMsgID, typeKey string) (err error)
+	UnLockMessageTypeKey(ctx context.Context, clientMsgID string, typeKey string) error
+	LockGlobalMessage(ctx context.Context, clientMsgID string) (err error)
+	UnLockGlobalMessage(ctx context.Context, clientMsgID string) (err error)
+}
+type LockerMessage struct {
+	cache cache.MsgModel
+}
+
+func NewLockerMessage(cache cache.MsgModel) *LockerMessage {
+	return &LockerMessage{cache: cache}
+}
+
+func (l *LockerMessage) LockMessageTypeKey(ctx context.Context, clientMsgID, typeKey string) (err error) {
+	for i := 0; i < 3; i++ {
+		err = l.cache.LockMessageTypeKey(ctx, clientMsgID, typeKey)
+		if err != nil {
+			time.Sleep(time.Millisecond * 100)
+			continue
+		} else {
+			break
+		}
+	}
+	return err
+}
+
+func (l *LockerMessage) LockGlobalMessage(ctx context.Context, clientMsgID string) (err error) {
+	for i := 0; i < 3; i++ {
+		err = l.cache.LockMessageTypeKey(ctx, clientMsgID, GlOBALLOCK)
+		if err != nil {
+			time.Sleep(time.Millisecond * 100)
+			continue
+		} else {
+			break
+		}
+	}
+	return err
+}
+
+func (l *LockerMessage) UnLockMessageTypeKey(ctx context.Context, clientMsgID string, typeKey string) error {
+	return l.cache.UnLockMessageTypeKey(ctx, clientMsgID, typeKey)
+}
+
+func (l *LockerMessage) UnLockGlobalMessage(ctx context.Context, clientMsgID string) error {
+	return l.cache.UnLockMessageTypeKey(ctx, clientMsgID, GlOBALLOCK)
+}

pkg/common/prom_metrics/func.go

@@ -1,12 +1,11 @@
-package prommetrics
+package prom_metrics
 
 import (
 	grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
+	config2 "github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/ginPrometheus"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/collectors"
-
-	config2 "github.com/openimsdk/open-im-server/v3/pkg/common/config"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/ginprometheus"
 )
 
 func NewGrpcPromObj(cusMetrics []prometheus.Collector) (*prometheus.Registry, *grpc_prometheus.ServerMetrics, error) {
@@ -36,11 +35,11 @@ func GetGrpcCusMetrics(registerName string) []prometheus.Collector {
 	}
 }
 
-func GetGinCusMetrics(name string) []*ginprometheus.Metric {
+func GetGinCusMetrics(name string) []*ginPrometheus.Metric {
 	switch name {
 	case "Api":
-		return []*ginprometheus.Metric{ApiCustomCnt}
+		return []*ginPrometheus.Metric{ApiCustomCnt}
 	default:
-		return []*ginprometheus.Metric{ApiCustomCnt}
+		return []*ginPrometheus.Metric{ApiCustomCnt}
 	}
 }

pkg/common/prom_metrics/gin-api.go

@@ -1,6 +1,6 @@
-package prommetrics
+package prom_metrics
 
-import ginProm "github.com/openimsdk/open-im-server/v3/pkg/common/ginprometheus"
+import ginProm "github.com/openimsdk/open-im-server/v3/pkg/common/ginPrometheus"
 
 /*
 labels := prometheus.Labels{"label_one": "any", "label_two": "value"}

pkg/common/prom_metrics/grpc-auth.go

@@ -1,4 +1,4 @@
-package prommetrics
+package prom_metrics
 
 import (
 	"github.com/prometheus/client_golang/prometheus"

pkg/common/prom_metrics/grpc-msg.go

@@ -1,4 +1,4 @@
-package prommetrics
+package prom_metrics
 
 import (
 	"github.com/prometheus/client_golang/prometheus"

pkg/common/prom_metrics/grpc-msggateway.go

@@ -1,4 +1,4 @@
-package prommetrics
+package prom_metrics
 
 import (
 	"github.com/prometheus/client_golang/prometheus"

pkg/common/prom_metrics/grpc_push.go

@@ -1,4 +1,4 @@
-package prommetrics
+package prom_metrics
 
 import (
 	"github.com/prometheus/client_golang/prometheus"

pkg/common/prom_metrics/transfer.go

@@ -1,4 +1,4 @@
-package prommetrics
+package prom_metrics
 
 import (
 	"github.com/prometheus/client_golang/prometheus"

pkg/common/prommetrics/prommetrics_test.go

@@ -1,61 +0,0 @@
-package prommetrics
-
-import (
-	"testing"
-
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/stretchr/testify/assert"
-
-	config2 "github.com/openimsdk/open-im-server/v3/pkg/common/config"
-)
-
-func TestNewGrpcPromObj(t *testing.T) {
-	// Create a custom metric to pass into the NewGrpcPromObj function.
-	customMetric := prometheus.NewCounter(prometheus.CounterOpts{
-		Name: "test_metric",
-		Help: "This is a test metric.",
-	})
-	cusMetrics := []prometheus.Collector{customMetric}
-
-	// Call NewGrpcPromObj with the custom metrics.
-	reg, grpcMetrics, err := NewGrpcPromObj(cusMetrics)
-
-	// Assert no error was returned.
-	assert.NoError(t, err)
-
-	// Assert the registry was correctly initialized.
-	assert.NotNil(t, reg)
-
-	// Assert the grpcMetrics was correctly initialized.
-	assert.NotNil(t, grpcMetrics)
-
-	// Assert that the custom metric is registered.
-	mfs, err := reg.Gather()
-	assert.NoError(t, err)
-	assert.NotEmpty(t, mfs) // Ensure some metrics are present.
-	found := false
-	for _, mf := range mfs {
-		if *mf.Name == "test_metric" {
-			found = true
-			break
-		}
-	}
-	assert.True(t, found, "Custom metric not found in registry")
-}
-
-func TestGetGrpcCusMetrics(t *testing.T) {
-	// Test various cases based on the switch statement in the GetGrpcCusMetrics function.
-	testCases := []struct {
-		name     string
-		expected int // The expected number of metrics for each case.
-	}{
-		{config2.Config.RpcRegisterName.OpenImMessageGatewayName, 1},
-	}
-
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			metrics := GetGrpcCusMetrics(tc.name)
-			assert.Len(t, metrics, tc.expected)
-		})
-	}
-}

pkg/common/startrpc/start.go

@@ -16,23 +16,21 @@ package startrpc
 
 import (
 	"fmt"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/prom_metrics"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"log"
 	"net"
 	"net/http"
 	"strconv"
 
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promhttp"
-
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/discovery_register"
 
 	grpcprometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 
-	kdisc "github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister"
-
 	"github.com/OpenIMSDK/tools/discoveryregistry"
 	"github.com/OpenIMSDK/tools/mw"
 	"github.com/OpenIMSDK/tools/network"
@@ -57,7 +55,7 @@ func Start(
 		return err
 	}
 	defer listener.Close()
-	client, err := kdisc.NewDiscoveryRegister(config.Config.Envs.Discovery)
+	client, err := discovery_register.NewDiscoveryRegister(config.Config.Envs.Discovery)
 	if err != nil {
 		return utils.Wrap1(err)
 	}
@@ -72,8 +70,8 @@ func Start(
 	// ctx 中间件
 	if config.Config.Prometheus.Enable {
 		//////////////////////////
-		cusMetrics := prommetrics.GetGrpcCusMetrics(rpcRegisterName)
-		reg, metric, err = prommetrics.NewGrpcPromObj(cusMetrics)
+		cusMetrics := prom_metrics.GetGrpcCusMetrics(rpcRegisterName)
+		reg, metric, err = prom_metrics.NewGrpcPromObj(cusMetrics)
 		options = append(options, mw.GrpcServer(), grpc.StreamInterceptor(metric.StreamServerInterceptor()),
 			grpc.UnaryInterceptor(metric.UnaryServerInterceptor()))
 	} else {

pkg/common/startrpc/start_test.go

@@ -1,52 +0,0 @@
-package startrpc
-
-import (
-	"fmt"
-	"net"
-	"testing"
-	"time"
-
-	"github.com/OpenIMSDK/tools/discoveryregistry"
-	"google.golang.org/grpc"
-)
-
-// mockRpcFn is a mock gRPC function for testing.
-func mockRpcFn(client discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) error {
-	// Implement a mock gRPC service registration logic if needed
-	return nil
-}
-
-// TestStart tests the Start function for starting the RPC server.
-func TestStart(t *testing.T) {
-	// Use an available port for testing purposes.
-	testRpcPort := 12345
-	testPrometheusPort := 12346
-	testRpcRegisterName := "testService"
-
-	doneChan := make(chan error, 1)
-
-	go func() {
-		err := Start(testRpcPort, testRpcRegisterName, testPrometheusPort, mockRpcFn)
-		doneChan <- err
-	}()
-
-	// Give some time for the server to start.
-	time.Sleep(2 * time.Second)
-
-	// Test if the server is listening on the RPC port.
-	conn, err := net.Dial("tcp", fmt.Sprintf(":%d", testRpcPort))
-	if err != nil {
-		// t.Fatalf("Failed to dial the RPC server: %v", err)
-		// TODO: Fix this test
-		t.Skip("Failed to dial the RPC server")
-	}
-	conn.Close()
-
-	// More tests could be added here to check the registration logic, Prometheus metrics, etc.
-
-	// Cleanup
-	err = <-doneChan // This will block until Start returns an error or finishes
-	if err != nil {
-		t.Fatalf("Start returned an error: %v", err)
-	}
-}

pkg/common/tls/tls.go

@@ -24,7 +24,6 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 )
 
-// decryptPEM decrypts a PEM block using a password.
 func decryptPEM(data []byte, passphrase []byte) ([]byte, error) {
 	if len(passphrase) == 0 {
 		return data, nil

pkg/msgprocessor/conversation.go

@@ -118,9 +118,8 @@ func GetNotificationConversationIDByConversationID(conversationID string) string
 	l := strings.Split(conversationID, "_")
 	if len(l) > 1 {
 		l[0] = "n"
-		return strings.Join(l, "_")
+		return conversationID
 	}
-
 	return ""
 }
 

pkg/msgprocessor/conversation_test.go

@@ -1,334 +0,0 @@
-// Copyright © 2023 OpenIM. All rights reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package msgprocessor
-
-import (
-	"testing"
-
-	"github.com/OpenIMSDK/protocol/sdkws"
-	"google.golang.org/protobuf/proto"
-)
-
-func TestGetNotificationConversationIDByMsg(t *testing.T) {
-	type args struct {
-		msg *sdkws.MsgData
-	}
-	tests := []struct {
-		name string
-		args args
-		want string
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if got := GetNotificationConversationIDByMsg(tt.args.msg); got != tt.want {
-				t.Errorf("GetNotificationConversationIDByMsg() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-func TestGetChatConversationIDByMsg(t *testing.T) {
-	type args struct {
-		msg *sdkws.MsgData
-	}
-	tests := []struct {
-		name string
-		args args
-		want string
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if got := GetChatConversationIDByMsg(tt.args.msg); got != tt.want {
-				t.Errorf("GetChatConversationIDByMsg() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-func TestGenConversationUniqueKey(t *testing.T) {
-	type args struct {
-		msg *sdkws.MsgData
-	}
-	tests := []struct {
-		name string
-		args args
-		want string
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if got := GenConversationUniqueKey(tt.args.msg); got != tt.want {
-				t.Errorf("GenConversationUniqueKey() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-func TestGetConversationIDByMsg(t *testing.T) {
-	type args struct {
-		msg *sdkws.MsgData
-	}
-	tests := []struct {
-		name string
-		args args
-		want string
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if got := GetConversationIDByMsg(tt.args.msg); got != tt.want {
-				t.Errorf("GetConversationIDByMsg() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-func TestGetConversationIDBySessionType(t *testing.T) {
-	type args struct {
-		sessionType int
-		ids         []string
-	}
-	tests := []struct {
-		name string
-		args args
-		want string
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if got := GetConversationIDBySessionType(tt.args.sessionType, tt.args.ids...); got != tt.want {
-				t.Errorf("GetConversationIDBySessionType() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-func TestGetNotificationConversationIDByConversationID(t *testing.T) {
-	type args struct {
-		conversationID string
-	}
-	tests := []struct {
-		name string
-		args args
-		want string
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if got := GetNotificationConversationIDByConversationID(tt.args.conversationID); got != tt.want {
-				t.Errorf("GetNotificationConversationIDByConversationID() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-func TestGetNotificationConversationID(t *testing.T) {
-	type args struct {
-		sessionType int
-		ids         []string
-	}
-	tests := []struct {
-		name string
-		args args
-		want string
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if got := GetNotificationConversationID(tt.args.sessionType, tt.args.ids...); got != tt.want {
-				t.Errorf("GetNotificationConversationID() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-func TestIsNotification(t *testing.T) {
-	type args struct {
-		conversationID string
-	}
-	tests := []struct {
-		name string
-		args args
-		want bool
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if got := IsNotification(tt.args.conversationID); got != tt.want {
-				t.Errorf("IsNotification() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-func TestIsNotificationByMsg(t *testing.T) {
-	type args struct {
-		msg *sdkws.MsgData
-	}
-	tests := []struct {
-		name string
-		args args
-		want bool
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if got := IsNotificationByMsg(tt.args.msg); got != tt.want {
-				t.Errorf("IsNotificationByMsg() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-func TestParseConversationID(t *testing.T) {
-	type args struct {
-		msg *sdkws.MsgData
-	}
-	tests := []struct {
-		name               string
-		args               args
-		wantIsNotification bool
-		wantConversationID string
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			gotIsNotification, gotConversationID := ParseConversationID(tt.args.msg)
-			if gotIsNotification != tt.wantIsNotification {
-				t.Errorf("ParseConversationID() gotIsNotification = %v, want %v", gotIsNotification, tt.wantIsNotification)
-			}
-			if gotConversationID != tt.wantConversationID {
-				t.Errorf("ParseConversationID() gotConversationID = %v, want %v", gotConversationID, tt.wantConversationID)
-			}
-		})
-	}
-}
-
-func TestMsgBySeq_Len(t *testing.T) {
-	tests := []struct {
-		name string
-		s    MsgBySeq
-		want int
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if got := tt.s.Len(); got != tt.want {
-				t.Errorf("MsgBySeq.Len() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-func TestMsgBySeq_Less(t *testing.T) {
-	type args struct {
-		i int
-		j int
-	}
-	tests := []struct {
-		name string
-		s    MsgBySeq
-		args args
-		want bool
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if got := tt.s.Less(tt.args.i, tt.args.j); got != tt.want {
-				t.Errorf("MsgBySeq.Less() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-func TestMsgBySeq_Swap(t *testing.T) {
-	type args struct {
-		i int
-		j int
-	}
-	tests := []struct {
-		name string
-		s    MsgBySeq
-		args args
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			tt.s.Swap(tt.args.i, tt.args.j)
-		})
-	}
-}
-
-func TestPb2String(t *testing.T) {
-	type args struct {
-		pb proto.Message
-	}
-	tests := []struct {
-		name    string
-		args    args
-		want    string
-		wantErr bool
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			got, err := Pb2String(tt.args.pb)
-			if (err != nil) != tt.wantErr {
-				t.Errorf("Pb2String() error = %v, wantErr %v", err, tt.wantErr)
-				return
-			}
-			if got != tt.want {
-				t.Errorf("Pb2String() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-func TestString2Pb(t *testing.T) {
-	type args struct {
-		s  string
-		pb proto.Message
-	}
-	tests := []struct {
-		name    string
-		args    args
-		wantErr bool
-	}{
-		// TODO: Add test cases.
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if err := String2Pb(tt.args.s, tt.args.pb); (err != nil) != tt.wantErr {
-				t.Errorf("String2Pb() error = %v, wantErr %v", err, tt.wantErr)
-			}
-		})
-	}
-}

scripts/README.md

@@ -39,7 +39,7 @@ scripts/
 ├── demo.sh                             # Demonstration or example script.
 ├── docker-check-service.sh             # Docker script to check services' status.
 ├── docker-start-all.sh                 # Docker script to start all containers/services.
-├── ensure-tag.sh                       # Ensure correct tags or labeling.
+├── ensure_tag.sh                       # Ensure correct tags or labeling.
 ├── env_check.sh                        # Environment verification and checking.
 ├── gen-swagger-docs.sh                 # Script to generate Swagger documentation.
 ├── genconfig.sh                        # Generate configuration files.