Compare commits
4 Commits
optimizati
...
sweep/feat
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
20325610b2 | ||
|
|
5fb8598d07 | ||
|
|
2955b251b3 | ||
|
|
c7997329c2 |
@@ -2,6 +2,7 @@
|
||||
// Reference Doc: https://code.visualstudio.com/remote/advancedcontainers/overview
|
||||
"name": "OpenIM Dev Environment",
|
||||
// Update the container version when you publish dev-container
|
||||
"dockerComposeFile": "docker-compose.yml",
|
||||
"build": { "dockerfile": "Dockerfile" },
|
||||
// Replace with uncommented line below to build your own local copy of the image
|
||||
// "dockerFile": "../docker/Dockerfile-dev",
|
||||
|
||||
1
.github/workflows/help-comment-issue.yml
vendored
1
.github/workflows/help-comment-issue.yml
vendored
@@ -17,7 +17,6 @@ on:
|
||||
issues:
|
||||
types:
|
||||
- labeled
|
||||
|
||||
jobs:
|
||||
add-comment:
|
||||
if: github.event.label.name == 'help wanted' || github.event.label.name == 'good first issue'
|
||||
|
||||
15
.github/workflows/openimci.yml
vendored
15
.github/workflows/openimci.yml
vendored
@@ -23,8 +23,6 @@ on:
|
||||
- "docs/**"
|
||||
- "README.md"
|
||||
- "README_zh-CN.md"
|
||||
- "**.md"
|
||||
- "docs/**"
|
||||
- "CONTRIBUTING.md"
|
||||
pull_request:
|
||||
branches:
|
||||
@@ -33,8 +31,7 @@ on:
|
||||
paths-ignore:
|
||||
- "README.md"
|
||||
- "README_zh-CN.md"
|
||||
- "CONTRIBUTING/**"
|
||||
- "**.md"
|
||||
- "CONTRIBUTING.md"
|
||||
- "docs/**"
|
||||
|
||||
env:
|
||||
@@ -70,9 +67,6 @@ jobs:
|
||||
version: '3.x' # If available, use the latest major version that's compatible
|
||||
repo-token: ${{ secrets.GITHUB_TOKEN }}
|
||||
|
||||
- name: OpenIM Scripts Verification(make verify)
|
||||
run: sudo make verify
|
||||
|
||||
- name: Module Operations
|
||||
run: |
|
||||
sudo make tidy
|
||||
@@ -97,6 +91,13 @@ jobs:
|
||||
- name: Cleanup Build
|
||||
run: sudo make clean
|
||||
|
||||
- name: Push Changes to Main
|
||||
uses: stefanzweifel/git-auto-commit-action@v5
|
||||
with:
|
||||
commit_message: "cicd: robot automated Change"
|
||||
branch: main
|
||||
continue-on-error: true
|
||||
|
||||
- name: Set Current Directory
|
||||
id: set_directory
|
||||
run: echo "::set-output name=directory::$(pwd)"
|
||||
|
||||
3
.github/workflows/pull-request.yml
vendored
3
.github/workflows/pull-request.yml
vendored
@@ -78,6 +78,9 @@ jobs:
|
||||
echo "Run unit test and get test coverage successfully"
|
||||
continue-on-error: true
|
||||
|
||||
- name: Perform CodeQL Analysis
|
||||
uses: github/codeql-action/analyze@v2
|
||||
|
||||
- name: OpenIM verify copyright
|
||||
run: |
|
||||
sudo make verify-copyright
|
||||
|
||||
3
.github/workflows/sync-release.yml
vendored
3
.github/workflows/sync-release.yml
vendored
@@ -3,7 +3,7 @@
|
||||
# you may not use this file except in compliance with the License.
|
||||
|
||||
# https://github.com/BetaHuhn/repo-file-sync-action
|
||||
name: Synchronize OpenIM Release Branch Public Code To Other Repositories
|
||||
name: Synchronize kubecub public code to other repositories
|
||||
on:
|
||||
push:
|
||||
paths:
|
||||
@@ -41,4 +41,3 @@ jobs:
|
||||
automerge
|
||||
ASSIGNEES: |
|
||||
kubbot
|
||||
continue-on-error: true
|
||||
5
.github/workflows/sync.yml
vendored
5
.github/workflows/sync.yml
vendored
@@ -3,7 +3,7 @@
|
||||
# you may not use this file except in compliance with the License.
|
||||
|
||||
# https://github.com/BetaHuhn/repo-file-sync-action
|
||||
name: Synchronize OpenIM Main Branch Public Code To Other Repositories
|
||||
name: Synchronize kubecub public code to other repositories
|
||||
on:
|
||||
push:
|
||||
branches:
|
||||
@@ -36,5 +36,4 @@ jobs:
|
||||
file-sync
|
||||
automerge
|
||||
ASSIGNEES: |
|
||||
kubbot
|
||||
continue-on-error: true
|
||||
kubbot
|
||||
1
.gitignore
vendored
1
.gitignore
vendored
@@ -34,7 +34,6 @@ deployments/charts/generated-configs/
|
||||
### OpenIM Config ###
|
||||
.env
|
||||
config/config.yaml
|
||||
config/openim.yaml
|
||||
config/alertmanager.yml
|
||||
config/prometheus.yml
|
||||
config/email.tmpl
|
||||
|
||||
15
CHANGELOG/CHANGELOG-3.5.0.md
Normal file
15
CHANGELOG/CHANGELOG-3.5.0.md
Normal file
@@ -0,0 +1,15 @@
|
||||
# Version logging for OpenIM
|
||||
|
||||
<!-- BEGIN MUNGE: GENERATED_TOC -->
|
||||
|
||||
<!-- END MUNGE: GENERATED_TOC -->
|
||||
|
||||
<a name="unreleased"></a>
|
||||
## [Unreleased]
|
||||
|
||||
|
||||
<a name="v3.5.0+3.97baaac"></a>
|
||||
## [v3.5.0+3.97baaac] - 2024-01-12
|
||||
|
||||
[Unreleased]: https://github.com/openimsdk/open-im-server/compare/v3.5.0+3.97baaac...HEAD
|
||||
[v3.5.0+3.97baaac]: https://github.com/openimsdk/open-im-server/compare/v3.5.0+5.950e970...v3.5.0+3.97baaac
|
||||
@@ -42,7 +42,7 @@ OpenIM is a service platform specifically designed for integrating chat, audio-v
|
||||
- 🛡️ API wrapping
|
||||
- 🌐 Connection management
|
||||
|
||||
+ 📚 Main Modules:
|
||||
## 📚 Main Modules:
|
||||
|
||||
1. 🚀 Initialization and Login
|
||||
2. 👤 User Management
|
||||
@@ -70,16 +70,11 @@ It is built using Golang and supports cross-platform deployment, ensuring a cons
|
||||
|
||||
## :rocket: Quick Start
|
||||
|
||||
We support many platforms. Here are the addresses for quick experience on the web side:
|
||||
|
||||
👉 **[OpenIM online web demo](https://web-enterprise.rentsoft.cn/)**
|
||||
|
||||
🤲 To facilitate user experience, we offer various deployment solutions. You can choose your deployment method from the list below:
|
||||
|
||||
+ **[Source Code Deployment Guide](https://docs.openim.io/guides/gettingStarted/imSourceCodeDeployment)**
|
||||
+ **[Docker Deployment Guide](https://docs.openim.io/guides/gettingStarted/dockerCompose)**
|
||||
+ **[Kubernetes Deployment Guide](https://docs.openim.io/guides/gettingStarted/k8s-deployment)**
|
||||
+ **[Mac Developer Deployment Guide](https://docs.openim.io/guides/gettingstarted/mac-deployment-guide)**
|
||||
|
||||
## :hammer_and_wrench: To Start Developing OpenIM
|
||||
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
|
||||
# -----------------------------------------------------------------
|
||||
# TODO: This config file is the template file
|
||||
# --| source: deployments/templates/config.yaml
|
||||
# --| source: deployments/templates/openim.yaml
|
||||
# --| env: scripts/install/environment
|
||||
# --| target: config/config.yaml
|
||||
# -----------------------------------------------------------------
|
||||
@@ -53,7 +53,7 @@ mongo:
|
||||
# Maximum connection pool size
|
||||
address: [ 172.28.0.1:37017 ]
|
||||
database: openim_v3
|
||||
username: openIM
|
||||
username: root
|
||||
password: openIM123
|
||||
maxPoolSize: 100
|
||||
|
||||
@@ -122,14 +122,14 @@ api:
|
||||
# minio.signEndpoint is minio public network address
|
||||
object:
|
||||
enable: "minio"
|
||||
apiURL: "http://172.28.0.1:10002"
|
||||
apiURL: "http://14.155.64.202:10002"
|
||||
minio:
|
||||
bucket: "openim"
|
||||
endpoint: "http://172.28.0.1:10005"
|
||||
accessKeyID: "root"
|
||||
secretAccessKey: "openIM123"
|
||||
sessionToken: ''
|
||||
signEndpoint: "http://172.28.0.1:10005"
|
||||
signEndpoint: "http://14.155.64.202:10005"
|
||||
publicRead: false
|
||||
cos:
|
||||
bucketURL: https://temp-1252357374.cos.ap-chengdu.myqcloud.com
|
||||
@@ -193,7 +193,7 @@ rpcRegisterName:
|
||||
# Whether to output in json format
|
||||
# Whether to include stack trace in logs
|
||||
log:
|
||||
storageLocation: /workspaces/open-im-server/logs/
|
||||
storageLocation: /data/workspaces/open-im-server/logs/
|
||||
rotationTime: 24
|
||||
remainRotationCount: 2
|
||||
remainLogLevel: 6
|
||||
@@ -247,14 +247,6 @@ manager:
|
||||
userID: [ "openIM123456", "openIM654321", "openIMAdmin" ]
|
||||
nickname: [ "system1", "system2", "system3" ]
|
||||
|
||||
# chatAdmin, use for send notification
|
||||
#
|
||||
# Built-in app system notification account ID
|
||||
# Built-in app system notification account nickname
|
||||
im-admin:
|
||||
userID: [ "imAdmin" ]
|
||||
nickname: [ "imAdmin" ]
|
||||
|
||||
# Multi-platform login policy
|
||||
# For each platform(Android, iOS, Windows, Mac, web), only one can be online at a time
|
||||
multiLoginPolicy: 1
|
||||
@@ -315,21 +307,21 @@ iosPush:
|
||||
# Timeout in seconds
|
||||
# Whether to continue execution if callback fails
|
||||
callback:
|
||||
url: "http://127.0.0.1:10008/callbackExample"
|
||||
url: ""
|
||||
beforeSendSingleMsg:
|
||||
enable: false
|
||||
timeout: 5
|
||||
failedContinue: true
|
||||
beforeUpdateUserInfoEx:
|
||||
enable: false
|
||||
enable: false
|
||||
timeout: 5
|
||||
failedContinue: true
|
||||
afterUpdateUserInfoEx:
|
||||
enable: false
|
||||
enable: false
|
||||
timeout: 5
|
||||
failedContinue: true
|
||||
afterSendSingleMsg:
|
||||
enable: true
|
||||
enable: false
|
||||
timeout: 5
|
||||
failedContinue: true
|
||||
beforeSendGroupMsg:
|
||||
@@ -513,8 +505,8 @@ 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: true
|
||||
grafanaUrl: http://172.28.0.1:13000/
|
||||
enable: false
|
||||
grafanaUrl: 172.28.0.1:13000
|
||||
apiPrometheusPort: [20100]
|
||||
userPrometheusPort: [ 20110 ]
|
||||
friendPrometheusPort: [ 20120 ]
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
# Copyright © 2024 OpenIM. All rights reserved.
|
||||
# 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.
|
||||
@@ -12,26 +12,31 @@
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# General Configuration
|
||||
# This section contains general configuration options for the entire environment.
|
||||
# These options can be set via environment variables. If both environment variables
|
||||
# and settings in this .env file exist, the environment variables take precedence.
|
||||
# -----------------------------------------------------------------------------
|
||||
# ==========================
|
||||
# General Configuration
|
||||
# ==========================
|
||||
# These settings apply to the overall environment.
|
||||
# ======================================
|
||||
# ========= Basic Configuration ========
|
||||
# ======================================
|
||||
|
||||
# Data storage directory for persistent data.
|
||||
# Example: DATA_DIR=/path/to/data
|
||||
DATA_DIR=/workspaces/open-im-server
|
||||
# The user for authentication or system operations.
|
||||
# Default: OPENIM_USER=root
|
||||
USER=root
|
||||
|
||||
# Docker image registry. Uncomment the preferred one.
|
||||
# Options: ghcr.io/openimsdk, openim, registry.cn-hangzhou.aliyuncs.com/openimsdk
|
||||
# IMAGE_REGISTRY="ghcr.io/openimsdk"
|
||||
# IMAGE_REGISTRY="openim"
|
||||
# IMAGE_REGISTRY="registry.cn-hangzhou.aliyuncs.com/openimsdk"
|
||||
# Password associated with the specified user for authentication.
|
||||
# Default: PASSWORD=openIM123
|
||||
PASSWORD=openIM123
|
||||
|
||||
# Base URL for the application programming interface (API).
|
||||
# Default: API_URL=http://172.28.0.1:10002
|
||||
API_URL=http://14.155.64.202:10002
|
||||
|
||||
# Directory path for storing data files or related information.
|
||||
# Default: DATA_DIR=./
|
||||
DATA_DIR=/data/workspaces/open-im-server
|
||||
|
||||
# Choose the appropriate image address, the default is GITHUB image,
|
||||
# you can choose docker hub, for Chinese users can choose Ali Cloud
|
||||
# export IMAGE_REGISTRY="ghcr.io/openimsdk"
|
||||
# export IMAGE_REGISTRY="openim"
|
||||
# export IMAGE_REGISTRY="registry.cn-hangzhou.aliyuncs.com/openimsdk"
|
||||
IMAGE_REGISTRY=ghcr.io/openimsdk
|
||||
|
||||
# ======================================
|
||||
@@ -42,9 +47,10 @@ IMAGE_REGISTRY=ghcr.io/openimsdk
|
||||
# Default: DOCKER_BRIDGE_SUBNET=172.28.0.0/16
|
||||
DOCKER_BRIDGE_SUBNET=172.28.0.0/16
|
||||
|
||||
# Set and specify the IP addresses of some containers. Generally speaking,
|
||||
# you do not need to modify these configurations to facilitate debugging
|
||||
# Gateway for the Docker network.
|
||||
# Default: DOCKER_BRIDGE_GATEWAY=172.28.0.1
|
||||
DOCKER_BRIDGE_GATEWAY=172.28.0.1
|
||||
|
||||
MONGO_NETWORK_ADDRESS=172.28.0.2
|
||||
REDIS_NETWORK_ADDRESS=172.28.0.3
|
||||
KAFKA_NETWORK_ADDRESS=172.28.0.4
|
||||
@@ -59,66 +65,45 @@ NODE_EXPORTER_NETWORK_ADDRESS=172.28.0.12
|
||||
OPENIM_ADMIN_FRONT_NETWORK_ADDRESS=172.28.0.13
|
||||
ALERT_MANAGER_NETWORK_ADDRESS=172.28.0.14
|
||||
|
||||
# ==============================================================================
|
||||
# Configuration Update Instructions
|
||||
# ==============================================================================
|
||||
# This header outlines the methods to update common variables in config.yaml and .env files.
|
||||
# These instructions are vital for maintaining the OpenIM environment's configuration.
|
||||
#
|
||||
# METHOD 1: Regenerate All Configurations
|
||||
# ----------------------------------------
|
||||
# Use this method to regenerate all configurations.
|
||||
# Steps:
|
||||
# 1. Delete existing config files:
|
||||
# - openim-server/config/config.yaml
|
||||
# - openim-chat/config/config.yaml
|
||||
# 2. Modify the .env file as required.
|
||||
# 3. Run 'docker compose up -d'. This will regenerate:
|
||||
# - config/config.yaml
|
||||
#
|
||||
# METHOD 2: Modify Individual Configuration Files
|
||||
# -----------------------------------------------
|
||||
# Use this method to update specific configuration files.
|
||||
# Steps:
|
||||
# 1. Modify the .env file as necessary.
|
||||
# 2. Update the corresponding entries in:
|
||||
# - config/config.yaml
|
||||
# 3. Restart the services with 'docker compose up -d'.
|
||||
# 4. Special Note: If you modify OPENIM_IP, API_OPENIM_PORT, or MINIO_PORT in .env,
|
||||
# ensure to update the corresponding services and configurations accordingly.
|
||||
#
|
||||
# It is essential to follow these methods to ensure consistent and correct application behavior.
|
||||
# ==============================================================================
|
||||
# Local IP address of the service. Modify if necessary.
|
||||
# Example: OPENIM_IP=172.28.0.1,
|
||||
OPENIM_IP=172.28.0.1
|
||||
# ===============================================
|
||||
# = Component Extension Configuration =
|
||||
# ===============================================
|
||||
|
||||
# ============ Component Extension Configuration ==========
|
||||
# ----- ZooKeeper Configuration -----
|
||||
# Address or hostname for the ZooKeeper service.
|
||||
# Default: ZOOKEEPER_ADDRESS=172.28.0.1
|
||||
ZOOKEEPER_ADDRESS=172.28.0.5
|
||||
|
||||
# Port for ZooKeeper service.
|
||||
# Default: ZOOKEEPER_PORT=12181
|
||||
ZOOKEEPER_PORT=12181
|
||||
|
||||
# MongoDB service port configuration.
|
||||
# ----- MongoDB Configuration -----
|
||||
# Address or hostname for the MongoDB service.
|
||||
# Default: MONGO_ADDRESS=172.28.0.1
|
||||
MONGO_ADDRESS=172.28.0.2
|
||||
|
||||
# Port on which MongoDB service is running.
|
||||
# Default: MONGO_PORT=37017
|
||||
# MONGO_PORT=37017
|
||||
|
||||
# Password for MongoDB admin user. Used for service authentication.
|
||||
# Username to authenticate with the MongoDB service.
|
||||
# Default: MONGO_USERNAME=root
|
||||
# MONGO_USERNAME=root
|
||||
|
||||
# Password to authenticate with the MongoDB service.
|
||||
# Default: MONGO_PASSWORD=openIM123
|
||||
MONGO_PASSWORD=openIM123
|
||||
|
||||
# Username for a regular OpenIM user in MongoDB.
|
||||
# Default: MONGO_OPENIM_USERNAME=openIM
|
||||
MONGO_OPENIM_USERNAME=openIM
|
||||
|
||||
# Password for a regular OpenIM user in MongoDB.
|
||||
# Default: MONGO_OPENIM_PASSWORD=openIM123456
|
||||
MONGO_OPENIM_PASSWORD=openIM123
|
||||
|
||||
# Specifies the database name to be used within MongoDB.
|
||||
# Name of the database in MongoDB to be used.
|
||||
# Default: MONGO_DATABASE=openim_v3
|
||||
MONGO_DATABASE=openim_v3
|
||||
|
||||
# ----- Redis Configuration -----
|
||||
# Address or hostname for the Redis service.
|
||||
# Default: REDIS_ADDRESS=172.28.0.1
|
||||
REDIS_ADDRESS=172.28.0.3
|
||||
|
||||
# Port on which Redis in-memory data structure store is running.
|
||||
# Default: REDIS_PORT=16379
|
||||
@@ -128,6 +113,11 @@ REDIS_PORT=16379
|
||||
# Default: REDIS_PASSWORD=openIM123
|
||||
REDIS_PASSWORD=openIM123
|
||||
|
||||
# ----- Kafka Configuration -----
|
||||
# Address or hostname for the Kafka service.
|
||||
# Default: KAFKA_ADDRESS=172.28.0.1
|
||||
KAFKA_ADDRESS=172.28.0.4
|
||||
|
||||
# Kakfa username to authenticate with the Kafka service.
|
||||
# KAFKA_USERNAME=''
|
||||
|
||||
@@ -139,13 +129,20 @@ KAFKA_PORT=19094
|
||||
# Default: KAFKA_LATESTMSG_REDIS_TOPIC=latestMsgToRedis
|
||||
KAFKA_LATESTMSG_REDIS_TOPIC=latestMsgToRedis
|
||||
|
||||
# MINIO_PORT
|
||||
# ----------
|
||||
# MINIO_PORT sets the port for the MinIO object storage service.
|
||||
# Upon changing this port, the MinIO endpoint URLs in the file must be updated
|
||||
# to reflect this change. The endpoints include both the 'endpoint' and 'signEndpoint'
|
||||
# under the MinIO configuration.
|
||||
#
|
||||
# Topic in Kafka for pushing messages (e.g. notifications or updates).
|
||||
# Default: KAFKA_MSG_PUSH_TOPIC=msgToPush
|
||||
KAFKA_MSG_PUSH_TOPIC=msgToPush
|
||||
|
||||
# Topic in Kafka for storing offline messages in MongoDB.
|
||||
# Default: KAFKA_OFFLINEMSG_MONGO_TOPIC=offlineMsgToMongoMysql
|
||||
KAFKA_OFFLINEMSG_MONGO_TOPIC=offlineMsgToMongoMysql
|
||||
|
||||
# ----- MinIO Configuration ----
|
||||
# Address or hostname for the MinIO object storage service.
|
||||
# Default: MINIO_ADDRESS=172.28.0.1
|
||||
MINIO_ADDRESS=172.28.0.6
|
||||
|
||||
# Port on which MinIO object storage service is running.
|
||||
# Default: MINIO_PORT=10005
|
||||
MINIO_PORT=10005
|
||||
|
||||
@@ -158,11 +155,19 @@ MINIO_PORT=10005
|
||||
MINIO_SECRET_KEY=openIM123
|
||||
|
||||
# ----- Prometheus Configuration -----
|
||||
# Address or hostname for the Prometheus service.
|
||||
# Default: PROMETHEUS_ADDRESS=172.28.0.1
|
||||
PROMETHEUS_ADDRESS=172.28.0.10
|
||||
|
||||
# Port on which Prometheus service is running.
|
||||
# Default: PROMETHEUS_PORT=19090
|
||||
PROMETHEUS_PORT=19090
|
||||
|
||||
# ----- Grafana Configuration -----
|
||||
# Address or hostname for the Grafana service.
|
||||
# Default: GRAFANA_ADDRESS=172.28.0.1
|
||||
GRAFANA_ADDRESS=172.28.0.11
|
||||
|
||||
# Port on which Grafana service is running.
|
||||
# Default: GRAFANA_PORT=13000
|
||||
GRAFANA_PORT=13000
|
||||
@@ -178,34 +183,41 @@ OPENIM_WEB_PORT=11001
|
||||
# ======================================
|
||||
# ========= OpenIM Server ==============
|
||||
# ======================================
|
||||
|
||||
# Address or hostname for the OpenIM server.
|
||||
# Default: OPENIM_SERVER_ADDRESS=172.28.0.1
|
||||
OPENIM_SERVER_ADDRESS=172.28.0.8
|
||||
|
||||
# Port for the OpenIM WebSockets.
|
||||
# Default: OPENIM_WS_PORT=10001
|
||||
OPENIM_WS_PORT=10001
|
||||
|
||||
# API_OPENIM_PORT
|
||||
# ---------------
|
||||
# This variable defines the port on which the OpenIM API service will listen.
|
||||
# When changing this port, it's essential to update the apiURL in the config.yaml file
|
||||
# to ensure the API service is accessible at the new port.
|
||||
#
|
||||
# Port for the OpenIM API.
|
||||
# Default: API_OPENIM_PORT=10002
|
||||
API_OPENIM_PORT=10002
|
||||
|
||||
|
||||
# ======================================
|
||||
# ========== OpenIM Chat ===============
|
||||
# ======================================
|
||||
|
||||
# Branch name for OpenIM chat.
|
||||
# Default: CHAT_IMAGE_VERSION=main
|
||||
# https://github.com/openimsdk/open-im-server/blob/main/docs/contrib/version.md
|
||||
CHAT_IMAGE_VERSION=main
|
||||
|
||||
# Address or hostname for the OpenIM chat service.
|
||||
# Default: OPENIM_CHAT_ADDRESS=172.28.0.1
|
||||
OPENIM_CHAT_ADDRESS=172.28.0.9
|
||||
|
||||
# Port for the OpenIM chat API.
|
||||
# Default: OPENIM_CHAT_API_PORT=10008
|
||||
OPENIM_CHAT_API_PORT=10008
|
||||
|
||||
# Port for the OpenIM admin API.
|
||||
# Default: OPENIM_ADMIN_API_PORT=10009
|
||||
OPENIM_ADMIN_API_PORT=10009
|
||||
# Directory path for storing data files or related information for OpenIM chat.
|
||||
# Default: OPENIM_CHAT_DATA_DIR=./openim-chat/main
|
||||
OPENIM_CHAT_DATA_DIR=./openim-chat/main
|
||||
|
||||
|
||||
# ======================================
|
||||
# ========== OpenIM Admin ==============
|
||||
@@ -214,6 +226,10 @@ OPENIM_ADMIN_API_PORT=10009
|
||||
# Branch name for OpenIM server.
|
||||
# Default: SERVER_IMAGE_VERSION=main
|
||||
SERVER_IMAGE_VERSION=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
|
||||
|
||||
@@ -169,4 +169,4 @@ server {
|
||||
|
||||
proxy_pass http://minio_console_2; # This uses the upstream directive definition to load balance
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1471,4 +1471,4 @@
|
||||
"uid": "f5f5de9a-6ec5-499a-841e-6e901c33b1f7",
|
||||
"version": 16,
|
||||
"weekStart": ""
|
||||
}
|
||||
}
|
||||
@@ -165,7 +165,7 @@ export MINIO_ENDPOINT="http://im-minio:9000"
|
||||
export MINIO_SIGN_ENDPOINT="https://openim.server.com/im-minio-api"
|
||||
|
||||
mkdir ./charts/generated-configs
|
||||
../scripts/genconfig.sh ../scripts/install/environment.sh ./templates/config.yaml > ./charts/generated-configs/config.yaml
|
||||
../scripts/genconfig.sh ../scripts/install/environment.sh ./templates/openim.yaml > ./charts/generated-configs/config.yaml
|
||||
cp ../config/notification.yaml ./charts/generated-configs/notification.yaml
|
||||
../scripts/genconfig.sh ../scripts/install/environment.sh ./templates/helm-image.yaml > ./charts/generated-configs/helm-image.yaml
|
||||
```
|
||||
|
||||
1276
deployments/openim-charts.yaml
Normal file
1276
deployments/openim-charts.yaml
Normal file
File diff suppressed because it is too large
Load Diff
@@ -14,7 +14,7 @@
|
||||
|
||||
# -----------------------------------------------------------------
|
||||
# TODO: This config file is the template file
|
||||
# --| source: deployments/templates/config.yaml
|
||||
# --| source: deployments/templates/openim.yaml
|
||||
# --| env: scripts/install/environment
|
||||
# --| target: config/config.yaml
|
||||
# -----------------------------------------------------------------
|
||||
@@ -527,39 +527,3 @@ prometheus:
|
||||
rtcPrometheusPort: [ ${RTC_PROM_PORT} ]
|
||||
thirdPrometheusPort: [ ${THIRD_PROM_PORT} ]
|
||||
messageTransferPrometheusPort: [ ${MSG_TRANSFER_PROM_PORT} ] # List of ports
|
||||
|
||||
###################### LocalCache configuration information ######################
|
||||
# topic: redis subscriber channel
|
||||
# slotNum: number of slots, multiple slots can prevent too many keys from competing for a lock
|
||||
# slotSize: number of slots, the number of cached keys per slot, the overall cache quantity is slotNum * slotSize
|
||||
# successExpire: successful cache time seconds
|
||||
# failedExpire: failed cache time seconds
|
||||
# disable local caching and annotate topic, slotNum, and slotSize
|
||||
localCache:
|
||||
user:
|
||||
topic: DELETE_CACHE_USER
|
||||
slotNum: 100
|
||||
slotSize: 2000
|
||||
successExpire: 300
|
||||
failedExpire: 5
|
||||
|
||||
group:
|
||||
topic: DELETE_CACHE_GROUP
|
||||
slotNum: 100
|
||||
slotSize: 2000
|
||||
successExpire: 300
|
||||
failedExpire: 5
|
||||
|
||||
friend:
|
||||
topic: DELETE_CACHE_FRIEND
|
||||
slotNum: 100
|
||||
slotSize: 2000
|
||||
successExpire: 300
|
||||
failedExpire: 5
|
||||
|
||||
conversation:
|
||||
topic: DELETE_CACHE_CONVERSATION
|
||||
slotNum: 100
|
||||
slotSize: 2000
|
||||
successExpire: 300
|
||||
failedExpire: 5
|
||||
@@ -214,7 +214,6 @@ services:
|
||||
# - "${OPENIM_ADMIN_API_PORT:-10009}:10009"
|
||||
# volumes:
|
||||
# - "${DATA_DIR:-./}/components/openim-chat/logs:/openim/openim-chat/logs"
|
||||
# - "${DATA_DIR:-./}/components/openim-chat/_output/logs:/openim/openim-chat/_output/logs"
|
||||
# - "${DATA_DIR:-./}/components/openim-chat/config:/openim/openim-chat/config"
|
||||
# restart: always
|
||||
# # user: root:root
|
||||
|
||||
@@ -31,7 +31,7 @@ docs/guide/en-US/cmd/openim/openim-rpc-user_list.md
|
||||
docs/guide/en-US/cmd/openim/openim-rpc-user_update.md
|
||||
docs/guide/en-US/cmd/openim/openim_validate.md
|
||||
docs/guide/en-US/cmd/openim/openim_version.md
|
||||
docs/guide/en-US/yaml/openim/config.yaml
|
||||
docs/guide/en-US/yaml/openim/openim.yaml
|
||||
docs/guide/en-US/yaml/openim/openim_color.yaml
|
||||
docs/guide/en-US/yaml/openim/openim_completion.yaml
|
||||
docs/guide/en-US/yaml/openim/openim_info.yaml
|
||||
|
||||
@@ -89,7 +89,7 @@ While the first two methods will be our main focus, it's worth noting that the t
|
||||
|
||||
### 1.2. <a name='SourceCodeDeployment'></a>Source Code Deployment
|
||||
|
||||
In the source code deployment method, the configuration generation process involves executing `make init`, which fundamentally runs the script `./scripts/init-config.sh`. This script utilizes variables defined in the [`environment.sh`](https://github.com/openimsdk/open-im-server/blob/main/scripts/install/environment.sh) script to render the [`config.yaml`](https://github.com/openimsdk/open-im-server/blob/main/deployments/templates/config.yaml) template file, subsequently generating the [`config.yaml`](https://github.com/openimsdk/open-im-server/blob/main/config/config.yaml) configuration file.
|
||||
In the source code deployment method, the configuration generation process involves executing `make init`, which fundamentally runs the script `./scripts/init-config.sh`. This script utilizes variables defined in the [`environment.sh`](https://github.com/openimsdk/open-im-server/blob/main/scripts/install/environment.sh) script to render the [`openim.yaml`](https://github.com/openimsdk/open-im-server/blob/main/deployments/templates/openim.yaml) template file, subsequently generating the [`config.yaml`](https://github.com/openimsdk/open-im-server/blob/main/config/config.yaml) configuration file.
|
||||
|
||||
### 1.3. <a name='DockerComposeDeployment'></a>Docker Compose Deployment
|
||||
|
||||
|
||||
@@ -45,6 +45,70 @@ Homebrew is an essential package manager for macOS. Install it using:
|
||||
git config --global user.email "your.email@example.com"
|
||||
```
|
||||
|
||||
HEAD
|
||||
=======
|
||||
### Forking and Cloning the Repository
|
||||
|
||||
To optimize and add logic to your instructions, particularly regarding deploying on a Mac, you can modify them as follows:
|
||||
|
||||
1. **Fork the OpenIM Repository**: Fork the OpenIM repository on GitHub to your account.
|
||||
|
||||
2. **Clone Your Fork to Your Local Machine**:
|
||||
Open your terminal and execute the following commands:
|
||||
|
||||
```sh
|
||||
git clone https://github.com/<your-username>/open-im-server.git
|
||||
|
||||
# Set Docker bridging network mode for Mac
|
||||
export DOCKER_BRIDGE_SUBNET=127.0.0.0/16
|
||||
export OPENIM_IP=<your-ip>
|
||||
make init
|
||||
docker compose up -d
|
||||
make start
|
||||
git clone https://github.com/<your-username>/open-im-server.git
|
||||
|
||||
# Set Docker bridging network mode for Mac
|
||||
export DOCKER_BRIDGE_SUBNET=127.0.0.0/16
|
||||
|
||||
# Set OpenIM IP
|
||||
export OPENIM_IP=<your-ip>
|
||||
|
||||
# Initialize configuration
|
||||
make init
|
||||
|
||||
# Start components using Docker
|
||||
docker compose up -d
|
||||
|
||||
# Start OpenIM Server
|
||||
make start
|
||||
```
|
||||
|
||||
3. **Additional Steps for Mac Deployment**:
|
||||
If you are deploying on a Mac and wish to use the chat feature, you need to modify the `docker-compose` file. Specifically, you'll need to uncomment the sections related to `openim-chat` and `mysql`.
|
||||
|
||||
Here's how to do it:
|
||||
|
||||
- Open the `docker-compose.yml` file in a text editor.
|
||||
- Find the sections for `openim-chat` and `mysql`.
|
||||
- Remove the comment marks (`#`) at the beginning of the lines in these sections to enable them.
|
||||
- Save the file after making these changes.
|
||||
|
||||
4. **Update and Restart Services**:
|
||||
After modifying the `docker-compose` file, you need to update and restart the services to apply these changes. Run the following command in your terminal:
|
||||
|
||||
```sh
|
||||
# Update and restart services
|
||||
docker compose up -d
|
||||
|
||||
# Check openim-chat start
|
||||
docker compose logs openim-chat
|
||||
```
|
||||
|
||||
This command will re-read the `docker-compose.yml` file, apply the new configuration, and restart the necessary containers.
|
||||
|
||||
Remember, replacing `<your-username>` and `<your-ip>` with your actual GitHub username and desired IP address for OpenIM is crucial. These steps should streamline the setup process, particularly for Mac users wishing to use the chat feature.
|
||||
|
||||
>>>>>>> origin/main
|
||||
### Setting Up the Devcontainer
|
||||
|
||||
`Devcontainers` provide a Docker-based isolated development environment.
|
||||
@@ -185,9 +249,12 @@ make start
|
||||
To check, ensure the following four processes start successfully:
|
||||
|
||||
```bash
|
||||
make check
|
||||
➜ chat git:(main) ✗ ps -ef | grep chat
|
||||
# [Output showing four running chat processes]
|
||||
```
|
||||
|
||||
Access the web interface: http://ip:11001
|
||||
|
||||
### Docker Deployment
|
||||
|
||||
Refer to https://github.com/openimsdk/openim-docker for Docker deployment instructions, which can be followed similarly on Linux.
|
||||
|
||||
10
go.mod
10
go.mod
@@ -4,8 +4,8 @@ go 1.19
|
||||
|
||||
require (
|
||||
firebase.google.com/go v3.13.0+incompatible
|
||||
github.com/OpenIMSDK/protocol v0.0.49
|
||||
github.com/OpenIMSDK/tools v0.0.28
|
||||
github.com/OpenIMSDK/protocol v0.0.48
|
||||
github.com/OpenIMSDK/tools v0.0.23
|
||||
github.com/bwmarrin/snowflake v0.3.0 // indirect
|
||||
github.com/dtm-labs/rockscache v0.1.1
|
||||
github.com/gin-gonic/gin v1.9.1
|
||||
@@ -18,7 +18,6 @@ require (
|
||||
github.com/minio/minio-go/v7 v7.0.63
|
||||
github.com/mitchellh/mapstructure v1.5.0
|
||||
github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646 // indirect
|
||||
github.com/openimsdk/localcache v0.0.1
|
||||
github.com/pkg/errors v0.9.1 // indirect
|
||||
github.com/prometheus/client_golang v1.17.0
|
||||
github.com/robfig/cron/v3 v3.0.1
|
||||
@@ -39,7 +38,6 @@ require (
|
||||
github.com/aliyun/aliyun-oss-go-sdk v2.2.9+incompatible
|
||||
github.com/go-redis/redis v6.15.9+incompatible
|
||||
github.com/redis/go-redis/v9 v9.2.1
|
||||
github.com/spf13/pflag v1.0.5
|
||||
github.com/stathat/consistent v1.0.0
|
||||
github.com/tencentyun/cos-go-sdk-v5 v0.7.45
|
||||
go.uber.org/automaxprocs v1.5.3
|
||||
@@ -83,7 +81,6 @@ require (
|
||||
github.com/hashicorp/errwrap v1.0.0 // indirect
|
||||
github.com/hashicorp/go-multierror v1.1.1 // indirect
|
||||
github.com/hashicorp/go-uuid v1.0.3 // indirect
|
||||
github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
|
||||
github.com/inconshreveable/mousetrap v1.1.0 // indirect
|
||||
github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 // indirect
|
||||
github.com/jcmturner/aescts/v2 v2.0.0 // indirect
|
||||
@@ -119,6 +116,7 @@ require (
|
||||
github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
|
||||
github.com/rs/xid v1.5.0 // indirect
|
||||
github.com/sergi/go-diff v1.0.0 // indirect
|
||||
github.com/spf13/pflag v1.0.5 // indirect
|
||||
github.com/src-d/gcfg v1.4.0 // indirect
|
||||
github.com/stretchr/objx v0.5.0 // indirect
|
||||
github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
|
||||
@@ -159,4 +157,4 @@ require (
|
||||
gopkg.in/ini.v1 v1.67.0 // indirect
|
||||
)
|
||||
|
||||
replace github.com/openimsdk/localcache => ./pkg/localcache
|
||||
replace github.com/OpenIMSDK/protocol v0.0.47 => github.com/AndrewZuo01/protocol v0.0.0-20240112093520-fd9c53e27b94
|
||||
|
||||
10
go.sum
10
go.sum
@@ -15,13 +15,13 @@ cloud.google.com/go/storage v1.30.1 h1:uOdMxAs8HExqBlnLtnQyP0YkvbiDpdGShGKtx6U/o
|
||||
cloud.google.com/go/storage v1.30.1/go.mod h1:NfxhC0UJE1aXSx7CIIbCf7y9HKT7BiccwkR7+P7gN8E=
|
||||
firebase.google.com/go v3.13.0+incompatible h1:3TdYC3DDi6aHn20qoRkxwGqNgdjtblwVAyRLQwGn/+4=
|
||||
firebase.google.com/go v3.13.0+incompatible/go.mod h1:xlah6XbEyW6tbfSklcfe5FHJIwjt8toICdV5Wh9ptHs=
|
||||
github.com/AndrewZuo01/protocol v0.0.0-20240112093520-fd9c53e27b94 h1:o86vkek41ZrQqoBGqyKvS0z6N0uJj64mpzK72OkDZVM=
|
||||
github.com/AndrewZuo01/protocol v0.0.0-20240112093520-fd9c53e27b94/go.mod h1:F25dFrwrIx3lkNoiuf6FkCfxuwf8L4Z8UIsdTHP/r0Y=
|
||||
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
|
||||
github.com/IBM/sarama v1.41.3 h1:MWBEJ12vHC8coMjdEXFq/6ftO6DUZnQlFYcxtOJFa7c=
|
||||
github.com/IBM/sarama v1.41.3/go.mod h1:Xxho9HkHd4K/MDUo/T/sOqwtX/17D33++E9Wib6hUdQ=
|
||||
github.com/OpenIMSDK/protocol v0.0.49 h1:wcqJOMBis7f153zNI7V82Fc4WyqA1GanMgXUQgL618k=
|
||||
github.com/OpenIMSDK/protocol v0.0.49/go.mod h1:F25dFrwrIx3lkNoiuf6FkCfxuwf8L4Z8UIsdTHP/r0Y=
|
||||
github.com/OpenIMSDK/tools v0.0.27 h1:J/kSRqM+y9U4XK/pQ9RkEB31oQ5BTYD1oA5r1PITPRA=
|
||||
github.com/OpenIMSDK/tools v0.0.27/go.mod h1:eg+q4A34Qmu73xkY0mt37FHGMCMfC6CtmOnm0kFEGFI=
|
||||
github.com/OpenIMSDK/tools v0.0.23 h1:xozfrGzhbpNPlDTap5DLVPk+JfgZ/ZyIj4Cuu3/bm9w=
|
||||
github.com/OpenIMSDK/tools v0.0.23/go.mod h1:eg+q4A34Qmu73xkY0mt37FHGMCMfC6CtmOnm0kFEGFI=
|
||||
github.com/QcloudApi/qcloud_sign_golang v0.0.0-20141224014652-e4130a326409/go.mod h1:1pk82RBxDY/JZnPQrtqHlUFfCctgdorsd9M06fMynOM=
|
||||
github.com/alcortesm/tgz v0.0.0-20161220082320-9c5fe88206d7 h1:uSoVVbwJiQipAclBbw+8quDsfcvFjOpI5iCf4p/cqCs=
|
||||
github.com/alcortesm/tgz v0.0.0-20161220082320-9c5fe88206d7/go.mod h1:6zEj6s6u/ghQa61ZWa/C2Aw3RkjiTBOix7dkqa1VLIs=
|
||||
@@ -171,8 +171,6 @@ github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9
|
||||
github.com/hashicorp/go-uuid v1.0.2/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
|
||||
github.com/hashicorp/go-uuid v1.0.3 h1:2gKiV6YVmrJ1i2CKKa9obLvRieoRGviZFL26PcT/Co8=
|
||||
github.com/hashicorp/go-uuid v1.0.3/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
|
||||
github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
|
||||
github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
|
||||
github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
|
||||
github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
|
||||
github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
|
||||
|
||||
@@ -16,7 +16,6 @@ package api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
|
||||
"github.com/OpenIMSDK/protocol/constant"
|
||||
@@ -44,7 +43,7 @@ import (
|
||||
)
|
||||
|
||||
func NewGinRouter(discov discoveryregistry.SvcDiscoveryRegistry, rdb redis.UniversalClient) *gin.Engine {
|
||||
discov.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin"))) // 默认RPC中间件
|
||||
discov.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials())) // 默认RPC中间件
|
||||
gin.SetMode(gin.ReleaseMode)
|
||||
r := gin.New()
|
||||
if v, ok := binding.Validator.Engine().(*validator.Validate); ok {
|
||||
|
||||
@@ -201,27 +201,27 @@ func (u *UserApi) GetSubscribeUsersStatus(c *gin.Context) {
|
||||
a2r.Call(user.UserClient.GetSubscribeUsersStatus, u.Client, c)
|
||||
}
|
||||
|
||||
// ProcessUserCommandAdd user general function add.
|
||||
// ProcessUserCommandAdd user general function add
|
||||
func (u *UserApi) ProcessUserCommandAdd(c *gin.Context) {
|
||||
a2r.Call(user.UserClient.ProcessUserCommandAdd, u.Client, c)
|
||||
}
|
||||
|
||||
// ProcessUserCommandDelete user general function delete.
|
||||
// ProcessUserCommandDelete user general function delete
|
||||
func (u *UserApi) ProcessUserCommandDelete(c *gin.Context) {
|
||||
a2r.Call(user.UserClient.ProcessUserCommandDelete, u.Client, c)
|
||||
}
|
||||
|
||||
// ProcessUserCommandUpdate user general function update.
|
||||
// ProcessUserCommandUpdate user general function update
|
||||
func (u *UserApi) ProcessUserCommandUpdate(c *gin.Context) {
|
||||
a2r.Call(user.UserClient.ProcessUserCommandUpdate, u.Client, c)
|
||||
}
|
||||
|
||||
// ProcessUserCommandGet user general function get.
|
||||
// ProcessUserCommandGet user general function get
|
||||
func (u *UserApi) ProcessUserCommandGet(c *gin.Context) {
|
||||
a2r.Call(user.UserClient.ProcessUserCommandGet, u.Client, c)
|
||||
}
|
||||
|
||||
// ProcessUserCommandGet user general function get all.
|
||||
// ProcessUserCommandGet user general function get all
|
||||
func (u *UserApi) ProcessUserCommandGetAll(c *gin.Context) {
|
||||
a2r.Call(user.UserClient.ProcessUserCommandGetAll, u.Client, c)
|
||||
}
|
||||
|
||||
@@ -105,7 +105,7 @@ func CallbackUserKickOff(ctx context.Context, userID string, platformID int) err
|
||||
// func callbackUserOnline(operationID, userID string, platformID int, token string, isAppBackground bool, connID
|
||||
// string) cbApi.CommonCallbackResp {
|
||||
// callbackResp := cbApi.CommonCallbackResp{OperationID: operationID}
|
||||
// if !config.Config.Callback.CallbackUserOnline.WithEnable {
|
||||
// if !config.Config.Callback.CallbackUserOnline.Enable {
|
||||
// return callbackResp
|
||||
// }
|
||||
// callbackUserOnlineReq := cbApi.CallbackUserOnlineReq{
|
||||
@@ -134,7 +134,7 @@ func CallbackUserKickOff(ctx context.Context, userID string, platformID int) err
|
||||
//}
|
||||
//func callbackUserOffline(operationID, userID string, platformID int, connID string) cbApi.CommonCallbackResp {
|
||||
// callbackResp := cbApi.CommonCallbackResp{OperationID: operationID}
|
||||
// if !config.Config.Callback.CallbackUserOffline.WithEnable {
|
||||
// if !config.Config.Callback.CallbackUserOffline.Enable {
|
||||
// return callbackResp
|
||||
// }
|
||||
// callbackOfflineReq := cbApi.CallbackUserOfflineReq{
|
||||
@@ -161,7 +161,7 @@ func CallbackUserKickOff(ctx context.Context, userID string, platformID int) err
|
||||
//}
|
||||
//func callbackUserKickOff(operationID string, userID string, platformID int) cbApi.CommonCallbackResp {
|
||||
// callbackResp := cbApi.CommonCallbackResp{OperationID: operationID}
|
||||
// if !config.Config.Callback.CallbackUserKickOff.WithEnable {
|
||||
// if !config.Config.Callback.CallbackUserKickOff.Enable {
|
||||
// return callbackResp
|
||||
// }
|
||||
// callbackUserKickOffReq := cbApi.CallbackUserKickOffReq{
|
||||
|
||||
@@ -25,6 +25,8 @@ import (
|
||||
"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"
|
||||
@@ -57,7 +59,7 @@ type Server struct {
|
||||
rpcPort int
|
||||
prometheusPort int
|
||||
LongConnServer LongConnServer
|
||||
pushTerminal map[int]struct{}
|
||||
pushTerminal []int
|
||||
}
|
||||
|
||||
func (s *Server) SetLongConnServer(LongConnServer LongConnServer) {
|
||||
@@ -65,15 +67,12 @@ func (s *Server) SetLongConnServer(LongConnServer LongConnServer) {
|
||||
}
|
||||
|
||||
func NewServer(rpcPort int, proPort int, longConnServer LongConnServer) *Server {
|
||||
s := &Server{
|
||||
return &Server{
|
||||
rpcPort: rpcPort,
|
||||
prometheusPort: proPort,
|
||||
LongConnServer: longConnServer,
|
||||
pushTerminal: make(map[int]struct{}),
|
||||
pushTerminal: []int{constant.IOSPlatformID, constant.AndroidPlatformID},
|
||||
}
|
||||
s.pushTerminal[constant.IOSPlatformID] = struct{}{}
|
||||
s.pushTerminal[constant.AndroidPlatformID] = struct{}{}
|
||||
return s
|
||||
}
|
||||
|
||||
func (s *Server) OnlinePushMsg(
|
||||
@@ -127,9 +126,13 @@ func (s *Server) OnlineBatchPushOneMsg(
|
||||
panic("implement me")
|
||||
}
|
||||
|
||||
func (s *Server) SuperGroupOnlineBatchPushOneMsg(ctx context.Context, req *msggateway.OnlineBatchPushOneMsgReq,
|
||||
func (s *Server) SuperGroupOnlineBatchPushOneMsg(
|
||||
ctx context.Context,
|
||||
req *msggateway.OnlineBatchPushOneMsgReq,
|
||||
) (*msggateway.OnlineBatchPushOneMsgResp, error) {
|
||||
|
||||
var singleUserResults []*msggateway.SingleMsgToUserResults
|
||||
|
||||
for _, v := range req.PushToUserIDs {
|
||||
var resp []*msggateway.SingleMsgToUserPlatform
|
||||
results := &msggateway.SingleMsgToUserResults{
|
||||
@@ -150,22 +153,23 @@ func (s *Server) SuperGroupOnlineBatchPushOneMsg(ctx context.Context, req *msgga
|
||||
}
|
||||
|
||||
userPlatform := &msggateway.SingleMsgToUserPlatform{
|
||||
PlatFormID: int32(client.PlatformID),
|
||||
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 = int64(errs.ErrPushMsgErr.Code())
|
||||
userPlatform.ResultCode = -2
|
||||
resp = append(resp, userPlatform)
|
||||
} else {
|
||||
if _, ok := s.pushTerminal[client.PlatformID]; ok {
|
||||
if utils.IsContainInt(client.PlatformID, s.pushTerminal) {
|
||||
results.OnlinePush = true
|
||||
resp = append(resp, userPlatform)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
userPlatform.ResultCode = int64(errs.ErrIOSBackgroundPushErr.Code())
|
||||
userPlatform.ResultCode = -3
|
||||
resp = append(resp, userPlatform)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -19,6 +19,7 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"github.com/OpenIMSDK/tools/apiresp"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/signal"
|
||||
@@ -28,8 +29,6 @@ import (
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/OpenIMSDK/tools/apiresp"
|
||||
|
||||
"github.com/go-playground/validator/v10"
|
||||
"github.com/redis/go-redis/v9"
|
||||
"golang.org/x/sync/errgroup"
|
||||
|
||||
@@ -66,7 +66,7 @@ func StartTransfer(prometheusPort int) error {
|
||||
if err := client.CreateRpcRootNodes(config.Config.GetServiceNames()); err != nil {
|
||||
return err
|
||||
}
|
||||
client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")))
|
||||
client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()))
|
||||
msgModel := cache.NewMsgCacheModel(rdb)
|
||||
msgDocModel := unrelation.NewMsgMongoDriver(mongo.GetDatabase())
|
||||
msgDatabase := controller.NewCommonMsgDatabase(msgDocModel, msgModel)
|
||||
|
||||
@@ -16,7 +16,6 @@ package push
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
|
||||
"github.com/OpenIMSDK/protocol/constant"
|
||||
"github.com/OpenIMSDK/protocol/sdkws"
|
||||
@@ -137,14 +136,3 @@ func callbackBeforeSuperGroupOnlinePush(
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func GetContent(msg *sdkws.MsgData) string {
|
||||
if msg.ContentType >= constant.NotificationBegin && msg.ContentType <= constant.NotificationEnd {
|
||||
var notification sdkws.NotificationElem
|
||||
if err := json.Unmarshal(msg.Content, ¬ification); err != nil {
|
||||
return ""
|
||||
}
|
||||
return notification.Detail
|
||||
} else {
|
||||
return string(msg.Content)
|
||||
}
|
||||
}
|
||||
|
||||
32
internal/push/consumer_init.go
Normal file
32
internal/push/consumer_init.go
Normal file
@@ -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 push
|
||||
|
||||
type Consumer struct {
|
||||
pushCh ConsumerHandler
|
||||
successCount uint64
|
||||
}
|
||||
|
||||
func NewConsumer(pusher *Pusher) *Consumer {
|
||||
return &Consumer{
|
||||
pushCh: *NewConsumerHandler(pusher),
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Consumer) Start() {
|
||||
// statistics.NewStatistics(&c.successCount, config.Config.ModuleName.PushName, fmt.Sprintf("%d second push to
|
||||
// msg_gateway count", constant.StatisticsTimeInterval), constant.StatisticsTimeInterval)
|
||||
go c.pushCh.pushConsumerGroup.RegisterHandleAndConsumer(&c.pushCh)
|
||||
}
|
||||
@@ -16,16 +16,17 @@ package dummy
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush"
|
||||
)
|
||||
|
||||
func NewDummy() *Dummy {
|
||||
func NewClient() *Dummy {
|
||||
return &Dummy{}
|
||||
}
|
||||
|
||||
type Dummy struct {
|
||||
}
|
||||
|
||||
func (d *Dummy) Push(ctx context.Context, userIDs []string, title, content string, opts *options.Opts) error {
|
||||
func (d *Dummy) Push(ctx context.Context, userIDs []string, title, content string, opts *offlinepush.Opts) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -16,7 +16,6 @@ package fcm
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
|
||||
"path/filepath"
|
||||
|
||||
firebase "firebase.google.com/go"
|
||||
@@ -26,6 +25,7 @@ import (
|
||||
|
||||
"github.com/OpenIMSDK/protocol/constant"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
|
||||
)
|
||||
@@ -39,7 +39,7 @@ type Fcm struct {
|
||||
cache cache.MsgModel
|
||||
}
|
||||
|
||||
func NewFcm(cache cache.MsgModel) *Fcm {
|
||||
func NewClient(cache cache.MsgModel) *Fcm {
|
||||
projectRoot := config.GetProjectRoot()
|
||||
credentialsFilePath := filepath.Join(projectRoot, "config", config.Config.Push.Fcm.ServiceAccount)
|
||||
opt := option.WithCredentialsFile(credentialsFilePath)
|
||||
@@ -56,7 +56,7 @@ func NewFcm(cache cache.MsgModel) *Fcm {
|
||||
return &Fcm{fcmMsgCli: fcmMsgClient, cache: cache}
|
||||
}
|
||||
|
||||
func (f *Fcm) Push(ctx context.Context, userIDs []string, title, content string, opts *options.Opts) error {
|
||||
func (f *Fcm) Push(ctx context.Context, userIDs []string, title, content string, opts *offlinepush.Opts) error {
|
||||
// accounts->registrationToken
|
||||
allTokens := make(map[string][]string, 0)
|
||||
for _, account := range userIDs {
|
||||
|
||||
@@ -19,7 +19,6 @@ import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -31,6 +30,7 @@ import (
|
||||
"github.com/OpenIMSDK/tools/mcontext"
|
||||
"github.com/OpenIMSDK/tools/utils/splitter"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
|
||||
http2 "github.com/openimsdk/open-im-server/v3/pkg/common/http"
|
||||
@@ -55,17 +55,17 @@ const (
|
||||
taskIDTTL = 1000 * 60 * 60 * 24
|
||||
)
|
||||
|
||||
type GeTui struct {
|
||||
type Client struct {
|
||||
cache cache.MsgModel
|
||||
tokenExpireTime int64
|
||||
taskIDTTL int64
|
||||
}
|
||||
|
||||
func NewGeTui(cache cache.MsgModel) *GeTui {
|
||||
return &GeTui{cache: cache, tokenExpireTime: tokenExpireTime, taskIDTTL: taskIDTTL}
|
||||
func NewClient(cache cache.MsgModel) *Client {
|
||||
return &Client{cache: cache, tokenExpireTime: tokenExpireTime, taskIDTTL: taskIDTTL}
|
||||
}
|
||||
|
||||
func (g *GeTui) Push(ctx context.Context, userIDs []string, title, content string, opts *options.Opts) error {
|
||||
func (g *Client) Push(ctx context.Context, userIDs []string, title, content string, opts *offlinepush.Opts) error {
|
||||
token, err := g.cache.GetGetuiToken(ctx)
|
||||
if err != nil {
|
||||
if errs.Unwrap(err) == redis.Nil {
|
||||
@@ -111,7 +111,7 @@ func (g *GeTui) Push(ctx context.Context, userIDs []string, title, content strin
|
||||
return err
|
||||
}
|
||||
|
||||
func (g *GeTui) Auth(ctx context.Context, timeStamp int64) (token string, expireTime int64, err error) {
|
||||
func (g *Client) Auth(ctx context.Context, timeStamp int64) (token string, expireTime int64, err error) {
|
||||
h := sha256.New()
|
||||
h.Write(
|
||||
[]byte(config.Config.Push.GeTui.AppKey + strconv.Itoa(int(timeStamp)) + config.Config.Push.GeTui.MasterSecret),
|
||||
@@ -131,7 +131,7 @@ func (g *GeTui) Auth(ctx context.Context, timeStamp int64) (token string, expire
|
||||
return respAuth.Token, int64(expire), err
|
||||
}
|
||||
|
||||
func (g *GeTui) GetTaskID(ctx context.Context, token string, pushReq PushReq) (string, error) {
|
||||
func (g *Client) GetTaskID(ctx context.Context, token string, pushReq PushReq) (string, error) {
|
||||
respTask := TaskResp{}
|
||||
ttl := int64(1000 * 60 * 5)
|
||||
pushReq.Settings = &Settings{TTL: &ttl}
|
||||
@@ -143,7 +143,7 @@ func (g *GeTui) GetTaskID(ctx context.Context, token string, pushReq PushReq) (s
|
||||
}
|
||||
|
||||
// max num is 999.
|
||||
func (g *GeTui) batchPush(ctx context.Context, token string, userIDs []string, pushReq PushReq) error {
|
||||
func (g *Client) batchPush(ctx context.Context, token string, userIDs []string, pushReq PushReq) error {
|
||||
taskID, err := g.GetTaskID(ctx, token, pushReq)
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -152,21 +152,21 @@ func (g *GeTui) batchPush(ctx context.Context, token string, userIDs []string, p
|
||||
return g.request(ctx, batchPushURL, pushReq, token, nil)
|
||||
}
|
||||
|
||||
func (g *GeTui) singlePush(ctx context.Context, token, userID string, pushReq PushReq) error {
|
||||
func (g *Client) singlePush(ctx context.Context, token, userID string, pushReq PushReq) error {
|
||||
operationID := mcontext.GetOperationID(ctx)
|
||||
pushReq.RequestID = &operationID
|
||||
pushReq.Audience = &Audience{Alias: []string{userID}}
|
||||
return g.request(ctx, pushURL, pushReq, token, nil)
|
||||
}
|
||||
|
||||
func (g *GeTui) request(ctx context.Context, url string, input any, token string, output any) error {
|
||||
func (g *Client) request(ctx context.Context, url string, input any, token string, output any) error {
|
||||
header := map[string]string{"token": token}
|
||||
resp := &Resp{}
|
||||
resp.Data = output
|
||||
return g.postReturn(ctx, config.Config.Push.GeTui.PushUrl+url, header, input, resp, 3)
|
||||
}
|
||||
|
||||
func (g *GeTui) postReturn(
|
||||
func (g *Client) postReturn(
|
||||
ctx context.Context,
|
||||
url string,
|
||||
header map[string]string,
|
||||
@@ -181,7 +181,7 @@ func (g *GeTui) postReturn(
|
||||
return output.parseError()
|
||||
}
|
||||
|
||||
func (g *GeTui) getTokenAndSave2Redis(ctx context.Context) (token string, err error) {
|
||||
func (g *Client) getTokenAndSave2Redis(ctx context.Context) (token string, err error) {
|
||||
token, _, err = g.Auth(ctx, time.Now().UnixNano()/1e6)
|
||||
if err != nil {
|
||||
return
|
||||
@@ -193,7 +193,7 @@ func (g *GeTui) getTokenAndSave2Redis(ctx context.Context) (token string, err er
|
||||
return token, nil
|
||||
}
|
||||
|
||||
func (g *GeTui) GetTaskIDAndSave2Redis(ctx context.Context, token string, pushReq PushReq) (taskID string, err error) {
|
||||
func (g *Client) GetTaskIDAndSave2Redis(ctx context.Context, token string, pushReq PushReq) (taskID string, err error) {
|
||||
pushReq.Settings = &Settings{TTL: &g.taskIDTTL}
|
||||
taskID, err = g.GetTaskID(ctx, token, pushReq)
|
||||
if err != nil {
|
||||
|
||||
@@ -18,8 +18,8 @@ import (
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"fmt"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/jpush/body"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||
http2 "github.com/openimsdk/open-im-server/v3/pkg/common/http"
|
||||
@@ -27,7 +27,7 @@ import (
|
||||
|
||||
type JPush struct{}
|
||||
|
||||
func NewJPush() *JPush {
|
||||
func NewClient() *JPush {
|
||||
return &JPush{}
|
||||
}
|
||||
|
||||
@@ -46,7 +46,7 @@ func (j *JPush) getAuthorization(appKey string, masterSecret string) string {
|
||||
return Authorization
|
||||
}
|
||||
|
||||
func (j *JPush) Push(ctx context.Context, userIDs []string, title, content string, opts *options.Opts) error {
|
||||
func (j *JPush) Push(ctx context.Context, userIDs []string, title, content string, opts *offlinepush.Opts) error {
|
||||
var pf body.Platform
|
||||
pf.SetAll()
|
||||
var au body.Audience
|
||||
|
||||
37
internal/push/offlinepush/offlinepush_interface.go
Normal file
37
internal/push/offlinepush/offlinepush_interface.go
Normal file
@@ -0,0 +1,37 @@
|
||||
// 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 offlinepush
|
||||
|
||||
import (
|
||||
"context"
|
||||
)
|
||||
|
||||
// OfflinePusher Offline Pusher.
|
||||
type OfflinePusher interface {
|
||||
Push(ctx context.Context, userIDs []string, title, content string, opts *Opts) error
|
||||
}
|
||||
|
||||
// Opts opts.
|
||||
type Opts struct {
|
||||
Signal *Signal
|
||||
IOSPushSound string
|
||||
IOSBadgeCount bool
|
||||
Ex string
|
||||
}
|
||||
|
||||
// Signal message id.
|
||||
type Signal struct {
|
||||
ClientMsgID string
|
||||
}
|
||||
@@ -1,52 +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 offlinepush
|
||||
|
||||
import (
|
||||
"context"
|
||||
"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"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
|
||||
)
|
||||
|
||||
const (
|
||||
GETUI = "getui"
|
||||
FIREBASE = "fcm"
|
||||
JPUSH = "jpush"
|
||||
)
|
||||
|
||||
// OfflinePusher Offline Pusher.
|
||||
type OfflinePusher interface {
|
||||
Push(ctx context.Context, userIDs []string, title, content string, opts *options.Opts) error
|
||||
}
|
||||
|
||||
func NewOfflinePusher(cache cache.MsgModel) OfflinePusher {
|
||||
var offlinePusher OfflinePusher
|
||||
switch config.Config.Push.Enable {
|
||||
case GETUI:
|
||||
offlinePusher = getui.NewGeTui(cache)
|
||||
case FIREBASE:
|
||||
offlinePusher = fcm.NewFcm(cache)
|
||||
case JPUSH:
|
||||
offlinePusher = jpush.NewJPush()
|
||||
default:
|
||||
offlinePusher = dummy.NewDummy()
|
||||
}
|
||||
return offlinePusher
|
||||
}
|
||||
@@ -1,14 +0,0 @@
|
||||
package options
|
||||
|
||||
// Opts opts.
|
||||
type Opts struct {
|
||||
Signal *Signal
|
||||
IOSPushSound string
|
||||
IOSBadgeCount bool
|
||||
Ex string
|
||||
}
|
||||
|
||||
// Signal message id.
|
||||
type Signal struct {
|
||||
ClientMsgID string
|
||||
}
|
||||
@@ -1,211 +0,0 @@
|
||||
package push
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/OpenIMSDK/protocol/msggateway"
|
||||
"github.com/OpenIMSDK/protocol/sdkws"
|
||||
"github.com/OpenIMSDK/tools/discoveryregistry"
|
||||
"github.com/OpenIMSDK/tools/log"
|
||||
"github.com/OpenIMSDK/tools/utils"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||
"golang.org/x/sync/errgroup"
|
||||
"google.golang.org/grpc"
|
||||
"os"
|
||||
"sync"
|
||||
)
|
||||
|
||||
const (
|
||||
ENVNAME = "ENVS_DISCOVERY"
|
||||
KUBERNETES = "k8s"
|
||||
ZOOKEEPER = "zookeeper"
|
||||
)
|
||||
|
||||
type OnlinePusher interface {
|
||||
GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.MsgData,
|
||||
pushToUserIDs []string) (wsResults []*msggateway.SingleMsgToUserResults, err error)
|
||||
GetOnlinePushFailedUserIDs(ctx context.Context, msg *sdkws.MsgData, wsResults []*msggateway.SingleMsgToUserResults,
|
||||
pushToUserIDs *[]string) []string
|
||||
}
|
||||
|
||||
type emptyOnlinePUsher struct{}
|
||||
|
||||
func newEmptyOnlinePUsher() *emptyOnlinePUsher {
|
||||
return &emptyOnlinePUsher{}
|
||||
}
|
||||
|
||||
func (emptyOnlinePUsher) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.MsgData,
|
||||
pushToUserIDs []string) (wsResults []*msggateway.SingleMsgToUserResults, err error) {
|
||||
log.ZWarn(ctx, "emptyOnlinePUsher GetConnsAndOnlinePush", nil)
|
||||
return nil, nil
|
||||
}
|
||||
func (u emptyOnlinePUsher) GetOnlinePushFailedUserIDs(ctx context.Context, msg *sdkws.MsgData,
|
||||
wsResults []*msggateway.SingleMsgToUserResults, pushToUserIDs *[]string) []string {
|
||||
log.ZWarn(ctx, "emptyOnlinePUsher GetOnlinePushFailedUserIDs", nil)
|
||||
return nil
|
||||
}
|
||||
|
||||
func NewOnlinePusher(disCov discoveryregistry.SvcDiscoveryRegistry) OnlinePusher {
|
||||
var envType string
|
||||
if value := os.Getenv(ENVNAME); value != "" {
|
||||
envType = os.Getenv(ENVNAME)
|
||||
} else {
|
||||
envType = config.Config.Envs.Discovery
|
||||
}
|
||||
switch envType {
|
||||
case KUBERNETES:
|
||||
return NewK8sStaticConsistentHash(disCov)
|
||||
case ZOOKEEPER:
|
||||
return NewDefaultAllNode(disCov)
|
||||
default:
|
||||
return newEmptyOnlinePUsher()
|
||||
}
|
||||
}
|
||||
|
||||
type DefaultAllNode struct {
|
||||
disCov discoveryregistry.SvcDiscoveryRegistry
|
||||
}
|
||||
|
||||
func NewDefaultAllNode(disCov discoveryregistry.SvcDiscoveryRegistry) *DefaultAllNode {
|
||||
return &DefaultAllNode{disCov: disCov}
|
||||
}
|
||||
|
||||
func (d *DefaultAllNode) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.MsgData,
|
||||
pushToUserIDs []string) (wsResults []*msggateway.SingleMsgToUserResults, err error) {
|
||||
conns, err := d.disCov.GetConns(ctx, config.Config.RpcRegisterName.OpenImMessageGatewayName)
|
||||
log.ZDebug(ctx, "get gateway conn", "conn length", len(conns))
|
||||
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
|
||||
})
|
||||
}
|
||||
|
||||
_ = wg.Wait()
|
||||
|
||||
// always return nil
|
||||
return wsResults, nil
|
||||
}
|
||||
|
||||
func (d *DefaultAllNode) GetOnlinePushFailedUserIDs(_ context.Context, msg *sdkws.MsgData,
|
||||
wsResults []*msggateway.SingleMsgToUserResults, pushToUserIDs *[]string) []string {
|
||||
|
||||
onlineSuccessUserIDs := []string{msg.SendID}
|
||||
for _, v := range wsResults {
|
||||
//message sender do not need offline push
|
||||
if msg.SendID == v.UserID {
|
||||
continue
|
||||
}
|
||||
// mobile online push success
|
||||
if v.OnlinePush {
|
||||
onlineSuccessUserIDs = append(onlineSuccessUserIDs, v.UserID)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
return utils.SliceSub(*pushToUserIDs, onlineSuccessUserIDs)
|
||||
}
|
||||
|
||||
type K8sStaticConsistentHash struct {
|
||||
disCov discoveryregistry.SvcDiscoveryRegistry
|
||||
}
|
||||
|
||||
func NewK8sStaticConsistentHash(disCov discoveryregistry.SvcDiscoveryRegistry) *K8sStaticConsistentHash {
|
||||
return &K8sStaticConsistentHash{disCov: disCov}
|
||||
}
|
||||
|
||||
func (k *K8sStaticConsistentHash) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.MsgData,
|
||||
pushToUserIDs []string) (wsResults []*msggateway.SingleMsgToUserResults, err error) {
|
||||
|
||||
var usersHost = make(map[string][]string)
|
||||
for _, v := range pushToUserIDs {
|
||||
tHost, err := k.disCov.GetUserIdHashGatewayHost(ctx, v)
|
||||
if err != nil {
|
||||
log.ZError(ctx, "get msg gateway hash error", err)
|
||||
return nil, err
|
||||
}
|
||||
tUsers, tbl := usersHost[tHost]
|
||||
if tbl {
|
||||
tUsers = append(tUsers, v)
|
||||
usersHost[tHost] = tUsers
|
||||
} else {
|
||||
usersHost[tHost] = []string{v}
|
||||
}
|
||||
}
|
||||
log.ZDebug(ctx, "genUsers send hosts struct:", "usersHost", usersHost)
|
||||
var usersConns = make(map[*grpc.ClientConn][]string)
|
||||
for host, userIds := range usersHost {
|
||||
tconn, _ := k.disCov.GetConn(ctx, host)
|
||||
usersConns[tconn] = userIds
|
||||
}
|
||||
var (
|
||||
mu sync.Mutex
|
||||
wg = errgroup.Group{}
|
||||
maxWorkers = config.Config.Push.MaxConcurrentWorkers
|
||||
)
|
||||
if maxWorkers < 3 {
|
||||
maxWorkers = 3
|
||||
}
|
||||
wg.SetLimit(maxWorkers)
|
||||
for conn, userIds := range usersConns {
|
||||
tcon := conn
|
||||
tuserIds := userIds
|
||||
wg.Go(func() error {
|
||||
input := &msggateway.OnlineBatchPushOneMsgReq{MsgData: msg, PushToUserIDs: tuserIds}
|
||||
msgClient := msggateway.NewMsgGatewayClient(tcon)
|
||||
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
|
||||
})
|
||||
}
|
||||
_ = wg.Wait()
|
||||
return wsResults, nil
|
||||
}
|
||||
func (k *K8sStaticConsistentHash) GetOnlinePushFailedUserIDs(_ context.Context, _ *sdkws.MsgData,
|
||||
wsResults []*msggateway.SingleMsgToUserResults, _ *[]string) []string {
|
||||
var needOfflinePushUserIDs []string
|
||||
for _, v := range wsResults {
|
||||
if !v.OnlinePush {
|
||||
needOfflinePushUserIDs = append(needOfflinePushUserIDs, v.UserID)
|
||||
}
|
||||
}
|
||||
return needOfflinePushUserIDs
|
||||
}
|
||||
@@ -1,51 +0,0 @@
|
||||
package push
|
||||
|
||||
import (
|
||||
"context"
|
||||
pbpush "github.com/OpenIMSDK/protocol/push"
|
||||
"github.com/OpenIMSDK/tools/discoveryregistry"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/controller"
|
||||
"google.golang.org/grpc"
|
||||
)
|
||||
|
||||
type pushServer struct {
|
||||
database controller.PushDatabase
|
||||
disCov discoveryregistry.SvcDiscoveryRegistry
|
||||
offlinePusher offlinepush.OfflinePusher
|
||||
pushCh *ConsumerHandler
|
||||
}
|
||||
|
||||
func (p pushServer) PushMsg(ctx context.Context, req *pbpush.PushMsgReq) (*pbpush.PushMsgResp, error) {
|
||||
//todo reserved Interface
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (p pushServer) DelUserPushToken(ctx context.Context,
|
||||
req *pbpush.DelUserPushTokenReq) (resp *pbpush.DelUserPushTokenResp, err error) {
|
||||
if err = p.database.DelFcmToken(ctx, req.UserID, int(req.PlatformID)); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &pbpush.DelUserPushTokenResp{}, nil
|
||||
}
|
||||
|
||||
func Start(disCov discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) error {
|
||||
rdb, err := cache.NewRedis()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
cacheModel := cache.NewMsgCacheModel(rdb)
|
||||
offlinePusher := offlinepush.NewOfflinePusher(cacheModel)
|
||||
database := controller.NewPushDatabase(cacheModel)
|
||||
|
||||
consumer := NewConsumerHandler(offlinePusher, rdb, disCov)
|
||||
pbpush.RegisterPushMsgServiceServer(server, &pushServer{
|
||||
database: database,
|
||||
disCov: disCov,
|
||||
offlinePusher: offlinePusher,
|
||||
pushCh: consumer,
|
||||
})
|
||||
go consumer.pushConsumerGroup.RegisterHandleAndConsumer(consumer)
|
||||
return nil
|
||||
}
|
||||
@@ -16,17 +16,6 @@ package push
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"github.com/OpenIMSDK/protocol/sdkws"
|
||||
"github.com/OpenIMSDK/tools/discoveryregistry"
|
||||
"github.com/OpenIMSDK/tools/mcontext"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
|
||||
"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/rpccache"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
|
||||
"github.com/redis/go-redis/v9"
|
||||
|
||||
"github.com/IBM/sarama"
|
||||
"google.golang.org/protobuf/proto"
|
||||
@@ -42,31 +31,18 @@ import (
|
||||
)
|
||||
|
||||
type ConsumerHandler struct {
|
||||
pushConsumerGroup *kfk.MConsumerGroup
|
||||
offlinePusher offlinepush.OfflinePusher
|
||||
onlinePusher OnlinePusher
|
||||
groupLocalCache *rpccache.GroupLocalCache
|
||||
conversationLocalCache *rpccache.ConversationLocalCache
|
||||
msgRpcClient rpcclient.MessageRpcClient
|
||||
conversationRpcClient rpcclient.ConversationRpcClient
|
||||
groupRpcClient rpcclient.GroupRpcClient
|
||||
pushConsumerGroup *kfk.MConsumerGroup
|
||||
pusher *Pusher
|
||||
}
|
||||
|
||||
func NewConsumerHandler(offlinePusher offlinepush.OfflinePusher,
|
||||
rdb redis.UniversalClient, disCov discoveryregistry.SvcDiscoveryRegistry) *ConsumerHandler {
|
||||
func NewConsumerHandler(pusher *Pusher) *ConsumerHandler {
|
||||
var consumerHandler ConsumerHandler
|
||||
consumerHandler.pusher = pusher
|
||||
consumerHandler.pushConsumerGroup = kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{
|
||||
KafkaVersion: sarama.V2_0_0_0,
|
||||
OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false,
|
||||
}, []string{config.Config.Kafka.MsgToPush.Topic}, config.Config.Kafka.Addr,
|
||||
config.Config.Kafka.ConsumerGroupID.MsgToPush)
|
||||
consumerHandler.offlinePusher = offlinePusher
|
||||
consumerHandler.onlinePusher = NewOnlinePusher(disCov)
|
||||
consumerHandler.groupRpcClient = rpcclient.NewGroupRpcClient(disCov)
|
||||
consumerHandler.groupLocalCache = rpccache.NewGroupLocalCache(consumerHandler.groupRpcClient, rdb)
|
||||
consumerHandler.msgRpcClient = rpcclient.NewMessageRpcClient(disCov)
|
||||
consumerHandler.conversationRpcClient = rpcclient.NewConversationRpcClient(disCov)
|
||||
consumerHandler.conversationLocalCache = rpccache.NewConversationLocalCache(consumerHandler.conversationRpcClient, rdb)
|
||||
return &consumerHandler
|
||||
}
|
||||
|
||||
@@ -89,7 +65,7 @@ func (c *ConsumerHandler) handleMs2PsChat(ctx context.Context, msg []byte) {
|
||||
var err error
|
||||
switch msgFromMQ.MsgData.SessionType {
|
||||
case constant.SuperGroupChatType:
|
||||
err = c.Push2SuperGroup(ctx, pbData.MsgData.GroupID, pbData.MsgData)
|
||||
err = c.pusher.Push2SuperGroup(ctx, pbData.MsgData.GroupID, pbData.MsgData)
|
||||
default:
|
||||
var pushUserIDList []string
|
||||
isSenderSync := utils.GetSwitchFromOptions(pbData.MsgData.Options, constant.IsSenderSync)
|
||||
@@ -98,14 +74,18 @@ func (c *ConsumerHandler) handleMs2PsChat(ctx context.Context, msg []byte) {
|
||||
} else {
|
||||
pushUserIDList = append(pushUserIDList, pbData.MsgData.RecvID, pbData.MsgData.SendID)
|
||||
}
|
||||
err = c.Push2User(ctx, pushUserIDList, pbData.MsgData)
|
||||
err = c.pusher.Push2User(ctx, pushUserIDList, pbData.MsgData)
|
||||
}
|
||||
if err != nil {
|
||||
log.ZError(ctx, "push failed", err, "msg", pbData.String())
|
||||
if err == errNoOfflinePusher {
|
||||
log.ZWarn(ctx, "offline push failed", err, "msg", pbData.String())
|
||||
} else {
|
||||
log.ZError(ctx, "push failed", err, "msg", pbData.String())
|
||||
}
|
||||
}
|
||||
}
|
||||
func (*ConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }
|
||||
func (*ConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
|
||||
func (ConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }
|
||||
func (ConsumerHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
|
||||
func (c *ConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
|
||||
claim sarama.ConsumerGroupClaim,
|
||||
) error {
|
||||
@@ -116,243 +96,3 @@ func (c *ConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession,
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Push2User Suitable for two types of conversations, one is SingleChatType and the other is NotificationChatType.
|
||||
func (c *ConsumerHandler) Push2User(ctx context.Context, userIDs []string, msg *sdkws.MsgData) error {
|
||||
log.ZDebug(ctx, "Get msg from msg_transfer And push msg", "userIDs", userIDs, "msg", msg.String())
|
||||
if err := callbackOnlinePush(ctx, userIDs, msg); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
wsResults, err := c.onlinePusher.GetConnsAndOnlinePush(ctx, msg, userIDs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
log.ZDebug(ctx, "single and notification push result", "result", wsResults, "msg", msg, "push_to_userID", userIDs)
|
||||
|
||||
if !c.shouldPushOffline(ctx, msg) {
|
||||
return nil
|
||||
}
|
||||
|
||||
for _, v := range wsResults {
|
||||
//message sender do not need offline push
|
||||
if msg.SendID == v.UserID {
|
||||
continue
|
||||
}
|
||||
//receiver online push success
|
||||
if v.OnlinePush {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
offlinePUshUserID := []string{msg.RecvID}
|
||||
//receiver offline push
|
||||
if err = callbackOfflinePush(ctx, offlinePUshUserID, msg, nil); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
err = c.offlinePushMsg(ctx, msg, offlinePUshUserID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *ConsumerHandler) 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 {
|
||||
return err
|
||||
}
|
||||
|
||||
err = c.groupMessagesHandler(ctx, groupID, &pushToUserIDs, msg)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
wsResults, err := c.onlinePusher.GetConnsAndOnlinePush(ctx, msg, pushToUserIDs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
log.ZDebug(ctx, "group push result", "result", wsResults, "msg", msg)
|
||||
|
||||
if !c.shouldPushOffline(ctx, msg) {
|
||||
return nil
|
||||
}
|
||||
needOfflinePushUserIDs := c.onlinePusher.GetOnlinePushFailedUserIDs(ctx, msg, wsResults, &pushToUserIDs)
|
||||
|
||||
//filter some user, like don not disturb or don't need offline push etc.
|
||||
needOfflinePushUserIDs, err = c.filterGroupMessageOfflinePush(ctx, groupID, msg, needOfflinePushUserIDs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// Use offline push messaging
|
||||
if len(needOfflinePushUserIDs) > 0 {
|
||||
var offlinePushUserIDs []string
|
||||
err = callbackOfflinePush(ctx, needOfflinePushUserIDs, msg, &offlinePushUserIDs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if len(offlinePushUserIDs) > 0 {
|
||||
needOfflinePushUserIDs = offlinePushUserIDs
|
||||
}
|
||||
|
||||
err = c.offlinePushMsg(ctx, msg, needOfflinePushUserIDs)
|
||||
if err != nil {
|
||||
log.ZError(ctx, "offlinePushMsg failed", err, "groupID", groupID, "msg", msg)
|
||||
return err
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *ConsumerHandler) offlinePushMsg(ctx context.Context, msg *sdkws.MsgData, offlinePushUserIDs []string) error {
|
||||
title, content, opts, err := c.getOfflinePushInfos(msg)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = c.offlinePusher.Push(ctx, offlinePushUserIDs, title, content, opts)
|
||||
if err != nil {
|
||||
prommetrics.MsgOfflinePushFailedCounter.Inc()
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *ConsumerHandler) filterGroupMessageOfflinePush(ctx context.Context, groupID string, msg *sdkws.MsgData,
|
||||
offlinePushUserIDs []string) (userIDs []string, err error) {
|
||||
|
||||
//todo local cache Obtain the difference set through local comparison.
|
||||
needOfflinePushUserIDs, err := c.conversationRpcClient.GetConversationOfflinePushUserIDs(
|
||||
ctx, utils.GenGroupConversationID(groupID), offlinePushUserIDs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return needOfflinePushUserIDs, nil
|
||||
}
|
||||
|
||||
func (c *ConsumerHandler) getOfflinePushInfos(msg *sdkws.MsgData) (title, content string, opts *options.Opts, err error) {
|
||||
type AtTextElem struct {
|
||||
Text string `json:"text,omitempty"`
|
||||
AtUserList []string `json:"atUserList,omitempty"`
|
||||
IsAtSelf bool `json:"isAtSelf"`
|
||||
}
|
||||
|
||||
opts = &options.Opts{Signal: &options.Signal{}}
|
||||
if msg.OfflinePushInfo != nil {
|
||||
opts.IOSBadgeCount = msg.OfflinePushInfo.IOSBadgeCount
|
||||
opts.IOSPushSound = msg.OfflinePushInfo.IOSPushSound
|
||||
opts.Ex = msg.OfflinePushInfo.Ex
|
||||
}
|
||||
|
||||
if msg.OfflinePushInfo != nil {
|
||||
title = msg.OfflinePushInfo.Title
|
||||
content = msg.OfflinePushInfo.Desc
|
||||
}
|
||||
if title == "" {
|
||||
switch msg.ContentType {
|
||||
case constant.Text:
|
||||
fallthrough
|
||||
case constant.Picture:
|
||||
fallthrough
|
||||
case constant.Voice:
|
||||
fallthrough
|
||||
case constant.Video:
|
||||
fallthrough
|
||||
case constant.File:
|
||||
title = constant.ContentType2PushContent[int64(msg.ContentType)]
|
||||
case constant.AtText:
|
||||
ac := AtTextElem{}
|
||||
_ = utils.JsonStringToStruct(string(msg.Content), &ac)
|
||||
case constant.SignalingNotification:
|
||||
title = constant.ContentType2PushContent[constant.SignalMsg]
|
||||
default:
|
||||
title = constant.ContentType2PushContent[constant.Common]
|
||||
}
|
||||
}
|
||||
if content == "" {
|
||||
content = title
|
||||
}
|
||||
return
|
||||
}
|
||||
func (c *ConsumerHandler) groupMessagesHandler(ctx context.Context, groupID string, pushToUserIDs *[]string, msg *sdkws.MsgData) (err error) {
|
||||
if len(*pushToUserIDs) == 0 {
|
||||
*pushToUserIDs, err = c.groupLocalCache.GetGroupMemberIDs(ctx, groupID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
switch msg.ContentType {
|
||||
case constant.MemberQuitNotification:
|
||||
var tips sdkws.MemberQuitTips
|
||||
if unmarshalNotificationElem(msg.Content, &tips) != nil {
|
||||
return err
|
||||
}
|
||||
if err = c.DeleteMemberAndSetConversationSeq(ctx, groupID, []string{tips.QuitUser.UserID}); err != nil {
|
||||
log.ZError(ctx, "MemberQuitNotification DeleteMemberAndSetConversationSeq", err, "groupID", groupID, "userID", tips.QuitUser.UserID)
|
||||
}
|
||||
*pushToUserIDs = append(*pushToUserIDs, tips.QuitUser.UserID)
|
||||
case constant.MemberKickedNotification:
|
||||
var tips sdkws.MemberKickedTips
|
||||
if unmarshalNotificationElem(msg.Content, &tips) != nil {
|
||||
return err
|
||||
}
|
||||
kickedUsers := utils.Slice(tips.KickedUserList, func(e *sdkws.GroupMemberFullInfo) string { return e.UserID })
|
||||
if err = c.DeleteMemberAndSetConversationSeq(ctx, groupID, kickedUsers); err != nil {
|
||||
log.ZError(ctx, "MemberKickedNotification DeleteMemberAndSetConversationSeq", err, "groupID", groupID, "userIDs", kickedUsers)
|
||||
}
|
||||
|
||||
*pushToUserIDs = append(*pushToUserIDs, kickedUsers...)
|
||||
case constant.GroupDismissedNotification:
|
||||
if msgprocessor.IsNotification(msgprocessor.GetConversationIDByMsg(msg)) { // 消息先到,通知后到
|
||||
var tips sdkws.GroupDismissedTips
|
||||
if unmarshalNotificationElem(msg.Content, &tips) != nil {
|
||||
return err
|
||||
}
|
||||
log.ZInfo(ctx, "GroupDismissedNotificationInfo****", "groupID", groupID, "num", len(*pushToUserIDs), "list", pushToUserIDs)
|
||||
if len(config.Config.Manager.UserID) > 0 {
|
||||
ctx = mcontext.WithOpUserIDContext(ctx, config.Config.Manager.UserID[0])
|
||||
}
|
||||
defer func(groupID string) {
|
||||
if err = c.groupRpcClient.DismissGroup(ctx, groupID); err != nil {
|
||||
log.ZError(ctx, "DismissGroup Notification clear members", err, "groupID", groupID)
|
||||
}
|
||||
}(groupID)
|
||||
}
|
||||
}
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func (c *ConsumerHandler) DeleteMemberAndSetConversationSeq(ctx context.Context, groupID string, userIDs []string) error {
|
||||
conversationID := msgprocessor.GetConversationIDBySessionType(constant.SuperGroupChatType, groupID)
|
||||
maxSeq, err := c.msgRpcClient.GetConversationMaxSeq(ctx, conversationID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return c.conversationRpcClient.SetConversationMaxSeq(ctx, userIDs, conversationID, maxSeq)
|
||||
}
|
||||
|
||||
func unmarshalNotificationElem(bytes []byte, t any) error {
|
||||
var notification sdkws.NotificationElem
|
||||
if err := json.Unmarshal(bytes, ¬ification); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return json.Unmarshal([]byte(notification.Detail), t)
|
||||
}
|
||||
|
||||
func (c *ConsumerHandler) shouldPushOffline(_ context.Context, msg *sdkws.MsgData) bool {
|
||||
isOfflinePush := utils.GetSwitchFromOptions(msg.Options, constant.IsOfflinePush)
|
||||
if !isOfflinePush {
|
||||
return false
|
||||
}
|
||||
if msg.ContentType == constant.SignalingNotification {
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
109
internal/push/push_rpc_server.go
Normal file
109
internal/push/push_rpc_server.go
Normal file
@@ -0,0 +1,109 @@
|
||||
// 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 push
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/OpenIMSDK/tools/utils"
|
||||
"sync"
|
||||
|
||||
"google.golang.org/grpc"
|
||||
|
||||
"github.com/OpenIMSDK/protocol/constant"
|
||||
pbpush "github.com/OpenIMSDK/protocol/push"
|
||||
"github.com/OpenIMSDK/tools/discoveryregistry"
|
||||
"github.com/OpenIMSDK/tools/log"
|
||||
|
||||
"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/rpcclient"
|
||||
)
|
||||
|
||||
type pushServer struct {
|
||||
pusher *Pusher
|
||||
}
|
||||
|
||||
func Start(client discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) error {
|
||||
rdb, err := cache.NewRedis()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
cacheModel := cache.NewMsgCacheModel(rdb)
|
||||
offlinePusher := NewOfflinePusher(cacheModel)
|
||||
database := controller.NewPushDatabase(cacheModel)
|
||||
groupRpcClient := rpcclient.NewGroupRpcClient(client)
|
||||
conversationRpcClient := rpcclient.NewConversationRpcClient(client)
|
||||
msgRpcClient := rpcclient.NewMessageRpcClient(client)
|
||||
pusher := NewPusher(
|
||||
client,
|
||||
offlinePusher,
|
||||
database,
|
||||
localcache.NewGroupLocalCache(&groupRpcClient),
|
||||
localcache.NewConversationLocalCache(&conversationRpcClient),
|
||||
&conversationRpcClient,
|
||||
&groupRpcClient,
|
||||
&msgRpcClient,
|
||||
)
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(2)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
pbpush.RegisterPushMsgServiceServer(server, &pushServer{
|
||||
pusher: pusher,
|
||||
})
|
||||
}()
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
consumer := NewConsumer(pusher)
|
||||
consumer.Start()
|
||||
}()
|
||||
wg.Wait()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *pushServer) PushMsg(ctx context.Context, pbData *pbpush.PushMsgReq) (resp *pbpush.PushMsgResp, err error) {
|
||||
switch pbData.MsgData.SessionType {
|
||||
case constant.SuperGroupChatType:
|
||||
err = r.pusher.Push2SuperGroup(ctx, pbData.MsgData.GroupID, pbData.MsgData)
|
||||
default:
|
||||
var pushUserIDList []string
|
||||
isSenderSync := utils.GetSwitchFromOptions(pbData.MsgData.Options, constant.IsSenderSync)
|
||||
if !isSenderSync {
|
||||
pushUserIDList = append(pushUserIDList, pbData.MsgData.RecvID)
|
||||
} else {
|
||||
pushUserIDList = append(pushUserIDList, pbData.MsgData.RecvID, pbData.MsgData.SendID)
|
||||
}
|
||||
err = r.pusher.Push2User(ctx, pushUserIDList, pbData.MsgData)
|
||||
}
|
||||
if err != nil {
|
||||
if err != errNoOfflinePusher {
|
||||
return nil, err
|
||||
} else {
|
||||
log.ZWarn(ctx, "offline push failed", err, "msg", pbData.String())
|
||||
}
|
||||
}
|
||||
return &pbpush.PushMsgResp{}, nil
|
||||
}
|
||||
|
||||
func (r *pushServer) DelUserPushToken(
|
||||
ctx context.Context,
|
||||
req *pbpush.DelUserPushTokenReq,
|
||||
) (resp *pbpush.DelUserPushTokenResp, err error) {
|
||||
if err = r.pusher.database.DelFcmToken(ctx, req.UserID, int(req.PlatformID)); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &pbpush.DelUserPushTokenResp{}, nil
|
||||
}
|
||||
511
internal/push/push_to_client.go
Normal file
511
internal/push/push_to_client.go
Normal file
@@ -0,0 +1,511 @@
|
||||
// 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 push
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"google.golang.org/grpc"
|
||||
"sync"
|
||||
|
||||
"golang.org/x/sync/errgroup"
|
||||
|
||||
"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"
|
||||
"github.com/OpenIMSDK/tools/log"
|
||||
"github.com/OpenIMSDK/tools/mcontext"
|
||||
"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"
|
||||
"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/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"
|
||||
)
|
||||
|
||||
type Pusher struct {
|
||||
database controller.PushDatabase
|
||||
discov discoveryregistry.SvcDiscoveryRegistry
|
||||
offlinePusher offlinepush.OfflinePusher
|
||||
groupLocalCache *localcache.GroupLocalCache
|
||||
conversationLocalCache *localcache.ConversationLocalCache
|
||||
msgRpcClient *rpcclient.MessageRpcClient
|
||||
conversationRpcClient *rpcclient.ConversationRpcClient
|
||||
groupRpcClient *rpcclient.GroupRpcClient
|
||||
}
|
||||
|
||||
var errNoOfflinePusher = errors.New("no offlinePusher is configured")
|
||||
|
||||
func NewPusher(discov discoveryregistry.SvcDiscoveryRegistry, offlinePusher offlinepush.OfflinePusher, database controller.PushDatabase,
|
||||
groupLocalCache *localcache.GroupLocalCache, conversationLocalCache *localcache.ConversationLocalCache,
|
||||
conversationRpcClient *rpcclient.ConversationRpcClient, groupRpcClient *rpcclient.GroupRpcClient, msgRpcClient *rpcclient.MessageRpcClient,
|
||||
) *Pusher {
|
||||
return &Pusher{
|
||||
discov: discov,
|
||||
database: database,
|
||||
offlinePusher: offlinePusher,
|
||||
groupLocalCache: groupLocalCache,
|
||||
conversationLocalCache: conversationLocalCache,
|
||||
msgRpcClient: msgRpcClient,
|
||||
conversationRpcClient: conversationRpcClient,
|
||||
groupRpcClient: groupRpcClient,
|
||||
}
|
||||
}
|
||||
|
||||
func NewOfflinePusher(cache cache.MsgModel) offlinepush.OfflinePusher {
|
||||
var offlinePusher offlinepush.OfflinePusher
|
||||
switch config.Config.Push.Enable {
|
||||
case "getui":
|
||||
offlinePusher = getui.NewClient(cache)
|
||||
case "fcm":
|
||||
offlinePusher = fcm.NewClient(cache)
|
||||
case "jpush":
|
||||
offlinePusher = jpush.NewClient()
|
||||
default:
|
||||
offlinePusher = dummy.NewClient()
|
||||
}
|
||||
return offlinePusher
|
||||
}
|
||||
|
||||
func (p *Pusher) DeleteMemberAndSetConversationSeq(ctx context.Context, groupID string, userIDs []string) error {
|
||||
conevrsationID := msgprocessor.GetConversationIDBySessionType(constant.SuperGroupChatType, groupID)
|
||||
maxSeq, err := p.msgRpcClient.GetConversationMaxSeq(ctx, conevrsationID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return p.conversationRpcClient.SetConversationMaxSeq(ctx, userIDs, conevrsationID, maxSeq)
|
||||
}
|
||||
|
||||
func (p *Pusher) Push2User(ctx context.Context, userIDs []string, msg *sdkws.MsgData) error {
|
||||
log.ZDebug(ctx, "Get msg from msg_transfer And push msg", "userIDs", userIDs, "msg", msg.String())
|
||||
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 !v.OnlinePush && msg.SendID == v.UserID {
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *Pusher) UnmarshalNotificationElem(bytes []byte, t any) error {
|
||||
var notification sdkws.NotificationElem
|
||||
if err := json.Unmarshal(bytes, ¬ification); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return json.Unmarshal([]byte(notification.Detail), t)
|
||||
}
|
||||
|
||||
/*
|
||||
k8s deployment,offline push group messages function
|
||||
*/
|
||||
func (p *Pusher) k8sOfflinePush2SuperGroup(ctx context.Context, groupID string, msg *sdkws.MsgData, wsResults []*msggateway.SingleMsgToUserResults) error {
|
||||
|
||||
var needOfflinePushUserIDs []string
|
||||
for _, v := range wsResults {
|
||||
if !v.OnlinePush {
|
||||
needOfflinePushUserIDs = append(needOfflinePushUserIDs, v.UserID)
|
||||
}
|
||||
}
|
||||
if len(needOfflinePushUserIDs) > 0 {
|
||||
var offlinePushUserIDs []string
|
||||
err := callbackOfflinePush(ctx, needOfflinePushUserIDs, msg, &offlinePushUserIDs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if len(offlinePushUserIDs) > 0 {
|
||||
needOfflinePushUserIDs = offlinePushUserIDs
|
||||
}
|
||||
if msg.ContentType != constant.SignalingNotification {
|
||||
resp, err := p.conversationRpcClient.Client.GetConversationOfflinePushUserIDs(
|
||||
ctx,
|
||||
&conversation.GetConversationOfflinePushUserIDsReq{ConversationID: utils.GenGroupConversationID(groupID), UserIDs: needOfflinePushUserIDs},
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if len(resp.UserIDs) > 0 {
|
||||
err = p.offlinePushMsg(ctx, groupID, msg, resp.UserIDs)
|
||||
if err != nil {
|
||||
log.ZError(ctx, "offlinePushMsg failed", err, "groupID", groupID, "msg", msg)
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
return nil
|
||||
}
|
||||
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 {
|
||||
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
|
||||
if p.UnmarshalNotificationElem(msg.Content, &tips) != nil {
|
||||
return err
|
||||
}
|
||||
defer func(groupID string, userIDs []string) {
|
||||
if err = p.DeleteMemberAndSetConversationSeq(ctx, groupID, userIDs); err != nil {
|
||||
log.ZError(ctx, "MemberQuitNotification DeleteMemberAndSetConversationSeq", err, "groupID", groupID, "userIDs", userIDs)
|
||||
}
|
||||
}(groupID, []string{tips.QuitUser.UserID})
|
||||
pushToUserIDs = append(pushToUserIDs, tips.QuitUser.UserID)
|
||||
case constant.MemberKickedNotification:
|
||||
var tips sdkws.MemberKickedTips
|
||||
if p.UnmarshalNotificationElem(msg.Content, &tips) != nil {
|
||||
return err
|
||||
}
|
||||
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 {
|
||||
log.ZError(ctx, "MemberKickedNotification DeleteMemberAndSetConversationSeq", err, "groupID", groupID, "userIDs", userIDs)
|
||||
}
|
||||
}(groupID, kickedUsers)
|
||||
pushToUserIDs = append(pushToUserIDs, kickedUsers...)
|
||||
case constant.GroupDismissedNotification:
|
||||
if msgprocessor.IsNotification(msgprocessor.GetConversationIDByMsg(msg)) { // 消息先到,通知后到
|
||||
var tips sdkws.GroupDismissedTips
|
||||
if p.UnmarshalNotificationElem(msg.Content, &tips) != nil {
|
||||
return err
|
||||
}
|
||||
log.ZInfo(ctx, "GroupDismissedNotificationInfo****", "groupID", groupID, "num", len(pushToUserIDs), "list", pushToUserIDs)
|
||||
if len(config.Config.Manager.UserID) > 0 {
|
||||
ctx = mcontext.WithOpUserIDContext(ctx, config.Config.Manager.UserID[0])
|
||||
}
|
||||
defer func(groupID string) {
|
||||
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)
|
||||
isOfflinePush := utils.GetSwitchFromOptions(msg.Options, constant.IsOfflinePush)
|
||||
if isOfflinePush && config.Config.Envs.Discovery == "k8s" {
|
||||
return p.k8sOfflinePush2SuperGroup(ctx, groupID, msg, wsResults)
|
||||
}
|
||||
if isOfflinePush && config.Config.Envs.Discovery == "zookeeper" {
|
||||
var (
|
||||
onlineSuccessUserIDs = []string{msg.SendID}
|
||||
webAndPcBackgroundUserIDs []string
|
||||
)
|
||||
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
needOfflinePushUserIDs := utils.DifferenceString(onlineSuccessUserIDs, pushToUserIDs)
|
||||
|
||||
// Use offline push messaging
|
||||
if len(needOfflinePushUserIDs) > 0 {
|
||||
var offlinePushUserIDs []string
|
||||
err = callbackOfflinePush(ctx, needOfflinePushUserIDs, msg, &offlinePushUserIDs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if len(offlinePushUserIDs) > 0 {
|
||||
needOfflinePushUserIDs = offlinePushUserIDs
|
||||
}
|
||||
if msg.ContentType != constant.SignalingNotification {
|
||||
resp, err := p.conversationRpcClient.Client.GetConversationOfflinePushUserIDs(
|
||||
ctx,
|
||||
&conversation.GetConversationOfflinePushUserIDsReq{ConversationID: utils.GenGroupConversationID(groupID), UserIDs: needOfflinePushUserIDs},
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if len(resp.UserIDs) > 0 {
|
||||
err = p.offlinePushMsg(ctx, groupID, msg, resp.UserIDs)
|
||||
if err != nil {
|
||||
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))
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *Pusher) k8sOnlinePush(ctx context.Context, msg *sdkws.MsgData, pushToUserIDs []string) (wsResults []*msggateway.SingleMsgToUserResults, err error) {
|
||||
var usersHost = make(map[string][]string)
|
||||
for _, v := range pushToUserIDs {
|
||||
tHost, err := p.discov.GetUserIdHashGatewayHost(ctx, v)
|
||||
if err != nil {
|
||||
log.ZError(ctx, "get msggateway hash error", err)
|
||||
return nil, err
|
||||
}
|
||||
tUsers, tbl := usersHost[tHost]
|
||||
if tbl {
|
||||
tUsers = append(tUsers, v)
|
||||
usersHost[tHost] = tUsers
|
||||
} else {
|
||||
usersHost[tHost] = []string{v}
|
||||
}
|
||||
}
|
||||
log.ZDebug(ctx, "genUsers send hosts struct:", "usersHost", usersHost)
|
||||
var usersConns = make(map[*grpc.ClientConn][]string)
|
||||
for host, userIds := range usersHost {
|
||||
tconn, _ := p.discov.GetConn(ctx, host)
|
||||
usersConns[tconn] = userIds
|
||||
}
|
||||
var (
|
||||
mu sync.Mutex
|
||||
wg = errgroup.Group{}
|
||||
maxWorkers = config.Config.Push.MaxConcurrentWorkers
|
||||
)
|
||||
if maxWorkers < 3 {
|
||||
maxWorkers = 3
|
||||
}
|
||||
wg.SetLimit(maxWorkers)
|
||||
for conn, userIds := range usersConns {
|
||||
tcon := conn
|
||||
tuserIds := userIds
|
||||
wg.Go(func() error {
|
||||
input := &msggateway.OnlineBatchPushOneMsgReq{MsgData: msg, PushToUserIDs: tuserIds}
|
||||
msgClient := msggateway.NewMsgGatewayClient(tcon)
|
||||
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
|
||||
})
|
||||
}
|
||||
_ = wg.Wait()
|
||||
return wsResults, nil
|
||||
}
|
||||
func (p *Pusher) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.MsgData, pushToUserIDs []string) (wsResults []*msggateway.SingleMsgToUserResults, err error) {
|
||||
if config.Config.Envs.Discovery == "k8s" {
|
||||
return p.k8sOnlinePush(ctx, msg, pushToUserIDs)
|
||||
}
|
||||
conns, err := p.discov.GetConns(ctx, config.Config.RpcRegisterName.OpenImMessageGatewayName)
|
||||
log.ZDebug(ctx, "get gateway conn", "conn length", len(conns))
|
||||
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
|
||||
})
|
||||
}
|
||||
|
||||
_ = wg.Wait()
|
||||
|
||||
// always return nil
|
||||
return wsResults, nil
|
||||
}
|
||||
|
||||
func (p *Pusher) offlinePushMsg(ctx context.Context, conversationID string, msg *sdkws.MsgData, offlinePushUserIDs []string) error {
|
||||
title, content, opts, err := p.getOfflinePushInfos(conversationID, msg)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = p.offlinePusher.Push(ctx, offlinePushUserIDs, title, content, opts)
|
||||
if err != nil {
|
||||
prommetrics.MsgOfflinePushFailedCounter.Inc()
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *Pusher) GetOfflinePushOpts(msg *sdkws.MsgData) (opts *offlinepush.Opts, err error) {
|
||||
opts = &offlinepush.Opts{Signal: &offlinepush.Signal{}}
|
||||
// if msg.ContentType > constant.SignalingNotificationBegin && msg.ContentType < constant.SignalingNotificationEnd {
|
||||
// req := &sdkws.SignalReq{}
|
||||
// if err := proto.Unmarshal(msg.Content, req); err != nil {
|
||||
// return nil, utils.Wrap(err, "")
|
||||
// }
|
||||
// switch req.Payload.(type) {
|
||||
// case *sdkws.SignalReq_Invite, *sdkws.SignalReq_InviteInGroup:
|
||||
// opts.Signal = &offlinepush.Signal{ClientMsgID: msg.ClientMsgID}
|
||||
// }
|
||||
// }
|
||||
if msg.OfflinePushInfo != nil {
|
||||
opts.IOSBadgeCount = msg.OfflinePushInfo.IOSBadgeCount
|
||||
opts.IOSPushSound = msg.OfflinePushInfo.IOSPushSound
|
||||
opts.Ex = msg.OfflinePushInfo.Ex
|
||||
}
|
||||
return opts, nil
|
||||
}
|
||||
|
||||
func (p *Pusher) getOfflinePushInfos(conversationID string, msg *sdkws.MsgData) (title, content string, opts *offlinepush.Opts, err error) {
|
||||
if p.offlinePusher == nil {
|
||||
err = errNoOfflinePusher
|
||||
return
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
if title == "" {
|
||||
switch msg.ContentType {
|
||||
case constant.Text:
|
||||
fallthrough
|
||||
case constant.Picture:
|
||||
fallthrough
|
||||
case constant.Voice:
|
||||
fallthrough
|
||||
case constant.Video:
|
||||
fallthrough
|
||||
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) {
|
||||
title = constant.ContentType2PushContent[constant.AtText] + constant.ContentType2PushContent[constant.Common]
|
||||
} else {
|
||||
title = constant.ContentType2PushContent[constant.GroupMsg]
|
||||
}
|
||||
case constant.SignalingNotification:
|
||||
title = constant.ContentType2PushContent[constant.SignalMsg]
|
||||
default:
|
||||
title = constant.ContentType2PushContent[constant.Common]
|
||||
}
|
||||
}
|
||||
if content == "" {
|
||||
content = title
|
||||
}
|
||||
return
|
||||
}
|
||||
32
internal/push/tools.go
Normal file
32
internal/push/tools.go
Normal file
@@ -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 push
|
||||
|
||||
import (
|
||||
"github.com/OpenIMSDK/protocol/constant"
|
||||
"github.com/OpenIMSDK/protocol/sdkws"
|
||||
"google.golang.org/protobuf/proto"
|
||||
)
|
||||
|
||||
func GetContent(msg *sdkws.MsgData) string {
|
||||
if msg.ContentType >= constant.NotificationBegin && msg.ContentType <= constant.NotificationEnd {
|
||||
var tips sdkws.TipsComm
|
||||
_ = proto.Unmarshal(msg.Content, &tips)
|
||||
content := tips.JsonDetail
|
||||
return content
|
||||
} else {
|
||||
return string(msg.Content)
|
||||
}
|
||||
}
|
||||
@@ -17,9 +17,8 @@ package conversation
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sort"
|
||||
|
||||
"github.com/OpenIMSDK/protocol/sdkws"
|
||||
"sort"
|
||||
|
||||
"github.com/OpenIMSDK/tools/tx"
|
||||
|
||||
|
||||
@@ -16,6 +16,7 @@ package friend
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/OpenIMSDK/tools/tx"
|
||||
|
||||
"github.com/OpenIMSDK/protocol/sdkws"
|
||||
|
||||
@@ -18,7 +18,6 @@ import (
|
||||
"context"
|
||||
|
||||
utils2 "github.com/OpenIMSDK/tools/utils"
|
||||
|
||||
cbapi "github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
@@ -44,7 +43,7 @@ func (m *msgServer) GetConversationsHasReadAndMaxSeq(ctx context.Context, req *m
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
conversations, err := m.ConversationLocalCache.GetConversations(ctx, req.UserID, conversationIDs)
|
||||
conversations, err := m.Conversation.GetConversations(ctx, req.UserID, conversationIDs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -107,7 +106,7 @@ func (m *msgServer) MarkMsgsAsRead(
|
||||
if hasReadSeq > maxSeq {
|
||||
return nil, errs.ErrArgs.Wrap("hasReadSeq must not be bigger than maxSeq")
|
||||
}
|
||||
conversation, err := m.ConversationLocalCache.GetConversation(ctx, req.UserID, req.ConversationID)
|
||||
conversation, err := m.Conversation.GetConversation(ctx, req.UserID, req.ConversationID)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
@@ -147,7 +146,7 @@ func (m *msgServer) MarkConversationAsRead(
|
||||
ctx context.Context,
|
||||
req *msg.MarkConversationAsReadReq,
|
||||
) (resp *msg.MarkConversationAsReadResp, err error) {
|
||||
conversation, err := m.ConversationLocalCache.GetConversation(ctx, req.UserID, req.ConversationID)
|
||||
conversation, err := m.Conversation.GetConversation(ctx, req.UserID, req.ConversationID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -47,7 +47,7 @@ func (m *msgServer) RevokeMsg(ctx context.Context, req *msg.RevokeMsgReq) (*msg.
|
||||
if err := authverify.CheckAccessV3(ctx, req.UserID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
user, err := m.UserLocalCache.GetUserInfo(ctx, req.UserID)
|
||||
user, err := m.User.GetUserInfo(ctx, req.UserID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -73,7 +73,12 @@ func (m *msgServer) RevokeMsg(ctx context.Context, req *msg.RevokeMsgReq) (*msg.
|
||||
}
|
||||
role = user.AppMangerLevel
|
||||
case constant.SuperGroupChatType:
|
||||
members, err := m.GroupLocalCache.GetGroupMemberInfoMap(ctx, msgs[0].GroupID, utils.Distinct([]string{req.UserID, msgs[0].SendID}))
|
||||
members, err := m.Group.GetGroupMemberInfoMap(
|
||||
ctx,
|
||||
msgs[0].GroupID,
|
||||
utils.Distinct([]string{req.UserID, msgs[0].SendID}),
|
||||
true,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -98,7 +98,7 @@ func (m *msgServer) setConversationAtInfo(nctx context.Context, msg *sdkws.MsgDa
|
||||
}
|
||||
tagAll := utils.IsContain(constant.AtAllString, msg.AtUserIDList)
|
||||
if tagAll {
|
||||
memberUserIDList, err := m.GroupLocalCache.GetGroupMemberIDs(ctx, msg.GroupID)
|
||||
memberUserIDList, err := m.Group.GetGroupMemberIDs(ctx, msg.GroupID)
|
||||
if err != nil {
|
||||
log.ZWarn(ctx, "GetGroupMemberIDs", err)
|
||||
return
|
||||
@@ -144,7 +144,6 @@ func (m *msgServer) sendMsgNotification(
|
||||
}
|
||||
|
||||
func (m *msgServer) sendMsgSingleChat(ctx context.Context, req *pbmsg.SendMsgReq) (resp *pbmsg.SendMsgResp, err error) {
|
||||
log.ZDebug(ctx, "sendMsgSingleChat return")
|
||||
if err := m.messageVerification(ctx, req); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -16,7 +16,6 @@ package msg
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
pbmsg "github.com/OpenIMSDK/protocol/msg"
|
||||
)
|
||||
|
||||
|
||||
@@ -16,7 +16,6 @@ package msg
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/rpccache"
|
||||
|
||||
"google.golang.org/grpc"
|
||||
|
||||
@@ -27,6 +26,7 @@ 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/db/unrelation"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
|
||||
)
|
||||
@@ -36,11 +36,12 @@ type (
|
||||
msgServer struct {
|
||||
RegisterCenter discoveryregistry.SvcDiscoveryRegistry
|
||||
MsgDatabase controller.CommonMsgDatabase
|
||||
Group *rpcclient.GroupRpcClient
|
||||
User *rpcclient.UserRpcClient
|
||||
Conversation *rpcclient.ConversationRpcClient
|
||||
UserLocalCache *rpccache.UserLocalCache
|
||||
FriendLocalCache *rpccache.FriendLocalCache
|
||||
GroupLocalCache *rpccache.GroupLocalCache
|
||||
ConversationLocalCache *rpccache.ConversationLocalCache
|
||||
friend *rpcclient.FriendRpcClient
|
||||
GroupLocalCache *localcache.GroupLocalCache
|
||||
ConversationLocalCache *localcache.ConversationLocalCache
|
||||
Handlers MessageInterceptorChain
|
||||
notificationSender *rpcclient.NotificationSender
|
||||
}
|
||||
@@ -80,15 +81,15 @@ func Start(client discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) e
|
||||
groupRpcClient := rpcclient.NewGroupRpcClient(client)
|
||||
friendRpcClient := rpcclient.NewFriendRpcClient(client)
|
||||
msgDatabase := controller.NewCommonMsgDatabase(msgDocModel, cacheModel)
|
||||
|
||||
s := &msgServer{
|
||||
Conversation: &conversationClient,
|
||||
User: &userRpcClient,
|
||||
Group: &groupRpcClient,
|
||||
MsgDatabase: msgDatabase,
|
||||
RegisterCenter: client,
|
||||
UserLocalCache: rpccache.NewUserLocalCache(userRpcClient, rdb),
|
||||
GroupLocalCache: rpccache.NewGroupLocalCache(groupRpcClient, rdb),
|
||||
ConversationLocalCache: rpccache.NewConversationLocalCache(conversationClient, rdb),
|
||||
FriendLocalCache: rpccache.NewFriendLocalCache(friendRpcClient, rdb),
|
||||
GroupLocalCache: localcache.NewGroupLocalCache(&groupRpcClient),
|
||||
ConversationLocalCache: localcache.NewConversationLocalCache(&conversationClient),
|
||||
friend: &friendRpcClient,
|
||||
}
|
||||
s.notificationSender = rpcclient.NewNotificationSender(rpcclient.WithLocalSendMsg(s.SendMsg))
|
||||
s.addInterceptorHandler(MessageHasReadEnabled)
|
||||
|
||||
@@ -41,7 +41,7 @@ func (m *msgServer) GetActiveUser(ctx context.Context, req *msg.GetActiveUserReq
|
||||
var pbUsers []*msg.ActiveUser
|
||||
if len(users) > 0 {
|
||||
userIDs := utils.Slice(users, func(e *unrelation.UserCount) string { return e.UserID })
|
||||
userMap, err := m.UserLocalCache.GetUsersInfoMap(ctx, userIDs)
|
||||
userMap, err := m.User.GetUsersInfoMap(ctx, userIDs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -83,7 +83,7 @@ func (m *msgServer) GetActiveGroup(ctx context.Context, req *msg.GetActiveGroupR
|
||||
var pbgroups []*msg.ActiveGroup
|
||||
if len(groups) > 0 {
|
||||
groupIDs := utils.Slice(groups, func(e *unrelation.GroupCount) string { return e.GroupID })
|
||||
resp, err := m.GroupLocalCache.GetGroupInfos(ctx, groupIDs)
|
||||
resp, err := m.Group.GetGroupInfos(ctx, groupIDs, false)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -37,7 +37,7 @@ func (m *msgServer) PullMessageBySeqs(
|
||||
resp.NotificationMsgs = make(map[string]*sdkws.PullMsgs)
|
||||
for _, seq := range req.SeqRanges {
|
||||
if !msgprocessor.IsNotification(seq.ConversationID) {
|
||||
conversation, err := m.ConversationLocalCache.GetConversation(ctx, req.UserID, seq.ConversationID)
|
||||
conversation, err := m.Conversation.GetConversation(ctx, req.UserID, seq.ConversationID)
|
||||
if err != nil {
|
||||
log.ZError(ctx, "GetConversation error", err, "conversationID", seq.ConversationID)
|
||||
continue
|
||||
@@ -140,7 +140,7 @@ func (m *msgServer) SearchMessage(ctx context.Context, req *msg.SearchMessageReq
|
||||
}
|
||||
}
|
||||
if len(sendIDs) != 0 {
|
||||
sendInfos, err := m.UserLocalCache.GetUsersInfo(ctx, sendIDs)
|
||||
sendInfos, err := m.User.GetUsersInfo(ctx, sendIDs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -149,7 +149,7 @@ func (m *msgServer) SearchMessage(ctx context.Context, req *msg.SearchMessageReq
|
||||
}
|
||||
}
|
||||
if len(recvIDs) != 0 {
|
||||
recvInfos, err := m.UserLocalCache.GetUsersInfo(ctx, recvIDs)
|
||||
recvInfos, err := m.User.GetUsersInfo(ctx, recvIDs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -158,7 +158,7 @@ func (m *msgServer) SearchMessage(ctx context.Context, req *msg.SearchMessageReq
|
||||
}
|
||||
}
|
||||
if len(groupIDs) != 0 {
|
||||
groupInfos, err := m.GroupLocalCache.GetGroupInfos(ctx, groupIDs)
|
||||
groupInfos, err := m.Group.GetGroupInfos(ctx, groupIDs, true)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -16,7 +16,6 @@ package msg
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/OpenIMSDK/tools/log"
|
||||
"math/rand"
|
||||
"strconv"
|
||||
"time"
|
||||
@@ -59,7 +58,7 @@ func (m *msgServer) messageVerification(ctx context.Context, data *msg.SendMsgRe
|
||||
data.MsgData.ContentType >= constant.NotificationBegin {
|
||||
return nil
|
||||
}
|
||||
black, err := m.FriendLocalCache.IsBlack(ctx, data.MsgData.SendID, data.MsgData.RecvID)
|
||||
black, err := m.friend.IsBlocked(ctx, data.MsgData.SendID, data.MsgData.RecvID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -67,7 +66,7 @@ func (m *msgServer) messageVerification(ctx context.Context, data *msg.SendMsgRe
|
||||
return errs.ErrBlockedByPeer.Wrap()
|
||||
}
|
||||
if *config.Config.MessageVerify.FriendVerify {
|
||||
friend, err := m.FriendLocalCache.IsFriend(ctx, data.MsgData.SendID, data.MsgData.RecvID)
|
||||
friend, err := m.friend.IsFriend(ctx, data.MsgData.SendID, data.MsgData.RecvID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -78,7 +77,7 @@ func (m *msgServer) messageVerification(ctx context.Context, data *msg.SendMsgRe
|
||||
}
|
||||
return nil
|
||||
case constant.SuperGroupChatType:
|
||||
groupInfo, err := m.GroupLocalCache.GetGroupInfo(ctx, data.MsgData.GroupID)
|
||||
groupInfo, err := m.Group.GetGroupInfoCache(ctx, data.MsgData.GroupID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -96,17 +95,17 @@ func (m *msgServer) messageVerification(ctx context.Context, data *msg.SendMsgRe
|
||||
data.MsgData.ContentType >= constant.NotificationBegin {
|
||||
return nil
|
||||
}
|
||||
memberIDs, err := m.GroupLocalCache.GetGroupMemberIDMap(ctx, data.MsgData.GroupID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, ok := memberIDs[data.MsgData.SendID]; !ok {
|
||||
return errs.ErrNotInGroupYet.Wrap()
|
||||
}
|
||||
// memberIDs, err := m.GroupLocalCache.GetGroupMemberIDs(ctx, data.MsgData.GroupID)
|
||||
// if err != nil {
|
||||
// return err
|
||||
// }
|
||||
// if !utils.IsContain(data.MsgData.SendID, memberIDs) {
|
||||
// return errs.ErrNotInGroupYet.Wrap()
|
||||
// }
|
||||
|
||||
groupMemberInfo, err := m.GroupLocalCache.GetGroupMember(ctx, data.MsgData.GroupID, data.MsgData.SendID)
|
||||
groupMemberInfo, err := m.Group.GetGroupMemberCache(ctx, data.MsgData.GroupID, data.MsgData.SendID)
|
||||
if err != nil {
|
||||
if errs.ErrRecordNotFound.Is(err) {
|
||||
if err == errs.ErrRecordNotFound {
|
||||
return errs.ErrNotInGroupYet.Wrap(err.Error())
|
||||
}
|
||||
return err
|
||||
@@ -187,8 +186,7 @@ func (m *msgServer) modifyMessageByUserMessageReceiveOpt(
|
||||
sessionType int,
|
||||
pb *msg.SendMsgReq,
|
||||
) (bool, error) {
|
||||
defer log.ZDebug(ctx, "modifyMessageByUserMessageReceiveOpt return")
|
||||
opt, err := m.UserLocalCache.GetUserGlobalMsgRecvOpt(ctx, userID)
|
||||
opt, err := m.User.GetUserGlobalMsgRecvOpt(ctx, userID)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
@@ -204,7 +202,7 @@ func (m *msgServer) modifyMessageByUserMessageReceiveOpt(
|
||||
return true, nil
|
||||
}
|
||||
// conversationID := utils.GetConversationIDBySessionType(conversationID, sessionType)
|
||||
singleOpt, err := m.ConversationLocalCache.GetSingleConversationRecvMsgOpt(ctx, userID, conversationID)
|
||||
singleOpt, err := m.Conversation.GetSingleConversationRecvMsgOpt(ctx, userID, conversationID)
|
||||
if errs.ErrRecordNotFound.Is(err) {
|
||||
return true, nil
|
||||
} else if err != nil {
|
||||
|
||||
@@ -19,14 +19,12 @@ import (
|
||||
"encoding/base64"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"github.com/google/uuid"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/authverify"
|
||||
"path"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/authverify"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/s3"
|
||||
|
||||
"github.com/OpenIMSDK/protocol/third"
|
||||
|
||||
@@ -16,7 +16,6 @@ package user
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
pbuser "github.com/OpenIMSDK/protocol/user"
|
||||
"github.com/OpenIMSDK/tools/utils"
|
||||
|
||||
|
||||
@@ -17,14 +17,12 @@ package user
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"github.com/OpenIMSDK/tools/pagination"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
|
||||
"math/rand"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/OpenIMSDK/tools/pagination"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
|
||||
|
||||
"github.com/OpenIMSDK/tools/tx"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/db/mgo"
|
||||
@@ -61,7 +59,6 @@ type userServer struct {
|
||||
RegisterCenter registry.SvcDiscoveryRegistry
|
||||
}
|
||||
|
||||
|
||||
func Start(client registry.SvcDiscoveryRegistry, server *grpc.Server) error {
|
||||
rdb, err := cache.NewRedis()
|
||||
if err != nil {
|
||||
@@ -244,7 +241,6 @@ func (s *userServer) GetPaginationUsers(ctx context.Context, req *pbuser.GetPagi
|
||||
return nil, err
|
||||
}
|
||||
return &pbuser.GetPaginationUsersResp{Total: int32(total), Users: convert.UsersDB2Pb(users)}, err
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -391,7 +387,7 @@ func (s *userServer) GetSubscribeUsersStatus(ctx context.Context,
|
||||
return &pbuser.GetSubscribeUsersStatusResp{StatusList: onlineStatusList}, nil
|
||||
}
|
||||
|
||||
// ProcessUserCommandAdd user general function add.
|
||||
// ProcessUserCommandAdd user general function add
|
||||
func (s *userServer) ProcessUserCommandAdd(ctx context.Context, req *pbuser.ProcessUserCommandAddReq) (*pbuser.ProcessUserCommandAddResp, error) {
|
||||
err := authverify.CheckAccessV3(ctx, req.UserID)
|
||||
if err != nil {
|
||||
@@ -422,7 +418,7 @@ func (s *userServer) ProcessUserCommandAdd(ctx context.Context, req *pbuser.Proc
|
||||
return &pbuser.ProcessUserCommandAddResp{}, nil
|
||||
}
|
||||
|
||||
// ProcessUserCommandDelete user general function delete.
|
||||
// ProcessUserCommandDelete user general function delete
|
||||
func (s *userServer) ProcessUserCommandDelete(ctx context.Context, req *pbuser.ProcessUserCommandDeleteReq) (*pbuser.ProcessUserCommandDeleteResp, error) {
|
||||
err := authverify.CheckAccessV3(ctx, req.UserID)
|
||||
if err != nil {
|
||||
@@ -441,11 +437,10 @@ func (s *userServer) ProcessUserCommandDelete(ctx context.Context, req *pbuser.P
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &pbuser.ProcessUserCommandDeleteResp{}, nil
|
||||
}
|
||||
|
||||
// ProcessUserCommandUpdate user general function update.
|
||||
// ProcessUserCommandUpdate user general function update
|
||||
func (s *userServer) ProcessUserCommandUpdate(ctx context.Context, req *pbuser.ProcessUserCommandUpdateReq) (*pbuser.ProcessUserCommandUpdateResp, error) {
|
||||
err := authverify.CheckAccessV3(ctx, req.UserID)
|
||||
if err != nil {
|
||||
@@ -478,7 +473,6 @@ func (s *userServer) ProcessUserCommandUpdate(ctx context.Context, req *pbuser.P
|
||||
}
|
||||
|
||||
func (s *userServer) ProcessUserCommandGet(ctx context.Context, req *pbuser.ProcessUserCommandGetReq) (*pbuser.ProcessUserCommandGetResp, error) {
|
||||
|
||||
err := authverify.CheckAccessV3(ctx, req.UserID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
||||
@@ -79,7 +79,7 @@ func InitMsgTool() (*MsgTool, error) {
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
discov.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")))
|
||||
discov.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()))
|
||||
userDB, err := mgo.NewUserMongo(mongo.GetDatabase())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
||||
@@ -1,15 +0,0 @@
|
||||
package cachekey
|
||||
|
||||
const (
|
||||
BlackIDsKey = "BLACK_IDS:"
|
||||
IsBlackKey = "IS_BLACK:" // local cache
|
||||
)
|
||||
|
||||
func GetBlackIDsKey(ownerUserID string) string {
|
||||
return BlackIDsKey + ownerUserID
|
||||
|
||||
}
|
||||
|
||||
func GetIsBlackIDsKey(possibleBlackUserID, userID string) string {
|
||||
return IsBlackKey + userID + "-" + possibleBlackUserID
|
||||
}
|
||||
@@ -1,44 +0,0 @@
|
||||
package cachekey
|
||||
|
||||
const (
|
||||
ConversationKey = "CONVERSATION:"
|
||||
ConversationIDsKey = "CONVERSATION_IDS:"
|
||||
ConversationIDsHashKey = "CONVERSATION_IDS_HASH:"
|
||||
ConversationHasReadSeqKey = "CONVERSATION_HAS_READ_SEQ:"
|
||||
RecvMsgOptKey = "RECV_MSG_OPT:"
|
||||
SuperGroupRecvMsgNotNotifyUserIDsKey = "SUPER_GROUP_RECV_MSG_NOT_NOTIFY_USER_IDS:"
|
||||
SuperGroupRecvMsgNotNotifyUserIDsHashKey = "SUPER_GROUP_RECV_MSG_NOT_NOTIFY_USER_IDS_HASH:"
|
||||
ConversationNotReceiveMessageUserIDsKey = "CONVERSATION_NOT_RECEIVE_MESSAGE_USER_IDS:"
|
||||
)
|
||||
|
||||
func GetConversationKey(ownerUserID, conversationID string) string {
|
||||
return ConversationKey + ownerUserID + ":" + conversationID
|
||||
}
|
||||
|
||||
func GetConversationIDsKey(ownerUserID string) string {
|
||||
return ConversationIDsKey + ownerUserID
|
||||
}
|
||||
|
||||
func GetSuperGroupRecvNotNotifyUserIDsKey(groupID string) string {
|
||||
return SuperGroupRecvMsgNotNotifyUserIDsKey + groupID
|
||||
}
|
||||
|
||||
func GetRecvMsgOptKey(ownerUserID, conversationID string) string {
|
||||
return RecvMsgOptKey + ownerUserID + ":" + conversationID
|
||||
}
|
||||
|
||||
func GetSuperGroupRecvNotNotifyUserIDsHashKey(groupID string) string {
|
||||
return SuperGroupRecvMsgNotNotifyUserIDsHashKey + groupID
|
||||
}
|
||||
|
||||
func GetConversationHasReadSeqKey(ownerUserID, conversationID string) string {
|
||||
return ConversationHasReadSeqKey + ownerUserID + ":" + conversationID
|
||||
}
|
||||
|
||||
func GetConversationNotReceiveMessageUserIDsKey(conversationID string) string {
|
||||
return ConversationNotReceiveMessageUserIDsKey + conversationID
|
||||
}
|
||||
|
||||
func GetUserConversationIDsHashKey(ownerUserID string) string {
|
||||
return ConversationIDsHashKey + ownerUserID
|
||||
}
|
||||
@@ -1,24 +0,0 @@
|
||||
package cachekey
|
||||
|
||||
const (
|
||||
FriendIDsKey = "FRIEND_IDS:"
|
||||
TwoWayFriendsIDsKey = "COMMON_FRIENDS_IDS:"
|
||||
FriendKey = "FRIEND_INFO:"
|
||||
IsFriendKey = "IS_FRIEND:" // local cache key
|
||||
)
|
||||
|
||||
func GetFriendIDsKey(ownerUserID string) string {
|
||||
return FriendIDsKey + ownerUserID
|
||||
}
|
||||
|
||||
func GetTwoWayFriendsIDsKey(ownerUserID string) string {
|
||||
return TwoWayFriendsIDsKey + ownerUserID
|
||||
}
|
||||
|
||||
func GetFriendKey(ownerUserID, friendUserID string) string {
|
||||
return FriendKey + ownerUserID + "-" + friendUserID
|
||||
}
|
||||
|
||||
func GetIsFriendKey(possibleFriendUserID, userID string) string {
|
||||
return IsFriendKey + possibleFriendUserID + "-" + userID
|
||||
}
|
||||
@@ -1,45 +0,0 @@
|
||||
package cachekey
|
||||
|
||||
import (
|
||||
"strconv"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
groupExpireTime = time.Second * 60 * 60 * 12
|
||||
GroupInfoKey = "GROUP_INFO:"
|
||||
GroupMemberIDsKey = "GROUP_MEMBER_IDS:"
|
||||
GroupMembersHashKey = "GROUP_MEMBERS_HASH2:"
|
||||
GroupMemberInfoKey = "GROUP_MEMBER_INFO:"
|
||||
JoinedGroupsKey = "JOIN_GROUPS_KEY:"
|
||||
GroupMemberNumKey = "GROUP_MEMBER_NUM_CACHE:"
|
||||
GroupRoleLevelMemberIDsKey = "GROUP_ROLE_LEVEL_MEMBER_IDS:"
|
||||
)
|
||||
|
||||
func GetGroupInfoKey(groupID string) string {
|
||||
return GroupInfoKey + groupID
|
||||
}
|
||||
|
||||
func GetJoinedGroupsKey(userID string) string {
|
||||
return JoinedGroupsKey + userID
|
||||
}
|
||||
|
||||
func GetGroupMembersHashKey(groupID string) string {
|
||||
return GroupMembersHashKey + groupID
|
||||
}
|
||||
|
||||
func GetGroupMemberIDsKey(groupID string) string {
|
||||
return GroupMemberIDsKey + groupID
|
||||
}
|
||||
|
||||
func GetGroupMemberInfoKey(groupID, userID string) string {
|
||||
return GroupMemberInfoKey + groupID + "-" + userID
|
||||
}
|
||||
|
||||
func GetGroupMemberNumKey(groupID string) string {
|
||||
return GroupMemberNumKey + groupID
|
||||
}
|
||||
|
||||
func GetGroupRoleLevelMemberIDsKey(groupID string, roleLevel int32) string {
|
||||
return GroupRoleLevelMemberIDsKey + groupID + "-" + strconv.Itoa(int(roleLevel))
|
||||
}
|
||||
@@ -1,14 +0,0 @@
|
||||
package cachekey
|
||||
|
||||
const (
|
||||
UserInfoKey = "USER_INFO:"
|
||||
UserGlobalRecvMsgOptKey = "USER_GLOBAL_RECV_MSG_OPT_KEY:"
|
||||
)
|
||||
|
||||
func GetUserInfoKey(userID string) string {
|
||||
return UserInfoKey + userID
|
||||
}
|
||||
|
||||
func GetUserGlobalRecvMsgOptKey(userID string) string {
|
||||
return UserGlobalRecvMsgOptKey + userID
|
||||
}
|
||||
@@ -16,7 +16,6 @@ package config
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"time"
|
||||
|
||||
"github.com/OpenIMSDK/tools/discoveryregistry"
|
||||
"gopkg.in/yaml.v3"
|
||||
@@ -259,8 +258,6 @@ type configStruct struct {
|
||||
FriendVerify *bool `yaml:"friendVerify"`
|
||||
} `yaml:"messageVerify"`
|
||||
|
||||
LocalCache localCache `yaml:"localCache"`
|
||||
|
||||
IOSPush struct {
|
||||
PushSound string `yaml:"pushSound"`
|
||||
BadgeCount bool `yaml:"badgeCount"`
|
||||
@@ -373,33 +370,6 @@ type notification struct {
|
||||
ConversationSetPrivate NotificationConf `yaml:"conversationSetPrivate"`
|
||||
}
|
||||
|
||||
type LocalCache struct {
|
||||
Topic string `yaml:"topic"`
|
||||
SlotNum int `yaml:"slotNum"`
|
||||
SlotSize int `yaml:"slotSize"`
|
||||
SuccessExpire int `yaml:"successExpire"` // second
|
||||
FailedExpire int `yaml:"failedExpire"` // second
|
||||
}
|
||||
|
||||
func (l LocalCache) Failed() time.Duration {
|
||||
return time.Second * time.Duration(l.FailedExpire)
|
||||
}
|
||||
|
||||
func (l LocalCache) Success() time.Duration {
|
||||
return time.Second * time.Duration(l.SuccessExpire)
|
||||
}
|
||||
|
||||
func (l LocalCache) Enable() bool {
|
||||
return l.Topic != "" && l.SlotNum > 0 && l.SlotSize > 0
|
||||
}
|
||||
|
||||
type localCache struct {
|
||||
User LocalCache `yaml:"user"`
|
||||
Group LocalCache `yaml:"group"`
|
||||
Friend LocalCache `yaml:"friend"`
|
||||
Conversation LocalCache `yaml:"conversation"`
|
||||
}
|
||||
|
||||
func (c *configStruct) GetServiceNames() []string {
|
||||
return []string{
|
||||
c.RpcRegisterName.OpenImUserName,
|
||||
|
||||
@@ -16,8 +16,6 @@ package config
|
||||
|
||||
import (
|
||||
_ "embed"
|
||||
"fmt"
|
||||
"gopkg.in/yaml.v3"
|
||||
"reflect"
|
||||
"testing"
|
||||
|
||||
@@ -117,13 +115,3 @@ func TestInitConfig(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestName(t *testing.T) {
|
||||
Config.LocalCache.Friend.Topic = "friend"
|
||||
Config.LocalCache.Friend.SlotNum = 500
|
||||
Config.LocalCache.Friend.SlotSize = 20000
|
||||
|
||||
data, _ := yaml.Marshal(&Config)
|
||||
|
||||
fmt.Println(string(data))
|
||||
}
|
||||
|
||||
@@ -17,7 +17,6 @@ package convert
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/OpenIMSDK/protocol/sdkws"
|
||||
"github.com/OpenIMSDK/tools/utils"
|
||||
|
||||
|
||||
15
pkg/common/db/cache/black.go
vendored
15
pkg/common/db/cache/black.go
vendored
@@ -16,9 +16,6 @@ package cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/OpenIMSDK/tools/log"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/cachekey"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||
"time"
|
||||
|
||||
"github.com/dtm-labs/rockscache"
|
||||
@@ -55,15 +52,11 @@ func NewBlackCacheRedis(
|
||||
options rockscache.Options,
|
||||
) BlackCache {
|
||||
rcClient := rockscache.NewClient(rdb, options)
|
||||
mc := NewMetaCacheRedis(rcClient)
|
||||
b := config.Config.LocalCache.Friend
|
||||
log.ZDebug(context.Background(), "black local cache init", "Topic", b.Topic, "SlotNum", b.SlotNum, "SlotSize", b.SlotSize, "enable", b.Enable())
|
||||
mc.SetTopic(b.Topic)
|
||||
mc.SetRawRedisClient(rdb)
|
||||
|
||||
return &BlackCacheRedis{
|
||||
expireTime: blackExpireTime,
|
||||
rcClient: rcClient,
|
||||
metaCache: mc,
|
||||
metaCache: NewMetaCacheRedis(rcClient),
|
||||
blackDB: blackDB,
|
||||
}
|
||||
}
|
||||
@@ -73,12 +66,12 @@ func (b *BlackCacheRedis) NewCache() BlackCache {
|
||||
expireTime: b.expireTime,
|
||||
rcClient: b.rcClient,
|
||||
blackDB: b.blackDB,
|
||||
metaCache: b.Copy(),
|
||||
metaCache: NewMetaCacheRedis(b.rcClient, b.metaCache.GetPreDelKeys()...),
|
||||
}
|
||||
}
|
||||
|
||||
func (b *BlackCacheRedis) getBlackIDsKey(ownerUserID string) string {
|
||||
return cachekey.GetBlackIDsKey(ownerUserID)
|
||||
return blackIDsKey + ownerUserID
|
||||
}
|
||||
|
||||
func (b *BlackCacheRedis) GetBlackIDs(ctx context.Context, userID string) (blackIDs []string, err error) {
|
||||
|
||||
66
pkg/common/db/cache/config.go
vendored
66
pkg/common/db/cache/config.go
vendored
@@ -1,66 +0,0 @@
|
||||
package cache
|
||||
|
||||
import (
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/cachekey"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||
"strings"
|
||||
"sync"
|
||||
)
|
||||
|
||||
var (
|
||||
once sync.Once
|
||||
subscribe map[string][]string
|
||||
)
|
||||
|
||||
func getPublishKey(topic string, key []string) []string {
|
||||
if topic == "" || len(key) == 0 {
|
||||
return nil
|
||||
}
|
||||
once.Do(func() {
|
||||
list := []struct {
|
||||
Local config.LocalCache
|
||||
Keys []string
|
||||
}{
|
||||
{
|
||||
Local: config.Config.LocalCache.User,
|
||||
Keys: []string{cachekey.UserInfoKey, cachekey.UserGlobalRecvMsgOptKey},
|
||||
},
|
||||
{
|
||||
Local: config.Config.LocalCache.Group,
|
||||
Keys: []string{cachekey.GroupMemberIDsKey, cachekey.GroupInfoKey, cachekey.GroupMemberInfoKey},
|
||||
},
|
||||
{
|
||||
Local: config.Config.LocalCache.Friend,
|
||||
Keys: []string{cachekey.FriendIDsKey, cachekey.BlackIDsKey},
|
||||
},
|
||||
{
|
||||
Local: config.Config.LocalCache.Conversation,
|
||||
Keys: []string{cachekey.ConversationIDsKey, cachekey.ConversationKey},
|
||||
},
|
||||
}
|
||||
subscribe = make(map[string][]string)
|
||||
for _, v := range list {
|
||||
if v.Local.Enable() {
|
||||
subscribe[v.Local.Topic] = v.Keys
|
||||
}
|
||||
}
|
||||
})
|
||||
prefix, ok := subscribe[topic]
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
res := make([]string, 0, len(key))
|
||||
for _, k := range key {
|
||||
var exist bool
|
||||
for _, p := range prefix {
|
||||
if strings.HasPrefix(k, p) {
|
||||
exist = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if exist {
|
||||
res = append(res, k)
|
||||
}
|
||||
}
|
||||
return res
|
||||
}
|
||||
51
pkg/common/db/cache/conversation.go
vendored
51
pkg/common/db/cache/conversation.go
vendored
@@ -17,9 +17,6 @@ package cache
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"github.com/OpenIMSDK/tools/log"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/cachekey"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||
"math/big"
|
||||
"strings"
|
||||
"time"
|
||||
@@ -33,14 +30,14 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
//conversationKey = "CONVERSATION:"
|
||||
//conversationIDsKey = "CONVERSATION_IDS:"
|
||||
//conversationIDsHashKey = "CONVERSATION_IDS_HASH:"
|
||||
//conversationHasReadSeqKey = "CONVERSATION_HAS_READ_SEQ:"
|
||||
//recvMsgOptKey = "RECV_MSG_OPT:"
|
||||
//superGroupRecvMsgNotNotifyUserIDsKey = "SUPER_GROUP_RECV_MSG_NOT_NOTIFY_USER_IDS:"
|
||||
//superGroupRecvMsgNotNotifyUserIDsHashKey = "SUPER_GROUP_RECV_MSG_NOT_NOTIFY_USER_IDS_HASH:"
|
||||
//conversationNotReceiveMessageUserIDsKey = "CONVERSATION_NOT_RECEIVE_MESSAGE_USER_IDS:"
|
||||
conversationKey = "CONVERSATION:"
|
||||
conversationIDsKey = "CONVERSATION_IDS:"
|
||||
conversationIDsHashKey = "CONVERSATION_IDS_HASH:"
|
||||
conversationHasReadSeqKey = "CONVERSATION_HAS_READ_SEQ:"
|
||||
recvMsgOptKey = "RECV_MSG_OPT:"
|
||||
superGroupRecvMsgNotNotifyUserIDsKey = "SUPER_GROUP_RECV_MSG_NOT_NOTIFY_USER_IDS:"
|
||||
superGroupRecvMsgNotNotifyUserIDsHashKey = "SUPER_GROUP_RECV_MSG_NOT_NOTIFY_USER_IDS_HASH:"
|
||||
conversationNotReceiveMessageUserIDsKey = "CONVERSATION_NOT_RECEIVE_MESSAGE_USER_IDS:"
|
||||
|
||||
conversationExpireTime = time.Second * 60 * 60 * 12
|
||||
)
|
||||
@@ -87,14 +84,10 @@ type ConversationCache interface {
|
||||
|
||||
func NewConversationRedis(rdb redis.UniversalClient, opts rockscache.Options, db relationtb.ConversationModelInterface) ConversationCache {
|
||||
rcClient := rockscache.NewClient(rdb, opts)
|
||||
mc := NewMetaCacheRedis(rcClient)
|
||||
c := config.Config.LocalCache.Conversation
|
||||
log.ZDebug(context.Background(), "black local cache init", "Topic", c.Topic, "SlotNum", c.SlotNum, "SlotSize", c.SlotSize, "enable", c.Enable())
|
||||
mc.SetTopic(c.Topic)
|
||||
mc.SetRawRedisClient(rdb)
|
||||
|
||||
return &ConversationRedisCache{
|
||||
rcClient: rcClient,
|
||||
metaCache: mc,
|
||||
metaCache: NewMetaCacheRedis(rcClient),
|
||||
conversationDB: db,
|
||||
expireTime: conversationExpireTime,
|
||||
}
|
||||
@@ -125,42 +118,38 @@ type ConversationRedisCache struct {
|
||||
func (c *ConversationRedisCache) NewCache() ConversationCache {
|
||||
return &ConversationRedisCache{
|
||||
rcClient: c.rcClient,
|
||||
metaCache: c.Copy(),
|
||||
metaCache: NewMetaCacheRedis(c.rcClient, c.metaCache.GetPreDelKeys()...),
|
||||
conversationDB: c.conversationDB,
|
||||
expireTime: c.expireTime,
|
||||
}
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getConversationKey(ownerUserID, conversationID string) string {
|
||||
return cachekey.GetConversationKey(ownerUserID, conversationID)
|
||||
return conversationKey + ownerUserID + ":" + conversationID
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getConversationIDsKey(ownerUserID string) string {
|
||||
return cachekey.GetConversationIDsKey(ownerUserID)
|
||||
return conversationIDsKey + ownerUserID
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getSuperGroupRecvNotNotifyUserIDsKey(groupID string) string {
|
||||
return cachekey.GetSuperGroupRecvNotNotifyUserIDsKey(groupID)
|
||||
return superGroupRecvMsgNotNotifyUserIDsKey + groupID
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getRecvMsgOptKey(ownerUserID, conversationID string) string {
|
||||
return cachekey.GetRecvMsgOptKey(ownerUserID, conversationID)
|
||||
return recvMsgOptKey + ownerUserID + ":" + conversationID
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getSuperGroupRecvNotNotifyUserIDsHashKey(groupID string) string {
|
||||
return cachekey.GetSuperGroupRecvNotNotifyUserIDsHashKey(groupID)
|
||||
return superGroupRecvMsgNotNotifyUserIDsHashKey + groupID
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getConversationHasReadSeqKey(ownerUserID, conversationID string) string {
|
||||
return cachekey.GetConversationHasReadSeqKey(ownerUserID, conversationID)
|
||||
return conversationHasReadSeqKey + ownerUserID + ":" + conversationID
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getConversationNotReceiveMessageUserIDsKey(conversationID string) string {
|
||||
return cachekey.GetConversationNotReceiveMessageUserIDsKey(conversationID)
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getUserConversationIDsHashKey(ownerUserID string) string {
|
||||
return cachekey.GetUserConversationIDsHashKey(ownerUserID)
|
||||
return conversationNotReceiveMessageUserIDsKey + conversationID
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) GetUserConversationIDs(ctx context.Context, ownerUserID string) ([]string, error) {
|
||||
@@ -180,6 +169,10 @@ func (c *ConversationRedisCache) DelConversationIDs(userIDs ...string) Conversat
|
||||
return cache
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) getUserConversationIDsHashKey(ownerUserID string) string {
|
||||
return conversationIDsHashKey + ownerUserID
|
||||
}
|
||||
|
||||
func (c *ConversationRedisCache) GetUserConversationIDsHash(ctx context.Context, ownerUserID string) (hash uint64, err error) {
|
||||
return getCache(
|
||||
ctx,
|
||||
|
||||
26
pkg/common/db/cache/friend.go
vendored
26
pkg/common/db/cache/friend.go
vendored
@@ -16,9 +16,6 @@ package cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/OpenIMSDK/tools/log"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/cachekey"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||
"time"
|
||||
|
||||
"github.com/dtm-labs/rockscache"
|
||||
@@ -30,10 +27,10 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
friendExpireTime = time.Second * 60 * 60 * 12
|
||||
//friendIDsKey = "FRIEND_IDS:"
|
||||
//TwoWayFriendsIDsKey = "COMMON_FRIENDS_IDS:"
|
||||
//friendKey = "FRIEND_INFO:"
|
||||
friendExpireTime = time.Second * 60 * 60 * 12
|
||||
friendIDsKey = "FRIEND_IDS:"
|
||||
TwoWayFriendsIDsKey = "COMMON_FRIENDS_IDS:"
|
||||
friendKey = "FRIEND_INFO:"
|
||||
)
|
||||
|
||||
// FriendCache is an interface for caching friend-related data.
|
||||
@@ -63,13 +60,8 @@ type FriendCacheRedis struct {
|
||||
func NewFriendCacheRedis(rdb redis.UniversalClient, friendDB relationtb.FriendModelInterface,
|
||||
options rockscache.Options) FriendCache {
|
||||
rcClient := rockscache.NewClient(rdb, options)
|
||||
mc := NewMetaCacheRedis(rcClient)
|
||||
f := config.Config.LocalCache.Friend
|
||||
log.ZDebug(context.Background(), "friend local cache init", "Topic", f.Topic, "SlotNum", f.SlotNum, "SlotSize", f.SlotSize, "enable", f.Enable())
|
||||
mc.SetTopic(f.Topic)
|
||||
mc.SetRawRedisClient(rdb)
|
||||
return &FriendCacheRedis{
|
||||
metaCache: mc,
|
||||
metaCache: NewMetaCacheRedis(rcClient),
|
||||
friendDB: friendDB,
|
||||
expireTime: friendExpireTime,
|
||||
rcClient: rcClient,
|
||||
@@ -80,7 +72,7 @@ func NewFriendCacheRedis(rdb redis.UniversalClient, friendDB relationtb.FriendMo
|
||||
func (f *FriendCacheRedis) NewCache() FriendCache {
|
||||
return &FriendCacheRedis{
|
||||
rcClient: f.rcClient,
|
||||
metaCache: f.Copy(),
|
||||
metaCache: NewMetaCacheRedis(f.rcClient, f.metaCache.GetPreDelKeys()...),
|
||||
friendDB: f.friendDB,
|
||||
expireTime: f.expireTime,
|
||||
}
|
||||
@@ -88,17 +80,17 @@ func (f *FriendCacheRedis) NewCache() FriendCache {
|
||||
|
||||
// getFriendIDsKey returns the key for storing friend IDs in the cache.
|
||||
func (f *FriendCacheRedis) getFriendIDsKey(ownerUserID string) string {
|
||||
return cachekey.GetFriendIDsKey(ownerUserID)
|
||||
return friendIDsKey + ownerUserID
|
||||
}
|
||||
|
||||
// getTwoWayFriendsIDsKey returns the key for storing two-way friend IDs in the cache.
|
||||
func (f *FriendCacheRedis) getTwoWayFriendsIDsKey(ownerUserID string) string {
|
||||
return cachekey.GetTwoWayFriendsIDsKey(ownerUserID)
|
||||
return TwoWayFriendsIDsKey + ownerUserID
|
||||
}
|
||||
|
||||
// getFriendKey returns the key for storing friend info in the cache.
|
||||
func (f *FriendCacheRedis) getFriendKey(ownerUserID, friendUserID string) string {
|
||||
return cachekey.GetFriendKey(ownerUserID, friendUserID)
|
||||
return friendKey + ownerUserID + "-" + friendUserID
|
||||
}
|
||||
|
||||
// GetFriendIDs retrieves friend IDs from the cache or the database if not found.
|
||||
|
||||
44
pkg/common/db/cache/group.go
vendored
44
pkg/common/db/cache/group.go
vendored
@@ -17,8 +17,7 @@ package cache
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/cachekey"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/OpenIMSDK/protocol/constant"
|
||||
@@ -35,14 +34,15 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
groupExpireTime = time.Second * 60 * 60 * 12
|
||||
//groupInfoKey = "GROUP_INFO:"
|
||||
//groupMemberIDsKey = "GROUP_MEMBER_IDS:"
|
||||
//groupMembersHashKey = "GROUP_MEMBERS_HASH2:"
|
||||
//groupMemberInfoKey = "GROUP_MEMBER_INFO:"
|
||||
//joinedGroupsKey = "JOIN_GROUPS_KEY:"
|
||||
//groupMemberNumKey = "GROUP_MEMBER_NUM_CACHE:"
|
||||
//groupRoleLevelMemberIDsKey = "GROUP_ROLE_LEVEL_MEMBER_IDS:"
|
||||
groupExpireTime = time.Second * 60 * 60 * 12
|
||||
groupInfoKey = "GROUP_INFO:"
|
||||
groupMemberIDsKey = "GROUP_MEMBER_IDS:"
|
||||
groupMembersHashKey = "GROUP_MEMBERS_HASH2:"
|
||||
groupMemberInfoKey = "GROUP_MEMBER_INFO:"
|
||||
//groupOwnerInfoKey = "GROUP_OWNER_INFO:".
|
||||
joinedGroupsKey = "JOIN_GROUPS_KEY:"
|
||||
groupMemberNumKey = "GROUP_MEMBER_NUM_CACHE:"
|
||||
groupRoleLevelMemberIDsKey = "GROUP_ROLE_LEVEL_MEMBER_IDS:"
|
||||
)
|
||||
|
||||
type GroupHash interface {
|
||||
@@ -105,16 +105,12 @@ func NewGroupCacheRedis(
|
||||
opts rockscache.Options,
|
||||
) GroupCache {
|
||||
rcClient := rockscache.NewClient(rdb, opts)
|
||||
mc := NewMetaCacheRedis(rcClient)
|
||||
g := config.Config.LocalCache.Group
|
||||
mc.SetTopic(g.Topic)
|
||||
log.ZDebug(context.Background(), "group local cache init", "Topic", g.Topic, "SlotNum", g.SlotNum, "SlotSize", g.SlotSize, "enable", g.Enable())
|
||||
mc.SetRawRedisClient(rdb)
|
||||
|
||||
return &GroupCacheRedis{
|
||||
rcClient: rcClient, expireTime: groupExpireTime,
|
||||
groupDB: groupDB, groupMemberDB: groupMemberDB, groupRequestDB: groupRequestDB,
|
||||
groupHash: hashCode,
|
||||
metaCache: mc,
|
||||
metaCache: NewMetaCacheRedis(rcClient),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -125,36 +121,36 @@ func (g *GroupCacheRedis) NewCache() GroupCache {
|
||||
groupDB: g.groupDB,
|
||||
groupMemberDB: g.groupMemberDB,
|
||||
groupRequestDB: g.groupRequestDB,
|
||||
metaCache: g.Copy(),
|
||||
metaCache: NewMetaCacheRedis(g.rcClient, g.metaCache.GetPreDelKeys()...),
|
||||
}
|
||||
}
|
||||
|
||||
func (g *GroupCacheRedis) getGroupInfoKey(groupID string) string {
|
||||
return cachekey.GetGroupInfoKey(groupID)
|
||||
return groupInfoKey + groupID
|
||||
}
|
||||
|
||||
func (g *GroupCacheRedis) getJoinedGroupsKey(userID string) string {
|
||||
return cachekey.GetJoinedGroupsKey(userID)
|
||||
return joinedGroupsKey + userID
|
||||
}
|
||||
|
||||
func (g *GroupCacheRedis) getGroupMembersHashKey(groupID string) string {
|
||||
return cachekey.GetGroupMembersHashKey(groupID)
|
||||
return groupMembersHashKey + groupID
|
||||
}
|
||||
|
||||
func (g *GroupCacheRedis) getGroupMemberIDsKey(groupID string) string {
|
||||
return cachekey.GetGroupMemberIDsKey(groupID)
|
||||
return groupMemberIDsKey + groupID
|
||||
}
|
||||
|
||||
func (g *GroupCacheRedis) getGroupMemberInfoKey(groupID, userID string) string {
|
||||
return cachekey.GetGroupMemberInfoKey(groupID, userID)
|
||||
return groupMemberInfoKey + groupID + "-" + userID
|
||||
}
|
||||
|
||||
func (g *GroupCacheRedis) getGroupMemberNumKey(groupID string) string {
|
||||
return cachekey.GetGroupMemberNumKey(groupID)
|
||||
return groupMemberNumKey + groupID
|
||||
}
|
||||
|
||||
func (g *GroupCacheRedis) getGroupRoleLevelMemberIDsKey(groupID string, roleLevel int32) string {
|
||||
return cachekey.GetGroupRoleLevelMemberIDsKey(groupID, roleLevel)
|
||||
return groupRoleLevelMemberIDsKey + groupID + "-" + strconv.Itoa(int(roleLevel))
|
||||
}
|
||||
|
||||
func (g *GroupCacheRedis) GetGroupIndex(group *relationtb.GroupModel, keys []string) (int, error) {
|
||||
|
||||
65
pkg/common/db/cache/meta_cache.go
vendored
65
pkg/common/db/cache/meta_cache.go
vendored
@@ -18,7 +18,6 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"github.com/redis/go-redis/v9"
|
||||
"time"
|
||||
|
||||
"github.com/OpenIMSDK/tools/mw/specialerror"
|
||||
@@ -45,9 +44,6 @@ type metaCache interface {
|
||||
AddKeys(keys ...string)
|
||||
ClearKeys()
|
||||
GetPreDelKeys() []string
|
||||
SetTopic(topic string)
|
||||
SetRawRedisClient(cli redis.UniversalClient)
|
||||
Copy() metaCache
|
||||
}
|
||||
|
||||
func NewMetaCacheRedis(rcClient *rockscache.Client, keys ...string) metaCache {
|
||||
@@ -55,36 +51,10 @@ func NewMetaCacheRedis(rcClient *rockscache.Client, keys ...string) metaCache {
|
||||
}
|
||||
|
||||
type metaCacheRedis struct {
|
||||
topic string
|
||||
rcClient *rockscache.Client
|
||||
keys []string
|
||||
maxRetryTimes int
|
||||
retryInterval time.Duration
|
||||
redisClient redis.UniversalClient
|
||||
}
|
||||
|
||||
func (m *metaCacheRedis) Copy() metaCache {
|
||||
var keys []string
|
||||
if len(m.keys) > 0 {
|
||||
keys = make([]string, 0, len(m.keys)*2)
|
||||
keys = append(keys, m.keys...)
|
||||
}
|
||||
return &metaCacheRedis{
|
||||
topic: m.topic,
|
||||
rcClient: m.rcClient,
|
||||
keys: keys,
|
||||
maxRetryTimes: m.maxRetryTimes,
|
||||
retryInterval: m.retryInterval,
|
||||
redisClient: redisClient,
|
||||
}
|
||||
}
|
||||
|
||||
func (m *metaCacheRedis) SetTopic(topic string) {
|
||||
m.topic = topic
|
||||
}
|
||||
|
||||
func (m *metaCacheRedis) SetRawRedisClient(cli redis.UniversalClient) {
|
||||
m.redisClient = cli
|
||||
}
|
||||
|
||||
func (m *metaCacheRedis) ExecDel(ctx context.Context, distinct ...bool) error {
|
||||
@@ -92,7 +62,7 @@ func (m *metaCacheRedis) ExecDel(ctx context.Context, distinct ...bool) error {
|
||||
m.keys = utils.Distinct(m.keys)
|
||||
}
|
||||
if len(m.keys) > 0 {
|
||||
log.ZDebug(ctx, "delete cache", "topic", m.topic, "keys", m.keys)
|
||||
log.ZDebug(ctx, "delete cache", "keys", m.keys)
|
||||
for _, key := range m.keys {
|
||||
for i := 0; i < m.maxRetryTimes; i++ {
|
||||
if err := m.rcClient.TagAsDeleted(key); err != nil {
|
||||
@@ -102,18 +72,31 @@ func (m *metaCacheRedis) ExecDel(ctx context.Context, distinct ...bool) error {
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
if pk := getPublishKey(m.topic, m.keys); len(pk) > 0 {
|
||||
data, err := json.Marshal(pk)
|
||||
if err != nil {
|
||||
log.ZError(ctx, "keys json marshal failed", err, "topic", m.topic, "keys", pk)
|
||||
} else {
|
||||
if err := m.redisClient.Publish(ctx, m.topic, string(data)).Err(); err != nil {
|
||||
log.ZError(ctx, "redis publish cache delete error", err, "topic", m.topic, "keys", pk)
|
||||
}
|
||||
}
|
||||
|
||||
//retryTimes := 0
|
||||
//for {
|
||||
// m.rcClient.TagAsDeleted()
|
||||
// if err := m.rcClient.TagAsDeletedBatch2(ctx, []string{key}); err != nil {
|
||||
// if retryTimes >= m.maxRetryTimes {
|
||||
// err = errs.ErrInternalServer.Wrap(
|
||||
// fmt.Sprintf(
|
||||
// "delete cache error: %v, keys: %v, retry times %d, please check redis server",
|
||||
// err,
|
||||
// key,
|
||||
// retryTimes,
|
||||
// ),
|
||||
// )
|
||||
// log.ZWarn(ctx, "delete cache failed, please handle keys", err, "keys", key)
|
||||
// return err
|
||||
// }
|
||||
// retryTimes++
|
||||
// } else {
|
||||
// break
|
||||
// }
|
||||
//}
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
24
pkg/common/db/cache/msg.go
vendored
24
pkg/common/db/cache/msg.go
vendored
@@ -17,7 +17,6 @@ package cache
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"github.com/dtm-labs/rockscache"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
@@ -45,12 +44,12 @@ const (
|
||||
conversationUserMinSeq = "CON_USER_MIN_SEQ:"
|
||||
hasReadSeq = "HAS_READ_SEQ:"
|
||||
|
||||
//appleDeviceToken = "DEVICE_TOKEN"
|
||||
getuiToken = "GETUI_TOKEN"
|
||||
getuiTaskID = "GETUI_TASK_ID"
|
||||
//signalCache = "SIGNAL_CACHE:"
|
||||
//signalListCache = "SIGNAL_LIST_CACHE:"
|
||||
FCM_TOKEN = "FCM_TOKEN:"
|
||||
appleDeviceToken = "DEVICE_TOKEN"
|
||||
getuiToken = "GETUI_TOKEN"
|
||||
getuiTaskID = "GETUI_TASK_ID"
|
||||
signalCache = "SIGNAL_CACHE:"
|
||||
signalListCache = "SIGNAL_LIST_CACHE:"
|
||||
FCM_TOKEN = "FCM_TOKEN:"
|
||||
|
||||
messageCache = "MESSAGE_CACHE:"
|
||||
messageDelUserList = "MESSAGE_DEL_USER_LIST:"
|
||||
@@ -129,8 +128,7 @@ type MsgModel interface {
|
||||
}
|
||||
|
||||
func NewMsgCacheModel(client redis.UniversalClient) MsgModel {
|
||||
rcClient := rockscache.NewClient(client, rockscache.NewDefaultOptions())
|
||||
return &msgCache{metaCache: NewMetaCacheRedis(rcClient), rdb: client}
|
||||
return &msgCache{rdb: client}
|
||||
}
|
||||
|
||||
type msgCache struct {
|
||||
@@ -150,10 +148,6 @@ func (c *msgCache) getHasReadSeqKey(conversationID string, userID string) string
|
||||
return hasReadSeq + userID + ":" + conversationID
|
||||
}
|
||||
|
||||
func (c *msgCache) getConversationUserMinSeqKey(conversationID, userID string) string {
|
||||
return conversationUserMinSeq + conversationID + "u:" + userID
|
||||
}
|
||||
|
||||
func (c *msgCache) setSeq(ctx context.Context, conversationID string, seq int64, getkey func(conversationID string) string) error {
|
||||
return utils.Wrap1(c.rdb.Set(ctx, getkey(conversationID), seq, 0).Err())
|
||||
}
|
||||
@@ -215,6 +209,10 @@ func (c *msgCache) GetMinSeq(ctx context.Context, conversationID string) (int64,
|
||||
return c.getSeq(ctx, conversationID, c.getMinSeqKey)
|
||||
}
|
||||
|
||||
func (c *msgCache) getConversationUserMinSeqKey(conversationID, userID string) string {
|
||||
return conversationUserMinSeq + conversationID + "u:" + userID
|
||||
}
|
||||
|
||||
func (c *msgCache) GetConversationUserMinSeq(ctx context.Context, conversationID string, userID string) (int64, error) {
|
||||
return utils.Wrap2(c.rdb.Get(ctx, c.getConversationUserMinSeqKey(conversationID, userID)).Int64())
|
||||
}
|
||||
|
||||
21
pkg/common/db/cache/user.go
vendored
21
pkg/common/db/cache/user.go
vendored
@@ -18,8 +18,6 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/cachekey"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||
"hash/crc32"
|
||||
"strconv"
|
||||
"time"
|
||||
@@ -38,8 +36,8 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
userExpireTime = time.Second * 60 * 60 * 12
|
||||
//userInfoKey = "USER_INFO:"
|
||||
userExpireTime = time.Second * 60 * 60 * 12
|
||||
userInfoKey = "USER_INFO:"
|
||||
userGlobalRecvMsgOptKey = "USER_GLOBAL_RECV_MSG_OPT_KEY:"
|
||||
olineStatusKey = "ONLINE_STATUS:"
|
||||
userOlineStatusExpireTime = time.Second * 60 * 60 * 24
|
||||
@@ -74,11 +72,7 @@ func NewUserCacheRedis(
|
||||
options rockscache.Options,
|
||||
) UserCache {
|
||||
rcClient := rockscache.NewClient(rdb, options)
|
||||
mc := NewMetaCacheRedis(rcClient)
|
||||
u := config.Config.LocalCache.User
|
||||
log.ZDebug(context.Background(), "user local cache init", "Topic", u.Topic, "SlotNum", u.SlotNum, "SlotSize", u.SlotSize, "enable", u.Enable())
|
||||
mc.SetTopic(u.Topic)
|
||||
mc.SetRawRedisClient(rdb)
|
||||
|
||||
return &UserCacheRedis{
|
||||
rdb: rdb,
|
||||
metaCache: NewMetaCacheRedis(rcClient),
|
||||
@@ -91,7 +85,7 @@ func NewUserCacheRedis(
|
||||
func (u *UserCacheRedis) NewCache() UserCache {
|
||||
return &UserCacheRedis{
|
||||
rdb: u.rdb,
|
||||
metaCache: u.Copy(),
|
||||
metaCache: NewMetaCacheRedis(u.rcClient, u.metaCache.GetPreDelKeys()...),
|
||||
userDB: u.userDB,
|
||||
expireTime: u.expireTime,
|
||||
rcClient: u.rcClient,
|
||||
@@ -99,17 +93,18 @@ func (u *UserCacheRedis) NewCache() UserCache {
|
||||
}
|
||||
|
||||
func (u *UserCacheRedis) getUserInfoKey(userID string) string {
|
||||
return cachekey.GetUserInfoKey(userID)
|
||||
return userInfoKey + userID
|
||||
}
|
||||
|
||||
func (u *UserCacheRedis) getUserGlobalRecvMsgOptKey(userID string) string {
|
||||
return cachekey.GetUserGlobalRecvMsgOptKey(userID)
|
||||
return userGlobalRecvMsgOptKey + userID
|
||||
}
|
||||
|
||||
func (u *UserCacheRedis) GetUserInfo(ctx context.Context, userID string) (userInfo *relationtb.UserModel, err error) {
|
||||
return getCache(ctx, u.rcClient, u.getUserInfoKey(userID), u.expireTime, func(ctx context.Context) (*relationtb.UserModel, error) {
|
||||
return u.userDB.Take(ctx, userID)
|
||||
})
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
func (u *UserCacheRedis) GetUsersInfo(ctx context.Context, userIDs []string) ([]*relationtb.UserModel, error) {
|
||||
|
||||
@@ -142,12 +142,12 @@ func (u *userDatabase) Find(ctx context.Context, userIDs []string) (users []*rel
|
||||
return u.cache.GetUsersInfo(ctx, userIDs)
|
||||
}
|
||||
|
||||
// Find userInfo By Nickname.
|
||||
// Find userInfo By Nickname
|
||||
func (u *userDatabase) FindByNickname(ctx context.Context, nickname string) (users []*relation.UserModel, err error) {
|
||||
return u.userDB.TakeByNickname(ctx, nickname)
|
||||
}
|
||||
|
||||
// Find notificationAccouts.
|
||||
// Find notificationAccouts
|
||||
func (u *userDatabase) FindNotification(ctx context.Context, level int64) (users []*relation.UserModel, err error) {
|
||||
return u.userDB.TakeNotification(ctx, level)
|
||||
}
|
||||
@@ -190,14 +190,7 @@ func (u *userDatabase) Page(ctx context.Context, pagination pagination.Paginatio
|
||||
func (u *userDatabase) PageFindUser(ctx context.Context, level1 int64, level2 int64, pagination pagination.Pagination) (count int64, users []*relation.UserModel, err error) {
|
||||
return u.userDB.PageFindUser(ctx, level1, level2, pagination)
|
||||
}
|
||||
|
||||
func (u *userDatabase) PageFindUserWithKeyword(
|
||||
ctx context.Context,
|
||||
level1 int64,
|
||||
level2 int64,
|
||||
userID, nickName string,
|
||||
pagination pagination.Pagination,
|
||||
) (count int64, users []*relation.UserModel, err error) {
|
||||
func (u *userDatabase) PageFindUserWithKeyword(ctx context.Context, level1 int64, level2 int64, userID, nickName string, pagination pagination.Pagination) (count int64, users []*relation.UserModel, err error) {
|
||||
return u.userDB.PageFindUserWithKeyword(ctx, level1, level2, userID, nickName, pagination)
|
||||
}
|
||||
|
||||
|
||||
87
pkg/common/db/localcache/conversation.go
Normal file
87
pkg/common/db/localcache/conversation.go
Normal file
@@ -0,0 +1,87 @@
|
||||
// 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 localcache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
|
||||
"github.com/OpenIMSDK/protocol/conversation"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
|
||||
)
|
||||
|
||||
type ConversationLocalCache struct {
|
||||
lock sync.Mutex
|
||||
superGroupRecvMsgNotNotifyUserIDs map[string]Hash
|
||||
conversationIDs map[string]Hash
|
||||
client *rpcclient.ConversationRpcClient
|
||||
}
|
||||
|
||||
type Hash struct {
|
||||
hash uint64
|
||||
ids []string
|
||||
}
|
||||
|
||||
func NewConversationLocalCache(client *rpcclient.ConversationRpcClient) *ConversationLocalCache {
|
||||
return &ConversationLocalCache{
|
||||
superGroupRecvMsgNotNotifyUserIDs: make(map[string]Hash),
|
||||
conversationIDs: make(map[string]Hash),
|
||||
client: client,
|
||||
}
|
||||
}
|
||||
|
||||
func (g *ConversationLocalCache) GetRecvMsgNotNotifyUserIDs(ctx context.Context, groupID string) ([]string, error) {
|
||||
resp, err := g.client.Client.GetRecvMsgNotNotifyUserIDs(ctx, &conversation.GetRecvMsgNotNotifyUserIDsReq{
|
||||
GroupID: groupID,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return resp.UserIDs, nil
|
||||
}
|
||||
|
||||
func (g *ConversationLocalCache) GetConversationIDs(ctx context.Context, userID string) ([]string, error) {
|
||||
resp, err := g.client.Client.GetUserConversationIDsHash(ctx, &conversation.GetUserConversationIDsHashReq{
|
||||
OwnerUserID: userID,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
g.lock.Lock()
|
||||
hash, ok := g.conversationIDs[userID]
|
||||
g.lock.Unlock()
|
||||
|
||||
if !ok || hash.hash != resp.Hash {
|
||||
conversationIDsResp, err := g.client.Client.GetConversationIDs(ctx, &conversation.GetConversationIDsReq{
|
||||
UserID: userID,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
g.lock.Lock()
|
||||
defer g.lock.Unlock()
|
||||
g.conversationIDs[userID] = Hash{
|
||||
hash: resp.Hash,
|
||||
ids: conversationIDsResp.ConversationIDs,
|
||||
}
|
||||
|
||||
return conversationIDsResp.ConversationIDs, nil
|
||||
}
|
||||
|
||||
return hash.ids, nil
|
||||
}
|
||||
15
pkg/common/db/localcache/doc.go
Normal file
15
pkg/common/db/localcache/doc.go
Normal file
@@ -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 localcache // import "github.com/openimsdk/open-im-server/v3/pkg/common/db/localcache"
|
||||
78
pkg/common/db/localcache/group.go
Normal file
78
pkg/common/db/localcache/group.go
Normal file
@@ -0,0 +1,78 @@
|
||||
// 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 localcache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
|
||||
"github.com/OpenIMSDK/protocol/group"
|
||||
"github.com/OpenIMSDK/tools/errs"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
|
||||
)
|
||||
|
||||
type GroupLocalCache struct {
|
||||
lock sync.Mutex
|
||||
cache map[string]GroupMemberIDsHash
|
||||
client *rpcclient.GroupRpcClient
|
||||
}
|
||||
|
||||
type GroupMemberIDsHash struct {
|
||||
memberListHash uint64
|
||||
userIDs []string
|
||||
}
|
||||
|
||||
func NewGroupLocalCache(client *rpcclient.GroupRpcClient) *GroupLocalCache {
|
||||
return &GroupLocalCache{
|
||||
cache: make(map[string]GroupMemberIDsHash, 0),
|
||||
client: client,
|
||||
}
|
||||
}
|
||||
|
||||
func (g *GroupLocalCache) GetGroupMemberIDs(ctx context.Context, groupID string) ([]string, error) {
|
||||
resp, err := g.client.Client.GetGroupAbstractInfo(ctx, &group.GetGroupAbstractInfoReq{
|
||||
GroupIDs: []string{groupID},
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(resp.GroupAbstractInfos) < 1 {
|
||||
return nil, errs.ErrGroupIDNotFound
|
||||
}
|
||||
|
||||
g.lock.Lock()
|
||||
localHashInfo, ok := g.cache[groupID]
|
||||
if ok && localHashInfo.memberListHash == resp.GroupAbstractInfos[0].GroupMemberListHash {
|
||||
g.lock.Unlock()
|
||||
return localHashInfo.userIDs, nil
|
||||
}
|
||||
g.lock.Unlock()
|
||||
|
||||
groupMembersResp, err := g.client.Client.GetGroupMemberUserIDs(ctx, &group.GetGroupMemberUserIDsReq{
|
||||
GroupID: groupID,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
g.lock.Lock()
|
||||
defer g.lock.Unlock()
|
||||
g.cache[groupID] = GroupMemberIDsHash{
|
||||
memberListHash: resp.GroupAbstractInfos[0].GroupMemberListHash,
|
||||
userIDs: groupMembersResp.UserIDs,
|
||||
}
|
||||
return g.cache[groupID].userIDs, nil
|
||||
}
|
||||
15
pkg/common/db/localcache/meta_local_cache.go
Normal file
15
pkg/common/db/localcache/meta_local_cache.go
Normal file
@@ -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 localcache
|
||||
@@ -16,7 +16,6 @@ package mgo
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/OpenIMSDK/tools/mgoutil"
|
||||
"github.com/OpenIMSDK/tools/pagination"
|
||||
"go.mongodb.org/mongo-driver/mongo/options"
|
||||
|
||||
@@ -16,11 +16,10 @@ package mgo
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/OpenIMSDK/protocol/user"
|
||||
"github.com/OpenIMSDK/tools/errs"
|
||||
"go.mongodb.org/mongo-driver/bson/primitive"
|
||||
"time"
|
||||
|
||||
"github.com/OpenIMSDK/tools/mgoutil"
|
||||
"github.com/OpenIMSDK/tools/pagination"
|
||||
@@ -90,15 +89,7 @@ func (u *UserMgo) PageFindUser(ctx context.Context, level1 int64, level2 int64,
|
||||
|
||||
return mgoutil.FindPage[*relation.UserModel](ctx, u.coll, query, pagination)
|
||||
}
|
||||
|
||||
func (u *UserMgo) PageFindUserWithKeyword(
|
||||
ctx context.Context,
|
||||
level1 int64,
|
||||
level2 int64,
|
||||
userID string,
|
||||
nickName string,
|
||||
pagination pagination.Pagination,
|
||||
) (count int64, users []*relation.UserModel, err error) {
|
||||
func (u *UserMgo) PageFindUserWithKeyword(ctx context.Context, level1 int64, level2 int64, userID string, nickName string, pagination pagination.Pagination) (count int64, users []*relation.UserModel, err error) {
|
||||
// Initialize the base query with level conditions
|
||||
query := bson.M{
|
||||
"$and": []bson.M{
|
||||
|
||||
@@ -36,9 +36,9 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
minPartSize int64 = 1024 * 1024 * 1 // 1MB
|
||||
maxPartSize int64 = 1024 * 1024 * 1024 * 5 // 5GB
|
||||
maxNumSize int64 = 1000
|
||||
minPartSize = 1024 * 1024 * 1 // 1MB
|
||||
maxPartSize = 1024 * 1024 * 1024 * 5 // 5GB
|
||||
maxNumSize = 1000
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -133,7 +133,7 @@ func (c *Cos) PartSize(ctx context.Context, size int64) (int64, error) {
|
||||
return 0, errors.New("size must be greater than 0")
|
||||
}
|
||||
if size > maxPartSize*maxNumSize {
|
||||
return 0, fmt.Errorf("COS size must be less than the maximum allowed limit")
|
||||
return 0, fmt.Errorf("size must be less than %db", maxPartSize*maxNumSize)
|
||||
}
|
||||
if size <= minPartSize*maxNumSize {
|
||||
return minPartSize, nil
|
||||
|
||||
@@ -45,9 +45,9 @@ const (
|
||||
)
|
||||
|
||||
const (
|
||||
minPartSize int64 = 1024 * 1024 * 5 // 1MB
|
||||
maxPartSize int64 = 1024 * 1024 * 1024 * 5 // 5GB
|
||||
maxNumSize int64 = 10000
|
||||
minPartSize = 1024 * 1024 * 5 // 1MB
|
||||
maxPartSize = 1024 * 1024 * 1024 * 5 // 5GB
|
||||
maxNumSize = 10000
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -240,7 +240,7 @@ func (m *Minio) PartSize(ctx context.Context, size int64) (int64, error) {
|
||||
return 0, errors.New("size must be greater than 0")
|
||||
}
|
||||
if size > maxPartSize*maxNumSize {
|
||||
return 0, fmt.Errorf("MINIO size must be less than the maximum allowed limit")
|
||||
return 0, fmt.Errorf("size must be less than %db", maxPartSize*maxNumSize)
|
||||
}
|
||||
if size <= minPartSize*maxNumSize {
|
||||
return minPartSize, nil
|
||||
|
||||
@@ -37,9 +37,9 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
minPartSize int64 = 1024 * 1024 * 1 // 1MB
|
||||
maxPartSize int64 = 1024 * 1024 * 1024 * 5 // 5GB
|
||||
maxNumSize int64 = 10000
|
||||
minPartSize = 1024 * 1024 * 1 // 1MB
|
||||
maxPartSize = 1024 * 1024 * 1024 * 5 // 5GB
|
||||
maxNumSize = 10000
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -141,7 +141,7 @@ func (o *OSS) PartSize(ctx context.Context, size int64) (int64, error) {
|
||||
return 0, errors.New("size must be greater than 0")
|
||||
}
|
||||
if size > maxPartSize*maxNumSize {
|
||||
return 0, fmt.Errorf("OSS size must be less than the maximum allowed limit")
|
||||
return 0, fmt.Errorf("size must be less than %db", maxPartSize*maxNumSize)
|
||||
}
|
||||
if size <= minPartSize*maxNumSize {
|
||||
return minPartSize, nil
|
||||
|
||||
@@ -24,7 +24,7 @@ import (
|
||||
type PartLimit struct {
|
||||
MinPartSize int64 `json:"minPartSize"`
|
||||
MaxPartSize int64 `json:"maxPartSize"`
|
||||
MaxNumSize int64 `json:"maxNumSize"`
|
||||
MaxNumSize int `json:"maxNumSize"`
|
||||
}
|
||||
|
||||
type InitiateMultipartUploadResult struct {
|
||||
|
||||
@@ -16,9 +16,8 @@ package relation
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/OpenIMSDK/protocol/user"
|
||||
"time"
|
||||
|
||||
"github.com/OpenIMSDK/tools/pagination"
|
||||
)
|
||||
|
||||
@@ -1,81 +0,0 @@
|
||||
package direct
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/OpenIMSDK/tools/log"
|
||||
"google.golang.org/grpc/resolver"
|
||||
"math/rand"
|
||||
"strings"
|
||||
)
|
||||
|
||||
const (
|
||||
slashSeparator = "/"
|
||||
// EndpointSepChar is the separator char in endpoints.
|
||||
EndpointSepChar = ','
|
||||
|
||||
subsetSize = 32
|
||||
scheme = "direct"
|
||||
)
|
||||
|
||||
type ResolverDirect struct {
|
||||
}
|
||||
|
||||
func NewResolverDirect() *ResolverDirect {
|
||||
return &ResolverDirect{}
|
||||
}
|
||||
|
||||
func (rd *ResolverDirect) Build(target resolver.Target, cc resolver.ClientConn, _ resolver.BuildOptions) (
|
||||
resolver.Resolver, error) {
|
||||
log.ZDebug(context.Background(), "Build", "target", target)
|
||||
endpoints := strings.FieldsFunc(GetEndpoints(target), func(r rune) bool {
|
||||
return r == EndpointSepChar
|
||||
})
|
||||
endpoints = subset(endpoints, subsetSize)
|
||||
addrs := make([]resolver.Address, 0, len(endpoints))
|
||||
|
||||
for _, val := range endpoints {
|
||||
addrs = append(addrs, resolver.Address{
|
||||
Addr: val,
|
||||
})
|
||||
}
|
||||
if err := cc.UpdateState(resolver.State{
|
||||
Addresses: addrs,
|
||||
}); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &nopResolver{cc: cc}, nil
|
||||
}
|
||||
func init() {
|
||||
resolver.Register(&ResolverDirect{})
|
||||
}
|
||||
func (rd *ResolverDirect) Scheme() string {
|
||||
return scheme // return your custom scheme name
|
||||
}
|
||||
|
||||
// GetEndpoints returns the endpoints from the given target.
|
||||
func GetEndpoints(target resolver.Target) string {
|
||||
return strings.Trim(target.URL.Path, slashSeparator)
|
||||
}
|
||||
func subset(set []string, sub int) []string {
|
||||
rand.Shuffle(len(set), func(i, j int) {
|
||||
set[i], set[j] = set[j], set[i]
|
||||
})
|
||||
if len(set) <= sub {
|
||||
return set
|
||||
}
|
||||
|
||||
return set[:sub]
|
||||
}
|
||||
|
||||
type nopResolver struct {
|
||||
cc resolver.ClientConn
|
||||
}
|
||||
|
||||
func (n nopResolver) ResolveNow(options resolver.ResolveNowOptions) {
|
||||
|
||||
}
|
||||
|
||||
func (n nopResolver) Close() {
|
||||
|
||||
}
|
||||
@@ -1,154 +0,0 @@
|
||||
package direct
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"github.com/OpenIMSDK/tools/errs"
|
||||
config2 "github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/credentials/insecure"
|
||||
)
|
||||
|
||||
type ServiceAddresses map[string][]int
|
||||
|
||||
func getServiceAddresses() ServiceAddresses {
|
||||
return ServiceAddresses{
|
||||
config2.Config.RpcRegisterName.OpenImUserName: config2.Config.RpcPort.OpenImUserPort,
|
||||
config2.Config.RpcRegisterName.OpenImFriendName: config2.Config.RpcPort.OpenImFriendPort,
|
||||
config2.Config.RpcRegisterName.OpenImMsgName: config2.Config.RpcPort.OpenImMessagePort,
|
||||
config2.Config.RpcRegisterName.OpenImMessageGatewayName: config2.Config.LongConnSvr.OpenImMessageGatewayPort,
|
||||
config2.Config.RpcRegisterName.OpenImGroupName: config2.Config.RpcPort.OpenImGroupPort,
|
||||
config2.Config.RpcRegisterName.OpenImAuthName: config2.Config.RpcPort.OpenImAuthPort,
|
||||
config2.Config.RpcRegisterName.OpenImPushName: config2.Config.RpcPort.OpenImPushPort,
|
||||
config2.Config.RpcRegisterName.OpenImConversationName: config2.Config.RpcPort.OpenImConversationPort,
|
||||
config2.Config.RpcRegisterName.OpenImThirdName: config2.Config.RpcPort.OpenImThirdPort,
|
||||
}
|
||||
}
|
||||
|
||||
type ConnDirect struct {
|
||||
additionalOpts []grpc.DialOption
|
||||
currentServiceAddress string
|
||||
conns map[string][]*grpc.ClientConn
|
||||
resolverDirect *ResolverDirect
|
||||
}
|
||||
|
||||
func (cd *ConnDirect) GetClientLocalConns() map[string][]*grpc.ClientConn {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (cd *ConnDirect) GetUserIdHashGatewayHost(ctx context.Context, userId string) (string, error) {
|
||||
return "", nil
|
||||
}
|
||||
|
||||
func (cd *ConnDirect) Register(serviceName, host string, port int, opts ...grpc.DialOption) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (cd *ConnDirect) UnRegister() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (cd *ConnDirect) CreateRpcRootNodes(serviceNames []string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (cd *ConnDirect) RegisterConf2Registry(key string, conf []byte) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (cd *ConnDirect) GetConfFromRegistry(key string) ([]byte, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (cd *ConnDirect) Close() {
|
||||
|
||||
}
|
||||
|
||||
func NewConnDirect() (*ConnDirect, error) {
|
||||
return &ConnDirect{
|
||||
conns: make(map[string][]*grpc.ClientConn),
|
||||
resolverDirect: NewResolverDirect(),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (cd *ConnDirect) GetConns(ctx context.Context,
|
||||
serviceName string, opts ...grpc.DialOption) ([]*grpc.ClientConn, error) {
|
||||
|
||||
if conns, exists := cd.conns[serviceName]; exists {
|
||||
return conns, nil
|
||||
}
|
||||
ports := getServiceAddresses()[serviceName]
|
||||
var connections []*grpc.ClientConn
|
||||
for _, port := range ports {
|
||||
conn, err := cd.dialServiceWithoutResolver(ctx, fmt.Sprintf(config2.Config.Rpc.ListenIP+":%d", port), append(cd.additionalOpts, opts...)...)
|
||||
if err != nil {
|
||||
fmt.Printf("connect to port %d failed,serviceName %s, IP %s\n", port, serviceName, config2.Config.Rpc.ListenIP)
|
||||
}
|
||||
connections = append(connections, conn)
|
||||
}
|
||||
|
||||
if len(connections) == 0 {
|
||||
return nil, fmt.Errorf("no connections found for service: %s", serviceName)
|
||||
}
|
||||
return connections, nil
|
||||
}
|
||||
|
||||
func (cd *ConnDirect) GetConn(ctx context.Context, serviceName string, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
|
||||
// Get service addresses
|
||||
addresses := getServiceAddresses()
|
||||
address, ok := addresses[serviceName]
|
||||
if !ok {
|
||||
return nil, errs.Wrap(errors.New("unknown service name"), "serviceName", serviceName)
|
||||
}
|
||||
var result string
|
||||
for _, addr := range address {
|
||||
if result != "" {
|
||||
result = result + "," + fmt.Sprintf(config2.Config.Rpc.ListenIP+":%d", addr)
|
||||
} else {
|
||||
result = fmt.Sprintf(config2.Config.Rpc.ListenIP+":%d", addr)
|
||||
}
|
||||
}
|
||||
// Try to dial a new connection
|
||||
conn, err := cd.dialService(ctx, result, append(cd.additionalOpts, opts...)...)
|
||||
if err != nil {
|
||||
return nil, errs.Wrap(err, "address", result)
|
||||
}
|
||||
|
||||
// Store the new connection
|
||||
cd.conns[serviceName] = append(cd.conns[serviceName], conn)
|
||||
return conn, nil
|
||||
}
|
||||
|
||||
func (cd *ConnDirect) GetSelfConnTarget() string {
|
||||
return cd.currentServiceAddress
|
||||
}
|
||||
|
||||
func (cd *ConnDirect) AddOption(opts ...grpc.DialOption) {
|
||||
cd.additionalOpts = append(cd.additionalOpts, opts...)
|
||||
}
|
||||
|
||||
func (cd *ConnDirect) CloseConn(conn *grpc.ClientConn) {
|
||||
if conn != nil {
|
||||
conn.Close()
|
||||
}
|
||||
}
|
||||
|
||||
func (cd *ConnDirect) dialService(ctx context.Context, address string, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
|
||||
options := append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
|
||||
conn, err := grpc.DialContext(ctx, cd.resolverDirect.Scheme()+":///"+address, options...)
|
||||
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return conn, nil
|
||||
}
|
||||
func (cd *ConnDirect) dialServiceWithoutResolver(ctx context.Context, address string, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
|
||||
options := append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
|
||||
conn, err := grpc.DialContext(ctx, address, options...)
|
||||
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return conn, nil
|
||||
}
|
||||
@@ -16,7 +16,6 @@ package discoveryregister
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister/direct"
|
||||
"os"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister/kubernetes"
|
||||
@@ -37,8 +36,6 @@ func NewDiscoveryRegister(envType string) (discoveryregistry.SvcDiscoveryRegistr
|
||||
return zookeeper.NewZookeeperDiscoveryRegister()
|
||||
case "k8s":
|
||||
return kubernetes.NewK8sDiscoveryRegister()
|
||||
case "direct":
|
||||
return direct.NewConnDirect()
|
||||
default:
|
||||
return nil, errors.New("envType not correct")
|
||||
}
|
||||
|
||||
@@ -40,7 +40,6 @@ func TestNewDiscoveryRegister(t *testing.T) {
|
||||
}{
|
||||
{"zookeeper", false, true},
|
||||
{"k8s", false, true}, // 假设 k8s 配置也已正确设置
|
||||
{"direct", false, true},
|
||||
{"invalid", true, false},
|
||||
}
|
||||
|
||||
|
||||
@@ -18,17 +18,15 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"github.com/stathat/consistent"
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/stathat/consistent"
|
||||
|
||||
"google.golang.org/grpc"
|
||||
|
||||
"github.com/OpenIMSDK/tools/discoveryregistry"
|
||||
"github.com/OpenIMSDK/tools/log"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||
)
|
||||
|
||||
@@ -104,7 +102,7 @@ func getSelfHost(ctx context.Context) string {
|
||||
return host
|
||||
}
|
||||
|
||||
// like openimserver-openim-msggateway-0.openimserver-openim-msggateway-headless.openim-lin.svc.cluster.local:88.
|
||||
// like openimserver-openim-msggateway-0.openimserver-openim-msggateway-headless.openim-lin.svc.cluster.local:88
|
||||
func getMsgGatewayHost(ctx context.Context) []string {
|
||||
port := 88
|
||||
instance := "openimserver"
|
||||
@@ -132,40 +130,20 @@ func getMsgGatewayHost(ctx context.Context) []string {
|
||||
// GetConns returns the gRPC client connections to the specified service.
|
||||
func (cli *K8sDR) GetConns(ctx context.Context, serviceName string, opts ...grpc.DialOption) ([]*grpc.ClientConn, error) {
|
||||
|
||||
// This conditional checks if the serviceName is not the OpenImMessageGatewayName.
|
||||
// It seems to handle a special case for the OpenImMessageGateway.
|
||||
if serviceName != config.Config.RpcRegisterName.OpenImMessageGatewayName {
|
||||
// DialContext creates a client connection to the given target (serviceName) using the specified context.
|
||||
// 'cli.options' are likely default or common options for all connections in this struct.
|
||||
// 'opts...' allows for additional gRPC dial options to be passed and used.
|
||||
conn, err := grpc.DialContext(ctx, serviceName, append(cli.options, opts...)...)
|
||||
|
||||
// The function returns a slice of client connections with the new connection, or an error if occurred.
|
||||
return []*grpc.ClientConn{conn}, err
|
||||
} else {
|
||||
// This block is executed if the serviceName is OpenImMessageGatewayName.
|
||||
// 'ret' will accumulate the connections to return.
|
||||
var ret []*grpc.ClientConn
|
||||
|
||||
// getMsgGatewayHost presumably retrieves hosts for the message gateway service.
|
||||
// The context is passed, likely for cancellation and timeout control.
|
||||
gatewayHosts := getMsgGatewayHost(ctx)
|
||||
|
||||
// Iterating over the retrieved gateway hosts.
|
||||
for _, host := range gatewayHosts {
|
||||
// Establishes a connection to each host.
|
||||
// Again, appending cli.options with any additional opts provided.
|
||||
conn, err := grpc.DialContext(ctx, host, append(cli.options, opts...)...)
|
||||
|
||||
// If there's an error while dialing any host, the function returns immediately with the error.
|
||||
if err != nil {
|
||||
return nil, err
|
||||
} else {
|
||||
// If the connection is successful, it is added to the 'ret' slice.
|
||||
ret = append(ret, conn)
|
||||
}
|
||||
}
|
||||
// After all hosts are processed, the slice of connections is returned.
|
||||
return ret, nil
|
||||
}
|
||||
}
|
||||
@@ -188,7 +166,7 @@ func (cli *K8sDR) CloseConn(conn *grpc.ClientConn) {
|
||||
conn.Close()
|
||||
}
|
||||
|
||||
// do not use this method for call rpc.
|
||||
// do not use this method for call rpc
|
||||
func (cli *K8sDR) GetClientLocalConns() map[string][]*grpc.ClientConn {
|
||||
fmt.Println("should not call this function!!!!!!!!!!!!!!!!!!!!!!!!!")
|
||||
return nil
|
||||
|
||||
@@ -1,16 +0,0 @@
|
||||
package redispubsub
|
||||
|
||||
import "github.com/redis/go-redis/v9"
|
||||
|
||||
type Publisher struct {
|
||||
client redis.UniversalClient
|
||||
channel string
|
||||
}
|
||||
|
||||
func NewPublisher(client redis.UniversalClient, channel string) *Publisher {
|
||||
return &Publisher{client: client, channel: channel}
|
||||
}
|
||||
|
||||
func (p *Publisher) Publish(message string) error {
|
||||
return p.client.Publish(ctx, p.channel, message).Err()
|
||||
}
|
||||
@@ -1,34 +0,0 @@
|
||||
package redispubsub
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
var ctx = context.Background()
|
||||
|
||||
type Subscriber struct {
|
||||
client redis.UniversalClient
|
||||
channel string
|
||||
}
|
||||
|
||||
func NewSubscriber(client redis.UniversalClient, channel string) *Subscriber {
|
||||
return &Subscriber{client: client, channel: channel}
|
||||
}
|
||||
|
||||
func (s *Subscriber) OnMessage(ctx context.Context, callback func(string)) error {
|
||||
messageChannel := s.client.Subscribe(ctx, s.channel).Channel()
|
||||
|
||||
go func() {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case msg := <-messageChannel:
|
||||
callback(msg.Payload)
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -71,7 +71,7 @@ func Start(
|
||||
}
|
||||
|
||||
defer client.Close()
|
||||
client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")))
|
||||
client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()))
|
||||
registerIP, err := network.GetRpcRegisterIP(config.Config.Rpc.RegisterIP)
|
||||
if err != nil {
|
||||
return err
|
||||
|
||||
@@ -1,112 +0,0 @@
|
||||
package localcache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/openimsdk/localcache/link"
|
||||
"github.com/openimsdk/localcache/lru"
|
||||
"hash/fnv"
|
||||
"unsafe"
|
||||
)
|
||||
|
||||
type Cache[V any] interface {
|
||||
Get(ctx context.Context, key string, fetch func(ctx context.Context) (V, error)) (V, error)
|
||||
GetLink(ctx context.Context, key string, fetch func(ctx context.Context) (V, error), link ...string) (V, error)
|
||||
Del(ctx context.Context, key ...string)
|
||||
DelLocal(ctx context.Context, key ...string)
|
||||
Stop()
|
||||
}
|
||||
|
||||
func New[V any](opts ...Option) Cache[V] {
|
||||
opt := defaultOption()
|
||||
for _, o := range opts {
|
||||
o(opt)
|
||||
}
|
||||
|
||||
c := cache[V]{opt: opt}
|
||||
if opt.localSlotNum > 0 && opt.localSlotSize > 0 {
|
||||
createSimpleLRU := func() lru.LRU[string, V] {
|
||||
if opt.expirationEvict {
|
||||
return lru.NewExpirationLRU[string, V](opt.localSlotSize, opt.localSuccessTTL, opt.localFailedTTL, opt.target, c.onEvict)
|
||||
} else {
|
||||
return lru.NewLayLRU[string, V](opt.localSlotSize, opt.localSuccessTTL, opt.localFailedTTL, opt.target, c.onEvict)
|
||||
}
|
||||
}
|
||||
if opt.localSlotNum == 1 {
|
||||
c.local = createSimpleLRU()
|
||||
} else {
|
||||
c.local = lru.NewSlotLRU[string, V](opt.localSlotNum, func(key string) uint64 {
|
||||
h := fnv.New64a()
|
||||
h.Write(*(*[]byte)(unsafe.Pointer(&key)))
|
||||
return h.Sum64()
|
||||
}, createSimpleLRU)
|
||||
}
|
||||
if opt.linkSlotNum > 0 {
|
||||
c.link = link.New(opt.linkSlotNum)
|
||||
}
|
||||
}
|
||||
return &c
|
||||
}
|
||||
|
||||
type cache[V any] struct {
|
||||
opt *option
|
||||
link link.Link
|
||||
local lru.LRU[string, V]
|
||||
}
|
||||
|
||||
func (c *cache[V]) onEvict(key string, value V) {
|
||||
if c.link != nil {
|
||||
lks := c.link.Del(key)
|
||||
for k := range lks {
|
||||
if key != k { // prevent deadlock
|
||||
c.local.Del(k)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *cache[V]) del(key ...string) {
|
||||
if c.local == nil {
|
||||
return
|
||||
}
|
||||
for _, k := range key {
|
||||
c.local.Del(k)
|
||||
if c.link != nil {
|
||||
lks := c.link.Del(k)
|
||||
for k := range lks {
|
||||
c.local.Del(k)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *cache[V]) Get(ctx context.Context, key string, fetch func(ctx context.Context) (V, error)) (V, error) {
|
||||
return c.GetLink(ctx, key, fetch)
|
||||
}
|
||||
|
||||
func (c *cache[V]) GetLink(ctx context.Context, key string, fetch func(ctx context.Context) (V, error), link ...string) (V, error) {
|
||||
if c.local != nil {
|
||||
return c.local.Get(key, func() (V, error) {
|
||||
if len(link) > 0 {
|
||||
c.link.Link(key, link...)
|
||||
}
|
||||
return fetch(ctx)
|
||||
})
|
||||
} else {
|
||||
return fetch(ctx)
|
||||
}
|
||||
}
|
||||
|
||||
func (c *cache[V]) Del(ctx context.Context, key ...string) {
|
||||
for _, fn := range c.opt.delFn {
|
||||
fn(ctx, key...)
|
||||
}
|
||||
c.del(key...)
|
||||
}
|
||||
|
||||
func (c *cache[V]) DelLocal(ctx context.Context, key ...string) {
|
||||
c.del(key...)
|
||||
}
|
||||
|
||||
func (c *cache[V]) Stop() {
|
||||
c.local.Stop()
|
||||
}
|
||||
@@ -1,79 +0,0 @@
|
||||
package localcache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"math/rand"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestName(t *testing.T) {
|
||||
c := New[string](WithExpirationEvict())
|
||||
//c := New[string]()
|
||||
ctx := context.Background()
|
||||
|
||||
const (
|
||||
num = 10000
|
||||
tNum = 10000
|
||||
kNum = 100000
|
||||
pNum = 100
|
||||
)
|
||||
|
||||
getKey := func(v uint64) string {
|
||||
return fmt.Sprintf("key_%d", v%kNum)
|
||||
}
|
||||
|
||||
start := time.Now()
|
||||
t.Log("start", start)
|
||||
|
||||
var (
|
||||
get atomic.Int64
|
||||
del atomic.Int64
|
||||
)
|
||||
|
||||
incrGet := func() {
|
||||
if v := get.Add(1); v%pNum == 0 {
|
||||
//t.Log("#get count", v/pNum)
|
||||
}
|
||||
}
|
||||
incrDel := func() {
|
||||
if v := del.Add(1); v%pNum == 0 {
|
||||
//t.Log("@del count", v/pNum)
|
||||
}
|
||||
}
|
||||
|
||||
var wg sync.WaitGroup
|
||||
|
||||
for i := 0; i < tNum; i++ {
|
||||
wg.Add(2)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for i := 0; i < num; i++ {
|
||||
c.Get(ctx, getKey(rand.Uint64()), func(ctx context.Context) (string, error) {
|
||||
return fmt.Sprintf("index_%d", i), nil
|
||||
})
|
||||
incrGet()
|
||||
}
|
||||
}()
|
||||
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
time.Sleep(time.Second / 10)
|
||||
for i := 0; i < num; i++ {
|
||||
c.Del(ctx, getKey(rand.Uint64()))
|
||||
incrDel()
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
end := time.Now()
|
||||
t.Log("end", end)
|
||||
t.Log("time", end.Sub(start))
|
||||
t.Log("get", get.Load())
|
||||
t.Log("del", del.Load())
|
||||
// 137.35s
|
||||
}
|
||||
@@ -1,5 +0,0 @@
|
||||
module github.com/openimsdk/localcache
|
||||
|
||||
go 1.19
|
||||
|
||||
require github.com/hashicorp/golang-lru/v2 v2.0.7
|
||||
@@ -1,109 +0,0 @@
|
||||
package link
|
||||
|
||||
import (
|
||||
"hash/fnv"
|
||||
"sync"
|
||||
"unsafe"
|
||||
)
|
||||
|
||||
type Link interface {
|
||||
Link(key string, link ...string)
|
||||
Del(key string) map[string]struct{}
|
||||
}
|
||||
|
||||
func newLinkKey() *linkKey {
|
||||
return &linkKey{
|
||||
data: make(map[string]map[string]struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
type linkKey struct {
|
||||
lock sync.Mutex
|
||||
data map[string]map[string]struct{}
|
||||
}
|
||||
|
||||
func (x *linkKey) link(key string, link ...string) {
|
||||
x.lock.Lock()
|
||||
defer x.lock.Unlock()
|
||||
v, ok := x.data[key]
|
||||
if !ok {
|
||||
v = make(map[string]struct{})
|
||||
x.data[key] = v
|
||||
}
|
||||
for _, k := range link {
|
||||
v[k] = struct{}{}
|
||||
}
|
||||
}
|
||||
|
||||
func (x *linkKey) del(key string) map[string]struct{} {
|
||||
x.lock.Lock()
|
||||
defer x.lock.Unlock()
|
||||
ks, ok := x.data[key]
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
delete(x.data, key)
|
||||
return ks
|
||||
}
|
||||
|
||||
func New(n int) Link {
|
||||
if n <= 0 {
|
||||
panic("must be greater than 0")
|
||||
}
|
||||
slots := make([]*linkKey, n)
|
||||
for i := 0; i < len(slots); i++ {
|
||||
slots[i] = newLinkKey()
|
||||
}
|
||||
return &slot{
|
||||
n: uint64(n),
|
||||
slots: slots,
|
||||
}
|
||||
}
|
||||
|
||||
type slot struct {
|
||||
n uint64
|
||||
slots []*linkKey
|
||||
}
|
||||
|
||||
func (x *slot) index(s string) uint64 {
|
||||
h := fnv.New64a()
|
||||
_, _ = h.Write(*(*[]byte)(unsafe.Pointer(&s)))
|
||||
return h.Sum64() % x.n
|
||||
}
|
||||
|
||||
func (x *slot) Link(key string, link ...string) {
|
||||
if len(link) == 0 {
|
||||
return
|
||||
}
|
||||
mk := key
|
||||
lks := make([]string, len(link))
|
||||
for i, k := range link {
|
||||
lks[i] = k
|
||||
}
|
||||
x.slots[x.index(mk)].link(mk, lks...)
|
||||
for _, lk := range lks {
|
||||
x.slots[x.index(lk)].link(lk, mk)
|
||||
}
|
||||
}
|
||||
|
||||
func (x *slot) Del(key string) map[string]struct{} {
|
||||
return x.delKey(key)
|
||||
}
|
||||
|
||||
func (x *slot) delKey(k string) map[string]struct{} {
|
||||
del := make(map[string]struct{})
|
||||
stack := []string{k}
|
||||
for len(stack) > 0 {
|
||||
curr := stack[len(stack)-1]
|
||||
stack = stack[:len(stack)-1]
|
||||
if _, ok := del[curr]; ok {
|
||||
continue
|
||||
}
|
||||
del[curr] = struct{}{}
|
||||
childKeys := x.slots[x.index(curr)].del(curr)
|
||||
for ck := range childKeys {
|
||||
stack = append(stack, ck)
|
||||
}
|
||||
}
|
||||
return del
|
||||
}
|
||||
@@ -1,20 +0,0 @@
|
||||
package link
|
||||
|
||||
import (
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestName(t *testing.T) {
|
||||
|
||||
v := New(1)
|
||||
|
||||
//v.Link("a:1", "b:1", "c:1", "d:1")
|
||||
v.Link("a:1", "b:1", "c:1")
|
||||
v.Link("z:1", "b:1")
|
||||
|
||||
//v.DelKey("a:1")
|
||||
v.Del("z:1")
|
||||
|
||||
t.Log(v)
|
||||
|
||||
}
|
||||
@@ -1,20 +0,0 @@
|
||||
package lru
|
||||
|
||||
import "github.com/hashicorp/golang-lru/v2/simplelru"
|
||||
|
||||
type EvictCallback[K comparable, V any] simplelru.EvictCallback[K, V]
|
||||
|
||||
type LRU[K comparable, V any] interface {
|
||||
Get(key K, fetch func() (V, error)) (V, error)
|
||||
Del(key K) bool
|
||||
Stop()
|
||||
}
|
||||
|
||||
type Target interface {
|
||||
IncrGetHit()
|
||||
IncrGetSuccess()
|
||||
IncrGetFailed()
|
||||
|
||||
IncrDelHit()
|
||||
IncrDelNotFound()
|
||||
}
|
||||
@@ -1,78 +0,0 @@
|
||||
package lru
|
||||
|
||||
import (
|
||||
"github.com/hashicorp/golang-lru/v2/expirable"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
func NewExpirationLRU[K comparable, V any](size int, successTTL, failedTTL time.Duration, target Target, onEvict EvictCallback[K, V]) LRU[K, V] {
|
||||
var cb expirable.EvictCallback[K, *expirationLruItem[V]]
|
||||
if onEvict != nil {
|
||||
cb = func(key K, value *expirationLruItem[V]) {
|
||||
onEvict(key, value.value)
|
||||
}
|
||||
}
|
||||
core := expirable.NewLRU[K, *expirationLruItem[V]](size, cb, successTTL)
|
||||
return &ExpirationLRU[K, V]{
|
||||
core: core,
|
||||
successTTL: successTTL,
|
||||
failedTTL: failedTTL,
|
||||
target: target,
|
||||
}
|
||||
}
|
||||
|
||||
type expirationLruItem[V any] struct {
|
||||
lock sync.RWMutex
|
||||
err error
|
||||
value V
|
||||
}
|
||||
|
||||
type ExpirationLRU[K comparable, V any] struct {
|
||||
lock sync.Mutex
|
||||
core *expirable.LRU[K, *expirationLruItem[V]]
|
||||
successTTL time.Duration
|
||||
failedTTL time.Duration
|
||||
target Target
|
||||
}
|
||||
|
||||
func (x *ExpirationLRU[K, V]) Get(key K, fetch func() (V, error)) (V, error) {
|
||||
x.lock.Lock()
|
||||
v, ok := x.core.Get(key)
|
||||
if ok {
|
||||
x.lock.Unlock()
|
||||
x.target.IncrGetSuccess()
|
||||
v.lock.RLock()
|
||||
defer v.lock.RUnlock()
|
||||
return v.value, v.err
|
||||
} else {
|
||||
v = &expirationLruItem[V]{}
|
||||
x.core.Add(key, v)
|
||||
v.lock.Lock()
|
||||
x.lock.Unlock()
|
||||
defer v.lock.Unlock()
|
||||
v.value, v.err = fetch()
|
||||
if v.err == nil {
|
||||
x.target.IncrGetSuccess()
|
||||
} else {
|
||||
x.target.IncrGetFailed()
|
||||
x.core.Remove(key)
|
||||
}
|
||||
return v.value, v.err
|
||||
}
|
||||
}
|
||||
|
||||
func (x *ExpirationLRU[K, V]) Del(key K) bool {
|
||||
x.lock.Lock()
|
||||
ok := x.core.Remove(key)
|
||||
x.lock.Unlock()
|
||||
if ok {
|
||||
x.target.IncrDelHit()
|
||||
} else {
|
||||
x.target.IncrDelNotFound()
|
||||
}
|
||||
return ok
|
||||
}
|
||||
|
||||
func (x *ExpirationLRU[K, V]) Stop() {
|
||||
}
|
||||
Some files were not shown because too many files have changed in this diff Show More
Reference in New Issue
Block a user