Compare commits

..

5 Commits

Author SHA1 Message Date
Xinwei Xiong(cubxxw) 73733caf48 feat: add openim server test about openim web
Signed-off-by: Xinwei Xiong(cubxxw) <3293172751nss@gmail.com>
2023-09-11 17:26:22 +08:00
Xinwei Xiong(cubxxw) ad8b5b484b feat: add openim server test about openim web
Signed-off-by: Xinwei Xiong(cubxxw) <3293172751nss@gmail.com>
2023-09-11 17:14:38 +08:00
Xinwei Xiong(cubxxw) 56144ed76d feat: add openim server test about openim web
Signed-off-by: Xinwei Xiong(cubxxw) <3293172751nss@gmail.com>
2023-09-11 16:31:59 +08:00
Xinwei Xiong(cubxxw-openim) 4ecb2f81bc feat: add options about helm deploy
Signed-off-by: Xinwei Xiong(cubxxw-openim) <3293172751nss@gmail.com>
2023-09-09 19:37:57 +08:00
Xinwei Xiong(cubxxw-openim) 10a0bd09e3 fix: fix docker compose file
Signed-off-by: Xinwei Xiong(cubxxw-openim) <3293172751nss@gmail.com>
2023-09-09 17:20:22 +08:00
56 changed files with 578 additions and 1499 deletions
+2 -110
View File
@@ -28,114 +28,6 @@ env:
GO_VERSION: "1.20"
jobs:
build-dockerhub:
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v3
- name: Set up QEMU
uses: docker/setup-qemu-action@v2
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v2
# docker.io/openim/openim-server:latest
- name: Extract metadata (tags, labels) for Docker
id: meta
uses: docker/metadata-action@v4.6.0
with:
images: openim/openim-server
# generate Docker tags based on the following events/attributes
tags: |
type=schedule
type=ref,event=branch
type=ref,event=pr
type=semver,pattern={{version}}
type=semver,pattern={{major}}.{{minor}}
type=semver,pattern={{major}}
type=sha
- name: Log in to Docker Hub
uses: docker/login-action@v2
with:
username: ${{ secrets.DOCKER_USERNAME }}
password: ${{ secrets.DOCKER_PASSWORD }}
- name: Build and push Docker image
uses: docker/build-push-action@v4
with:
context: .
# linux/ppc64le,linux/s390x
platforms: linux/amd64,linux/arm64
push: ${{ github.event_name != 'pull_request' }}
tags: ${{ steps.meta.outputs.tags }}
labels: ${{ steps.meta.outputs.labels }}
build-aliyun:
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v3
- name: Set up QEMU
uses: docker/setup-qemu-action@v2
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v2
# registry.cn-hangzhou.aliyuncs.com/openimsdk/openim-server:latest
- name: Extract metadata (tags, labels) for Docker
id: meta2
uses: docker/metadata-action@v4.6.0
with:
images: registry.cn-hangzhou.aliyuncs.com/openimsdk/openim-web
- name: Log in to AliYun Docker Hub
uses: docker/login-action@v2
with:
registry: registry.cn-hangzhou.aliyuncs.com
username: ${{ secrets.ALIREGISTRY_USERNAME }}
password: ${{ secrets.ALIREGISTRY_TOKEN }}
- name: Build and push Docker image
uses: docker/build-push-action@v4
with:
context: .
# linux/ppc64le,linux/s390x
platforms: linux/amd64,linux/arm64
push: ${{ github.event_name != 'pull_request' }}
tags: ${{ steps.meta2.outputs.tags }}
labels: ${{ steps.meta2.outputs.labels }}
build-ghcr:
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v3
- name: Set up QEMU
uses: docker/setup-qemu-action@v2
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v2
# ghcr.io/openimsdk/openim-server:latest
- name: Extract metadata (tags, labels) for Docker
id: meta3
uses: docker/metadata-action@v4.6.0
with:
images: ghcr.io/openimsdk/openim-web
- name: Log in to GitHub Container Registry
uses: docker/login-action@v2
with:
registry: ghcr.io
username: ${{ github.repository_owner }}
password: ${{ secrets.GITHUB_TOKEN }}
- name: Build and push Docker image
uses: docker/build-push-action@v4
with:
context: .
# linux/ppc64le,linux/s390x
platforms: linux/amd64,linux/arm64
push: ${{ github.event_name != 'pull_request' }}
tags: ${{ steps.meta3.outputs.tags }}
labels: ${{ steps.meta3.outputs.labels }}
build-openim-web-dockerhub:
runs-on: ubuntu-latest
steps:
@@ -146,12 +38,12 @@ jobs:
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v2
# docker.io/openim/openim-server:latest
# docker.io/openim/openim-web:latest
- name: Extract metadata (tags, labels) for Docker
id: meta
uses: docker/metadata-action@v4.6.0
with:
images: openim/openim-server
images: openim/openim-web
# generate Docker tags based on the following events/attributes
tags: |
type=schedule
+1 -1
View File
@@ -151,7 +151,7 @@ jobs:
version: 2.x
- name: Docker Operations
run: |
curl -o docker-compose.yaml https://raw.githubusercontent.com/OpenIMSDK/openim-docker/main/example/basic-openim-server-dependency.yml
curl -o docker-compose.yml https://raw.githubusercontent.com/OpenIMSDK/openim-docker/main/example/basic-openim-server-dependency.yml
sudo docker compose up -d
sudo sleep 60
+1 -1
View File
@@ -166,7 +166,7 @@ Deploy basic components at the click of a command:
```bash
# install openim dependency
$ git clone https://github.com/openimsdk/open-im-server openim/openim-server && export openim=$(pwd)/openim/openim-server && cd $openim/openim-server && git checkout $OPENIM_VERSION
$ make init && docker compose -f basic-openim-server-dependency.yml up -d && make start && make check
$ make init && docker compose up -d && make start && make check
```
> `make help` to help you see the instructions supported by OpenIM.
+1 -4
View File
@@ -132,7 +132,7 @@ api:
# minio.signEndpoint is minio public network address
object:
enable: "minio"
apiURL: "http://127.0.0.1:10002"
apiURL: "http://http://127.0.0.1:10002"
minio:
bucket: "openim"
endpoint: "http://172.28.0.1:10005"
@@ -140,13 +140,11 @@ object:
secretAccessKey: "openIM123"
sessionToken: ''
signEndpoint: "http://127.0.0.1:10005"
publicRead: false
cos:
bucketURL: https://temp-1252357374.cos.ap-chengdu.myqcloud.com
secretID: ''
secretKey: ''
sessionToken: ''
publicRead: false
oss:
endpoint: "https://oss-cn-chengdu.aliyuncs.com"
bucket: "demo-9999999"
@@ -154,7 +152,6 @@ object:
accessKeyID: ''
accessKeySecret: ''
sessionToken: ''
publicRead: false
###################### RPC Port Configuration ######################
+201
View File
@@ -0,0 +1,201 @@
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "[]"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.
Copyright [yyyy] [name of copyright owner]
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.
+37 -1
View File
@@ -35,4 +35,40 @@ version: 0.1.0
# incremented each time you make changes to the application. Versions are not expected to
# follow Semantic Versioning. They should reflect the version the application is using.
# It is recommended to use it with quotes.
appVersion: "1.16.0"
appVersion: "default"
icon: https://raw.githubusercontent.com/openimsdk/open-im-server/main/assets/openim-logo-gradient.svg
maintainers:
- name: "OpenIM"
url: "https://github.com/openimsdk"
keywords:
- openim
- im
- chat
sources:
- "https://github.com/openimsdk/open-im-server"
- "https://github.com/openimsdk/openim-sdk-core"
- "https://github.com/openimsdk/openim-docker"
dependencies:
- name: mysql
version: 8.0.25
repository: https://charts.bitnami.com/bitnami
- name: redis
version: 14.6.1
repository: https://charts.bitnami.com/bitnami
- name: mongodb
version: 10.1.0
repository: https://charts.bitnami.com/bitnami
- name: kafka
version: 14.1.0
repository: https://charts.bitnami.com/bitnami
- name: minio
version: 8.1.2
repository: https://charts.bitnami.com/bitnami
- name: zookeeper
version: 5.17.0
repository: https://charts.bitnami.com/bitnami
+201
View File
@@ -0,0 +1,201 @@
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "[]"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.
Copyright [yyyy] [name of copyright owner]
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.
+15
View File
@@ -0,0 +1,15 @@
# OpenIM Server Chat
## 目录结构
```bash
openim-server/
Chart.yaml # 包含了chart信息的YAML文件
LICENSE # 包含OpenIM Chart许可证的纯文本文件
README.md # OpenIM 可读的README文件
values.yaml # chart 默认的配置值
charts/ # 包含chart依赖的其他chart
crds/ # 自定义资源的定义
templates/ # 模板目录, 当和values 结合时,可生成有效的Kubernetes manifest文件
templates/NOTES.txt # 包含简要使用说明的纯文本文件
```
+1 -3
View File
@@ -140,13 +140,11 @@ object:
secretAccessKey: "${MINIO_SECRET_KEY}"
sessionToken: ${MINIO_SESSION_TOKEN}
signEndpoint: "${MINIO_SIGN_ENDPOINT}"
publicRead: ${MINIO_PUBLIC_READ}
cos:
bucketURL: ${COS_BUCKET_URL}
secretID: ${COS_SECRET_ID}
secretKey: ${COS_SECRET_KEY}
sessionToken: ${COS_SESSION_TOKEN}
publicRead: ${COS_PUBLIC_READ}
oss:
endpoint: "${OSS_ENDPOINT}"
bucket: "${OSS_BUCKET}"
@@ -154,7 +152,7 @@ object:
accessKeyID: ${OSS_ACCESS_KEY_ID}
accessKeySecret: ${OSS_ACCESS_KEY_SECRET}
sessionToken: ${OSS_SESSION_TOKEN}
publicRead: ${OSS_PUBLIC_READ}
###################### RPC Port Configuration ######################
# RPC service ports
+1 -1
View File
@@ -37,7 +37,7 @@ require github.com/google/uuid v1.3.1
require (
github.com/IBM/sarama v1.41.1
github.com/OpenIMSDK/protocol v0.0.23
github.com/OpenIMSDK/protocol v0.0.18
github.com/OpenIMSDK/tools v0.0.14
github.com/aliyun/aliyun-oss-go-sdk v2.2.9+incompatible
github.com/go-redis/redis v6.15.9+incompatible
+2 -2
View File
@@ -19,8 +19,8 @@ firebase.google.com/go v3.13.0+incompatible/go.mod h1:xlah6XbEyW6tbfSklcfe5FHJIw
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
github.com/IBM/sarama v1.41.1 h1:B4/TdHce/8Ipza+qrLIeNJ9D1AOxZVp/3uDv6H/dp2M=
github.com/IBM/sarama v1.41.1/go.mod h1:JFCPURVskaipJdKRFkiE/OZqQHw7jqliaJmRwXCmSSw=
github.com/OpenIMSDK/protocol v0.0.23 h1:L545aRQez6Ro+AaJB1Z6Mz7ojnDtp41WqASxYveCkcE=
github.com/OpenIMSDK/protocol v0.0.23/go.mod h1:F25dFrwrIx3lkNoiuf6FkCfxuwf8L4Z8UIsdTHP/r0Y=
github.com/OpenIMSDK/protocol v0.0.18 h1:hXukFiDMLZx7s+hDCQePIK9ABiHyNlobNL4MppvOuMY=
github.com/OpenIMSDK/protocol v0.0.18/go.mod h1:F25dFrwrIx3lkNoiuf6FkCfxuwf8L4Z8UIsdTHP/r0Y=
github.com/OpenIMSDK/tools v0.0.14 h1:WLof/+WxyPyRST+QkoTKubYCiV73uCLiL8pgnpH/yKQ=
github.com/OpenIMSDK/tools v0.0.14/go.mod h1:eg+q4A34Qmu73xkY0mt37FHGMCMfC6CtmOnm0kFEGFI=
github.com/QcloudApi/qcloud_sign_golang v0.0.0-20141224014652-e4130a326409/go.mod h1:1pk82RBxDY/JZnPQrtqHlUFfCctgdorsd9M06fMynOM=
-1
View File
@@ -11,5 +11,4 @@ use (
./tools/versionchecker
./tools/yamlfmt
./tools/component
./tools/url2im
)
-4
View File
@@ -44,7 +44,3 @@ func (o *ConversationApi) GetConversations(c *gin.Context) {
func (o *ConversationApi) SetConversations(c *gin.Context) {
a2r.Call(conversation.ConversationClient.SetConversations, o.Client, c)
}
func (o *ConversationApi) GetConversationOfflinePushUserIDs(c *gin.Context) {
a2r.Call(conversation.ConversationClient.GetConversationOfflinePushUserIDs, o.Client, c)
}
+2 -3
View File
@@ -58,7 +58,7 @@ func (m MessageApi) newUserSendMsgReq(c *gin.Context, params *apistruct.SendMsg)
options := make(map[string]bool, 5)
switch params.ContentType {
case constant.Text:
fallthrough
newContent = params.Content["text"].(string)
case constant.Picture:
fallthrough
case constant.Custom:
@@ -100,7 +100,6 @@ func (m MessageApi) newUserSendMsgReq(c *gin.Context, params *apistruct.SendMsg)
ContentType: params.ContentType,
Content: []byte(newContent),
CreateTime: utils.GetCurrentTimestampByMill(),
SendTime: params.SendTime,
Options: options,
OfflinePushInfo: params.OfflinePushInfo,
},
@@ -208,6 +207,7 @@ func (m *MessageApi) SendMessage(c *gin.Context) {
apiresp.GinError(c, errs.ErrArgs.WithDetail(err.Error()).Wrap())
return
}
log.ZInfo(c, "SendMessage", "req", req)
if !authverify.IsAppManagerUid(c) {
apiresp.GinError(c, errs.ErrNoPermission.Wrap("only app manager can send message"))
return
@@ -224,7 +224,6 @@ func (m *MessageApi) SendMessage(c *gin.Context) {
respPb, err := m.Client.SendMsg(c, sendMsgReq)
if err != nil {
status = constant.MsgSendFailed
log.ZError(c, "send message err", err)
apiresp.GinError(c, err)
return
}
-6
View File
@@ -156,11 +156,6 @@ func NewGinRouter(discov discoveryregistry.SvcDiscoveryRegistry, rdb redis.Unive
thirdGroup.POST("/fcm_update_token", t.FcmUpdateToken)
thirdGroup.POST("/set_app_badge", t.SetAppBadge)
logs := thirdGroup.Group("/logs")
logs.POST("/upload", t.UploadLogs)
logs.POST("/delete", t.DeleteLogs)
logs.POST("/search", t.SearchLogs)
objectGroup := r.Group("/object", ParseToken)
objectGroup.POST("/part_limit", t.PartLimit)
@@ -203,7 +198,6 @@ func NewGinRouter(discov discoveryregistry.SvcDiscoveryRegistry, rdb redis.Unive
conversationGroup.POST("/get_conversation", c.GetConversation)
conversationGroup.POST("/get_conversations", c.GetConversations)
conversationGroup.POST("/set_conversations", c.SetConversations)
conversationGroup.POST("/get_conversation_offline_push_user_ids", c.GetConversationOfflinePushUserIDs)
}
statisticsGroup := r.Group("/statistics", ParseToken)
-13
View File
@@ -105,16 +105,3 @@ func (o *ThirdApi) ObjectRedirect(c *gin.Context) {
}
c.Redirect(http.StatusFound, resp.Url)
}
// #################### logs ####################
func (o *ThirdApi) UploadLogs(c *gin.Context) {
a2r.Call(third.ThirdClient.UploadLogs, o.Client, c)
}
func (o *ThirdApi) DeleteLogs(c *gin.Context) {
a2r.Call(third.ThirdClient.DeleteLogs, o.Client, c)
}
func (o *ThirdApi) SearchLogs(c *gin.Context) {
a2r.Call(third.ThirdClient.SearchLogs, o.Client, c)
}
+3 -8
View File
@@ -59,7 +59,7 @@ const (
PongMessage = 10
)
type PingPongHandler func(string) error
type PongHandler func(string) error
type Client struct {
w *sync.Mutex
@@ -107,12 +107,8 @@ func (c *Client) ResetClient(
c.token = token
}
func (c *Client) pingHandler(_ string) error {
func (c *Client) pongHandler(_ string) error {
c.conn.SetReadDeadline(pongWait)
err := c.writePongMsg()
if err != nil {
return err
}
return nil
}
@@ -126,11 +122,10 @@ func (c *Client) readMessage() {
}()
c.conn.SetReadLimit(maxMessageSize)
_ = c.conn.SetReadDeadline(pongWait)
c.conn.SetPingHandler(c.pingHandler)
c.conn.SetPongHandler(c.pongHandler)
for {
messageType, message, returnErr := c.conn.ReadMessage()
if returnErr != nil {
log.ZWarn(c.ctx, "readMessage", returnErr, "messageType", messageType)
c.closedErr = returnErr
return
}
+2 -6
View File
@@ -41,8 +41,7 @@ type LongConn interface {
SetConnNil()
// SetReadLimit sets the maximum size for a message read from the peer.bytes
SetReadLimit(limit int64)
SetPongHandler(handler PingPongHandler)
SetPingHandler(handler PingPongHandler)
SetPongHandler(handler PongHandler)
// GenerateLongConn Check the connection of the current and when it was sent are the same
GenerateLongConn(w http.ResponseWriter, r *http.Request) error
}
@@ -117,12 +116,9 @@ func (d *GWebSocket) SetReadLimit(limit int64) {
d.conn.SetReadLimit(limit)
}
func (d *GWebSocket) SetPongHandler(handler PingPongHandler) {
func (d *GWebSocket) SetPongHandler(handler PongHandler) {
d.conn.SetPongHandler(handler)
}
func (d *GWebSocket) SetPingHandler(handler PingPongHandler) {
d.conn.SetPingHandler(handler)
}
//func (d *GWebSocket) CheckSendConnDiffNow() bool {
// return d.conn == d.sendConn
@@ -60,7 +60,7 @@ func (pc *PersistentConsumerHandler) handleChatWs2Mysql(
log.ZError(ctx, "msg_transfer Unmarshal msg err", err)
return
}
return
log.ZDebug(ctx, "handleChatWs2Mysql", "msg", msgFromMQ.MsgData)
// Control whether to store history messages (mysql)
isPersist := utils.GetSwitchFromOptions(msgFromMQ.MsgData.Options, constant.IsPersistent)
+7 -16
View File
@@ -19,8 +19,6 @@ import (
"encoding/json"
"errors"
"github.com/OpenIMSDK/protocol/conversation"
"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
"github.com/OpenIMSDK/protocol/constant"
@@ -119,6 +117,7 @@ func (p *Pusher) Push2User(ctx context.Context, userIDs []string, msg *sdkws.Msg
if err != nil {
return err
}
break
}
}
}
@@ -235,23 +234,15 @@ func (p *Pusher) Push2SuperGroup(ctx context.Context, groupID string, msg *sdkws
if len(offlinePushUserIDs) > 0 {
needOfflinePushUserIDs = offlinePushUserIDs
}
resp, err := p.conversationRpcClient.Client.GetConversationOfflinePushUserIDs(
ctx,
&conversation.GetConversationOfflinePushUserIDsReq{ConversationID: utils.GenGroupConversationID(groupID), UserIDs: needOfflinePushUserIDs},
)
err = p.offlinePushMsg(ctx, groupID, msg, offlinePushUserIDs)
if err != nil {
log.ZError(ctx, "offlinePushMsg failed", err, "groupID", groupID, "msg", msg)
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
}
_, err := p.GetConnsAndOnlinePush(ctx, msg, utils.IntersectString(needOfflinePushUserIDs, WebAndPcBackgroundUserIDs))
if err != nil {
log.ZError(ctx, "offlinePushMsg failed", err, "groupID", groupID, "msg", msg, "userIDs", utils.IntersectString(needOfflinePushUserIDs, WebAndPcBackgroundUserIDs))
return err
}
}
}
-27
View File
@@ -300,30 +300,3 @@ func (c *conversationServer) GetConversationsByConversationID(
}
return &pbconversation.GetConversationsByConversationIDResp{Conversations: convert.ConversationsDB2Pb(conversations)}, nil
}
func (c *conversationServer) GetConversationOfflinePushUserIDs(
ctx context.Context,
req *pbconversation.GetConversationOfflinePushUserIDsReq,
) (*pbconversation.GetConversationOfflinePushUserIDsResp, error) {
if req.ConversationID == "" {
return nil, errs.ErrArgs.Wrap("conversationID is empty")
}
if len(req.UserIDs) == 0 {
return &pbconversation.GetConversationOfflinePushUserIDsResp{}, nil
}
userIDs, err := c.conversationDatabase.GetConversationNotReceiveMessageUserIDs(ctx, req.ConversationID)
if err != nil {
return nil, err
}
if len(userIDs) == 0 {
return &pbconversation.GetConversationOfflinePushUserIDsResp{UserIDs: req.UserIDs}, nil
}
userIDSet := make(map[string]struct{})
for _, userID := range req.UserIDs {
userIDSet[userID] = struct{}{}
}
for _, userID := range userIDs {
delete(userIDSet, userID)
}
return &pbconversation.GetConversationOfflinePushUserIDsResp{UserIDs: utils.Keys(userIDSet)}, nil
}
-146
View File
@@ -1,146 +0,0 @@
package third
import (
"context"
"crypto/rand"
"fmt"
"time"
"github.com/OpenIMSDK/protocol/constant"
"github.com/OpenIMSDK/protocol/third"
"github.com/OpenIMSDK/tools/errs"
"github.com/OpenIMSDK/tools/utils"
utils2 "github.com/OpenIMSDK/tools/utils"
"github.com/openimsdk/open-im-server/v3/pkg/authverify"
relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
)
func genLogID() string {
const dataLen = 10
data := make([]byte, dataLen)
rand.Read(data)
chars := []byte("0123456789")
for i := 0; i < len(data); i++ {
if i == 0 {
data[i] = chars[1:][data[i]%9]
} else {
data[i] = chars[data[i]%10]
}
}
return string(data)
}
func (t *thirdServer) UploadLogs(ctx context.Context, req *third.UploadLogsReq) (*third.UploadLogsResp, error) {
var DBlogs []*relationtb.Log
userID := ctx.Value(constant.OpUserID).(string)
platform := constant.PlatformID2Name[int(req.Platform)]
for _, fileURL := range req.FileURLs {
log := relationtb.Log{
Version: req.Version,
SystemType: req.SystemType,
Platform: platform,
UserID: userID,
CreateTime: time.Now(),
Url: fileURL.URL,
FileName: fileURL.Filename,
}
for i := 0; i < 20; i++ {
id := genLogID()
logs, err := t.thirdDatabase.GetLogs(ctx, []string{id}, "")
if err != nil {
return nil, err
}
if len(logs) == 0 {
log.LogID = id
break
}
}
if log.LogID == "" {
return nil, errs.ErrData.Wrap("Log id gen error")
}
DBlogs = append(DBlogs, &log)
}
err := t.thirdDatabase.UploadLogs(ctx, DBlogs)
if err != nil {
return nil, err
}
return &third.UploadLogsResp{}, nil
}
func (t *thirdServer) DeleteLogs(ctx context.Context, req *third.DeleteLogsReq) (*third.DeleteLogsResp, error) {
if err := authverify.CheckAdmin(ctx); err != nil {
return nil, err
}
userID := ""
logs, err := t.thirdDatabase.GetLogs(ctx, req.LogIDs, userID)
if err != nil {
return nil, err
}
var logIDs []string
for _, log := range logs {
logIDs = append(logIDs, log.LogID)
}
if ids := utils2.Single(req.LogIDs, logIDs); len(ids) > 0 {
return nil, errs.ErrRecordNotFound.Wrap(fmt.Sprintf("logIDs not found%#v", ids))
}
err = t.thirdDatabase.DeleteLogs(ctx, req.LogIDs, userID)
if err != nil {
return nil, err
}
return &third.DeleteLogsResp{}, nil
}
func dbToPbLogInfos(logs []*relationtb.Log) []*third.LogInfo {
db2pbForLogInfo := func(log *relationtb.Log) *third.LogInfo {
return &third.LogInfo{
Filename: log.FileName,
UserID: log.UserID,
Platform: utils.StringToInt32(log.Platform),
Url: log.Url,
CreateTime: log.CreateTime.UnixMilli(),
LogID: log.LogID,
SystemType: log.SystemType,
Version: log.Version,
Ex: log.Ex,
}
}
return utils.Slice(logs, db2pbForLogInfo)
}
func (t *thirdServer) SearchLogs(ctx context.Context, req *third.SearchLogsReq) (*third.SearchLogsResp, error) {
if err := authverify.CheckAdmin(ctx); err != nil {
return nil, err
}
var (
resp third.SearchLogsResp
userIDs []string
)
if req.StartTime > req.EndTime {
return nil, errs.ErrArgs.Wrap("startTime>endTime")
}
total, logs, err := t.thirdDatabase.SearchLogs(ctx, req.Keyword, time.UnixMilli(req.StartTime), time.UnixMilli(req.EndTime), req.Pagination.PageNumber, req.Pagination.ShowNumber)
if err != nil {
return nil, err
}
pbLogs := dbToPbLogInfos(logs)
for _, log := range logs {
userIDs = append(userIDs, log.UserID)
}
users, err := t.thirdDatabase.FindUsers(ctx, userIDs)
if err != nil {
return nil, err
}
IDtoName := make(map[string]string)
for _, user := range users {
IDtoName[user.UserID] = user.Nickname
}
for _, pbLog := range pbLogs {
pbLog.Nickname = IDtoName[pbLog.UserID]
}
resp.LogsInfos = pbLogs
resp.Total = total
return &resp, nil
}
+1 -2
View File
@@ -35,7 +35,6 @@ import (
"github.com/openimsdk/open-im-server/v3/pkg/common/db/controller"
"github.com/openimsdk/open-im-server/v3/pkg/common/db/relation"
relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
)
@@ -80,7 +79,7 @@ func Start(client discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) e
}
third.RegisterThirdServer(server, &thirdServer{
apiURL: apiURL,
thirdDatabase: controller.NewThirdDatabase(cache.NewMsgCacheModel(rdb), db),
thirdDatabase: controller.NewThirdDatabase(cache.NewMsgCacheModel(rdb)),
userRpcClient: rpcclient.NewUserRpcClient(client),
s3dataBase: controller.NewS3Database(o, relation.NewObjectInfo(db)),
defaultExpire: time.Hour * 24 * 7,
-1
View File
@@ -29,7 +29,6 @@ type SendMsg struct {
SessionType int32 `json:"sessionType" binding:"required"`
IsOnlineOnly bool `json:"isOnlineOnly"`
NotOfflinePush bool `json:"notOfflinePush"`
SendTime int64 `json:"sendTime"`
OfflinePushInfo *sdkws.OfflinePushInfo `json:"offlinePushInfo"`
}
-3
View File
@@ -128,14 +128,12 @@ type configStruct struct {
SecretAccessKey string `yaml:"secretAccessKey"`
SessionToken string `yaml:"sessionToken"`
SignEndpoint string `yaml:"signEndpoint"`
PublicRead bool `yaml:"publicRead"`
} `yaml:"minio"`
Cos struct {
BucketURL string `yaml:"bucketURL"`
SecretID string `yaml:"secretID"`
SecretKey string `yaml:"secretKey"`
SessionToken string `yaml:"sessionToken"`
PublicRead bool `yaml:"publicRead"`
} `yaml:"cos"`
Oss struct {
Endpoint string `yaml:"endpoint"`
@@ -144,7 +142,6 @@ type configStruct struct {
AccessKeyID string `yaml:"accessKeyID"`
AccessKeySecret string `yaml:"accessKeySecret"`
SessionToken string `yaml:"sessionToken"`
PublicRead bool `yaml:"publicRead"`
} `yaml:"oss"`
} `yaml:"object"`
+13 -33
View File
@@ -38,7 +38,6 @@ const (
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
)
@@ -59,8 +58,11 @@ type ConversationCache interface {
DelConversations(ownerUserID string, conversationIDs ...string) ConversationCache
DelUsersConversation(conversationID string, ownerUserIDs ...string) ConversationCache
// get one conversation from msgCache
GetConversations(ctx context.Context, ownerUserID string,
conversationIDs []string) ([]*relationtb.ConversationModel, error)
GetConversations(
ctx context.Context,
ownerUserID string,
conversationIDs []string,
) ([]*relationtb.ConversationModel, error)
// get one user's all conversations from msgCache
GetUserAllConversations(ctx context.Context, ownerUserID string) ([]*relationtb.ConversationModel, error)
// get user conversation recv msg from msgCache
@@ -76,11 +78,11 @@ type ConversationCache interface {
GetUserAllHasReadSeqs(ctx context.Context, ownerUserID string) (map[string]int64, error)
DelUserAllHasReadSeqs(ownerUserID string, conversationIDs ...string) ConversationCache
GetConversationsByConversationID(ctx context.Context,
conversationIDs []string) ([]*relationtb.ConversationModel, error)
GetConversationsByConversationID(
ctx context.Context,
conversationIDs []string,
) ([]*relationtb.ConversationModel, error)
DelConversationByConversationID(conversationIDs ...string) ConversationCache
GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error)
DelConversationNotReceiveMessageUserIDs(conversationIDs ...string) ConversationCache
}
func NewConversationRedis(
@@ -151,10 +153,6 @@ func (c *ConversationRedisCache) getConversationHasReadSeqKey(ownerUserID, conve
return conversationHasReadSeqKey + ownerUserID + ":" + conversationID
}
func (c *ConversationRedisCache) getConversationNotReceiveMessageUserIDsKey(conversationID string) string {
return conversationNotReceiveMessageUserIDsKey + conversationID
}
func (c *ConversationRedisCache) GetUserConversationIDs(ctx context.Context, ownerUserID string) ([]string, error) {
return getCache(
ctx,
@@ -413,8 +411,10 @@ func (c *ConversationRedisCache) GetUserAllHasReadSeqs(
)
}
func (c *ConversationRedisCache) DelUserAllHasReadSeqs(ownerUserID string,
conversationIDs ...string) ConversationCache {
func (c *ConversationRedisCache) DelUserAllHasReadSeqs(
ownerUserID string,
conversationIDs ...string,
) ConversationCache {
cache := c.NewCache()
for _, conversationID := range conversationIDs {
cache.AddKeys(c.getConversationHasReadSeqKey(ownerUserID, conversationID))
@@ -432,23 +432,3 @@ func (c *ConversationRedisCache) GetConversationsByConversationID(
func (c *ConversationRedisCache) DelConversationByConversationID(conversationIDs ...string) ConversationCache {
panic("implement me")
}
func (c *ConversationRedisCache) GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error) {
return getCache(
ctx,
c.rcClient,
c.getConversationNotReceiveMessageUserIDsKey(conversationID),
c.expireTime,
func(ctx context.Context) ([]string, error) {
return c.conversationDB.GetConversationNotReceiveMessageUserIDs(ctx, conversationID)
},
)
}
func (c *ConversationRedisCache) DelConversationNotReceiveMessageUserIDs(conversationIDs ...string) ConversationCache {
cache := c.NewCache()
for _, conversationID := range conversationIDs {
cache.AddKeys(c.getConversationNotReceiveMessageUserIDsKey(conversationID))
}
return cache
}
+3 -17
View File
@@ -53,7 +53,6 @@ type ConversationDatabase interface {
GetUserAllHasReadSeqs(ctx context.Context, ownerUserID string) (map[string]int64, error)
GetConversationsByConversationID(ctx context.Context, conversationIDs []string) ([]*relationtb.ConversationModel, error)
GetConversationIDsNeedDestruct(ctx context.Context) ([]*relationtb.ConversationModel, error)
GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error)
}
func NewConversationDatabase(conversation relationtb.ConversationModelInterface, cache cache.ConversationCache, tx tx.Tx) ConversationDatabase {
@@ -89,9 +88,6 @@ func (c *conversationDatabase) SetUsersConversationFiledTx(ctx context.Context,
cache = cache.DelUserAllHasReadSeqs(userID, conversation.ConversationID)
}
}
if _, ok := filedMap["recv_msg_opt"]; ok {
cache = cache.DelConversationNotReceiveMessageUserIDs(conversation.ConversationID)
}
}
NotUserIDs := utils.DifferenceString(haveUserIDs, userIDs)
log.ZDebug(ctx, "SetUsersConversationFiledTx", "NotUserIDs", NotUserIDs, "haveUserIDs", haveUserIDs, "userIDs", userIDs)
@@ -125,12 +121,7 @@ func (c *conversationDatabase) UpdateUsersConversationFiled(ctx context.Context,
if err != nil {
return err
}
cache := c.cache.NewCache()
cache = cache.DelUsersConversation(conversationID, userIDs...)
if _, ok := args["recv_msg_opt"]; ok {
cache = cache.DelConversationNotReceiveMessageUserIDs(conversationID)
}
return cache.ExecDel(ctx)
return c.cache.DelUsersConversation(conversationID, userIDs...).ExecDel(ctx)
}
func (c *conversationDatabase) CreateConversation(ctx context.Context, conversations []*relationtb.ConversationModel) error {
@@ -141,7 +132,6 @@ func (c *conversationDatabase) CreateConversation(ctx context.Context, conversat
cache := c.cache.NewCache()
for _, conversation := range conversations {
cache = cache.DelConversations(conversation.OwnerUserID, conversation.ConversationID)
cache = cache.DelConversationNotReceiveMessageUserIDs(conversation.ConversationID)
userIDs = append(userIDs, conversation.OwnerUserID)
}
return cache.DelConversationIDs(userIDs...).DelUserConversationIDsHash(userIDs...).ExecDel(ctx)
@@ -234,7 +224,7 @@ func (c *conversationDatabase) SetUserConversations(ctx context.Context, ownerUs
if err != nil {
return err
}
cache = cache.DelConversationIDs(ownerUserID).DelUserConversationIDsHash(ownerUserID).DelConversationNotReceiveMessageUserIDs(utils.Slice(notExistConversations, func(e *relationtb.ConversationModel) string { return e.ConversationID })...)
cache = cache.DelConversationIDs(ownerUserID).DelUserConversationIDsHash(ownerUserID)
}
return nil
}); err != nil {
@@ -260,7 +250,7 @@ func (c *conversationDatabase) CreateGroupChatConversation(ctx context.Context,
for _, v := range notExistUserIDs {
conversation := relationtb.ConversationModel{ConversationType: constant.SuperGroupChatType, GroupID: groupID, OwnerUserID: v, ConversationID: conversationID}
conversations = append(conversations, &conversation)
cache = cache.DelConversations(v, conversationID).DelConversationNotReceiveMessageUserIDs(conversationID)
cache = cache.DelConversations(v, conversationID)
}
cache = cache.DelConversationIDs(notExistUserIDs...).DelUserConversationIDsHash(notExistUserIDs...)
if len(conversations) > 0 {
@@ -306,7 +296,3 @@ func (c *conversationDatabase) GetConversationsByConversationID(ctx context.Cont
func (c *conversationDatabase) GetConversationIDsNeedDestruct(ctx context.Context) ([]*relationtb.ConversationModel, error) {
return c.conversationDB.GetConversationIDsNeedDestruct(ctx)
}
func (c *conversationDatabase) GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error) {
return c.cache.GetConversationNotReceiveMessageUserIDs(ctx, conversationID)
}
+3 -42
View File
@@ -16,60 +16,21 @@ package controller
import (
"context"
"time"
"gorm.io/gorm"
"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
dbimpl "github.com/openimsdk/open-im-server/v3/pkg/common/db/relation"
"github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
)
type ThirdDatabase interface {
FcmUpdateToken(ctx context.Context, account string, platformID int, fcmToken string, expireTime int64) error
SetAppBadge(ctx context.Context, userID string, value int) error
//about log for debug
UploadLogs(ctx context.Context, logs []*relation.Log) error
DeleteLogs(ctx context.Context, logID []string, userID string) error
SearchLogs(ctx context.Context, keyword string, start time.Time, end time.Time, pageNumber int32, showNumber int32) (uint32, []*relation.Log, error)
GetLogs(ctx context.Context, LogIDs []string, userID string) ([]*relation.Log, error)
FindUsers(ctx context.Context, userIDs []string) ([]*relation.UserModel, error)
}
type thirdDatabase struct {
cache cache.MsgModel
logdb relation.LogInterface
userdb relation.UserModelInterface
cache cache.MsgModel
}
// FindUsers implements ThirdDatabase.
func (t *thirdDatabase) FindUsers(ctx context.Context, userIDs []string) ([]*relation.UserModel, error) {
return t.userdb.Find(ctx, userIDs)
}
// DeleteLogs implements ThirdDatabase.
func (t *thirdDatabase) DeleteLogs(ctx context.Context, logID []string, userID string) error {
return t.logdb.Delete(ctx, logID, userID)
}
// GetLogs implements ThirdDatabase.
func (t *thirdDatabase) GetLogs(ctx context.Context, LogIDs []string, userID string) ([]*relation.Log, error) {
return t.logdb.Get(ctx, LogIDs, userID)
}
// SearchLogs implements ThirdDatabase.
func (t *thirdDatabase) SearchLogs(ctx context.Context, keyword string, start time.Time, end time.Time, pageNumber int32, showNumber int32) (uint32, []*relation.Log, error) {
return t.logdb.Search(ctx, keyword, start, end, pageNumber, showNumber)
}
// UploadLogs implements ThirdDatabase.
func (t *thirdDatabase) UploadLogs(ctx context.Context, logs []*relation.Log) error {
return t.logdb.Create(ctx, logs)
}
func NewThirdDatabase(cache cache.MsgModel, db *gorm.DB) ThirdDatabase {
return &thirdDatabase{cache: cache, logdb: dbimpl.NewLogGorm(db), userdb: dbimpl.NewUserGorm(db)}
func NewThirdDatabase(cache cache.MsgModel) ThirdDatabase {
return &thirdDatabase{cache: cache}
}
func (t *thirdDatabase) FcmUpdateToken(
@@ -17,8 +17,6 @@ package relation
import (
"context"
"github.com/OpenIMSDK/tools/errs"
"gorm.io/gorm"
"github.com/OpenIMSDK/protocol/constant"
@@ -216,24 +214,3 @@ func (c *ConversationGorm) GetConversationIDsNeedDestruct(
"",
)
}
func (c *ConversationGorm) GetConversationRecvMsgOpt(ctx context.Context, userID string, conversationID string) (int32, error) {
var recvMsgOpt int32
return recvMsgOpt, errs.Wrap(
c.db(ctx).
Model(&relation.ConversationModel{}).
Where("conversation_id = ? and owner_user_id in ?", conversationID, userID).
Pluck("recv_msg_opt", &recvMsgOpt).
Error,
)
}
func (c *ConversationGorm) GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error) {
var userIDs []string
return userIDs, errs.Wrap(
c.db(ctx).
Model(&relation.ConversationModel{}).
Where("conversation_id = ? and recv_msg_opt <> ?", conversationID, constant.ReceiveMessage).
Pluck("owner_user_id", &userIDs).Error,
)
}
-47
View File
@@ -1,47 +0,0 @@
package relation
import (
"context"
"time"
"github.com/OpenIMSDK/tools/errs"
"github.com/OpenIMSDK/tools/ormutil"
"gorm.io/gorm"
relationtb "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/relation"
)
type LogGorm struct {
db *gorm.DB
}
func (l *LogGorm) Create(ctx context.Context, log []*relationtb.Log) error {
return errs.Wrap(l.db.WithContext(ctx).Create(log).Error)
}
func (l *LogGorm) Search(ctx context.Context, keyword string, start time.Time, end time.Time, pageNumber int32, showNumber int32) (uint32, []*relationtb.Log, error) {
db := l.db.WithContext(ctx).Where("create_time >= ?", start)
if end.UnixMilli() != 0 {
db = l.db.WithContext(ctx).Where("create_time <= ?", end)
}
return ormutil.GormSearch[relationtb.Log](db, []string{"user_id"}, keyword, pageNumber, showNumber)
}
func (l *LogGorm) Delete(ctx context.Context, logIDs []string, userID string) error {
if userID == "" {
return errs.Wrap(l.db.WithContext(ctx).Where("log_id in ?", logIDs).Delete(&relationtb.Log{}).Error)
}
return errs.Wrap(l.db.WithContext(ctx).Where("log_id in ? and user_id=?", logIDs, userID).Delete(&relationtb.Log{}).Error)
}
func (l *LogGorm) Get(ctx context.Context, logIDs []string, userID string) ([]*relationtb.Log, error) {
var logs []*relationtb.Log
if userID == "" {
return logs, errs.Wrap(l.db.WithContext(ctx).Where("log_id in ?", logIDs).Find(&logs).Error)
}
return logs, errs.Wrap(l.db.WithContext(ctx).Where("log_id in ? and user_id=?", logIDs, userID).Find(&logs).Error)
}
func NewLogGorm(db *gorm.DB) relationtb.LogInterface {
db.AutoMigrate(&relationtb.Log{})
return &LogGorm{db: db}
}
+5 -15
View File
@@ -288,7 +288,7 @@ func (c *Cos) AccessURL(ctx context.Context, name string, expire time.Duration,
style = append(style, "format/"+opt.Image.Format)
}
if len(style) > 0 {
imageMogr = "imageMogr2/thumbnail/" + strings.Join(style, "/") + "/ignore-error/1"
imageMogr = "&imageMogr2/thumbnail/" + strings.Join(style, "/") + "/ignore-error/1"
}
}
if opt.ContentType != "" {
@@ -306,23 +306,13 @@ func (c *Cos) AccessURL(ctx context.Context, name string, expire time.Duration,
} else if expire < time.Second {
expire = time.Second
}
rawURL, err := c.getPresignedURL(ctx, name, expire, &option)
rawURL, err := c.client.Object.GetPresignedURL(ctx, http.MethodGet, name, c.credential.SecretID, c.credential.SecretKey, expire, &option)
if err != nil {
return "", err
}
urlStr := rawURL.String()
if imageMogr != "" {
if rawURL.RawQuery == "" {
rawURL.RawQuery = imageMogr
} else {
rawURL.RawQuery = rawURL.RawQuery + "&" + imageMogr
}
urlStr += imageMogr
}
return rawURL.String(), nil
}
func (c *Cos) getPresignedURL(ctx context.Context, name string, expire time.Duration, opt *cos.PresignedURLOptions) (*url.URL, error) {
if !config.Config.Object.Cos.PublicRead {
return c.client.Object.GetPresignedURL(ctx, http.MethodGet, name, c.credential.SecretID, c.credential.SecretKey, expire, opt)
}
return c.client.Object.GetObjectURL(name), nil
return urlStr, nil
}
+15
View 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 cos // import "github.com/openimsdk/open-im-server/v3/pkg/common/db/s3/cos"
-13
View File
@@ -1,13 +0,0 @@
package cos
import (
"context"
"net/http"
"net/url"
_ "unsafe"
"github.com/tencentyun/cos-go-sdk-v5"
)
//go:linkname newRequest github.com/tencentyun/cos-go-sdk-v5.(*Client).newRequest
func newRequest(c *cos.Client, ctx context.Context, baseURL *url.URL, uri, method string, body interface{}, optQuery interface{}, optHeader interface{}) (req *http.Request, err error)
+15
View 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 minio // import "github.com/openimsdk/open-im-server/v3/pkg/common/db/s3/minio"
-11
View File
@@ -1,11 +0,0 @@
package minio
import (
"net/url"
_ "unsafe"
"github.com/minio/minio-go/v7"
)
//go:linkname makeTargetURL github.com/minio/minio-go/v7.(*Client).makeTargetURL
func makeTargetURL(client *minio.Client, bucketName, objectName, bucketLocation string, isVirtualHostStyle bool, queryValues url.Values) (*url.URL, error)
+1 -18
View File
@@ -139,15 +139,6 @@ func (m *Minio) initMinio(ctx context.Context) error {
return fmt.Errorf("make bucket error: %w", err)
}
}
if conf.PublicRead {
policy := fmt.Sprintf(
`{"Version": "2012-10-17","Statement": [{"Action": ["s3:GetObject","s3:PutObject"],"Effect": "Allow","Principal": {"AWS": ["*"]},"Resource": ["arn:aws:s3:::%s/*"],"Sid": ""}]}`,
conf.Bucket,
)
if err := m.core.Client.SetBucketPolicy(ctx, conf.Bucket, policy); err != nil {
return err
}
}
m.location, err = m.core.Client.GetBucketLocation(ctx, conf.Bucket)
if err != nil {
return err
@@ -384,15 +375,7 @@ func (m *Minio) presignedGetObject(ctx context.Context, name string, expire time
} else if expire < time.Second {
expire = time.Second
}
var (
rawURL *url.URL
err error
)
if config.Config.Object.Minio.PublicRead {
rawURL, err = makeTargetURL(m.sign, m.bucket, name, m.location, false, query)
} else {
rawURL, err = m.sign.PresignedGetObject(ctx, m.bucket, name, expire, query)
}
rawURL, err := m.sign.PresignedGetObject(ctx, m.bucket, name, expire, query)
if err != nil {
return "", err
}
+15
View 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 oss // import "github.com/openimsdk/open-im-server/v3/pkg/common/db/s3/oss"
+7 -21
View File
@@ -20,7 +20,6 @@ import (
"fmt"
"net/http"
"net/url"
"reflect"
"strconv"
"strings"
"time"
@@ -70,7 +69,6 @@ func NewOSS() (s3.Interface, error) {
bucketURL: conf.BucketURL,
bucket: bucket,
credentials: client.Config.GetCredentials(),
um: *(*urlMaker)(reflect.ValueOf(bucket.Client.Conn).Elem().FieldByName("url").UnsafePointer()),
}, nil
}
@@ -78,7 +76,6 @@ type OSS struct {
bucketURL string
bucket *oss.Bucket
credentials oss.Credentials
um urlMaker
}
func (o *OSS) Engine() string {
@@ -166,7 +163,7 @@ func (o *OSS) AuthSign(ctx context.Context, uploadID string, name string, expire
request.Header.Set(oss.HTTPHeaderHost, request.Host)
request.Header.Set(oss.HTTPHeaderDate, now)
request.Header.Set(oss.HttpHeaderOssDate, now)
signHeader(*o.bucket.Client.Conn, request, fmt.Sprintf(`/%s/%s?partNumber=%d&uploadId=%s`, o.bucket.BucketName, name, partNumber, uploadID))
ossSignHeader(o.bucket.Client.Conn, request, fmt.Sprintf(`/%s/%s?partNumber=%d&uploadId=%s`, o.bucket.BucketName, name, partNumber, uploadID))
delete(request.Header, oss.HTTPHeaderDate)
result.Parts[i] = s3.SignPart{
PartNumber: partNumber,
@@ -275,7 +272,6 @@ func (o *OSS) ListUploadedParts(ctx context.Context, uploadID string, name strin
}
func (o *OSS) AccessURL(ctx context.Context, name string, expire time.Duration, opt *s3.AccessURLOption) (string, error) {
publicRead := config.Config.Object.Oss.PublicRead
var opts []oss.Option
if opt != nil {
if opt.Image != nil {
@@ -303,13 +299,11 @@ func (o *OSS) AccessURL(ctx context.Context, name string, expire time.Duration,
process += ",format," + format
opts = append(opts, oss.Process(process))
}
if !publicRead {
if opt.ContentType != "" {
opts = append(opts, oss.ResponseContentType(opt.ContentType))
}
if opt.Filename != "" {
opts = append(opts, oss.ResponseContentDisposition(`attachment; filename=`+strconv.Quote(opt.Filename)))
}
if opt.ContentType != "" {
opts = append(opts, oss.ResponseContentType(opt.ContentType))
}
if opt.Filename != "" {
opts = append(opts, oss.ResponseContentDisposition(`attachment; filename=`+strconv.Quote(opt.Filename)))
}
}
if expire <= 0 {
@@ -317,13 +311,5 @@ func (o *OSS) AccessURL(ctx context.Context, name string, expire time.Duration,
} else if expire < time.Second {
expire = time.Second
}
if !publicRead {
return o.bucket.SignURL(name, http.MethodGet, int64(expire/time.Second), opts...)
}
rawParams, err := oss.GetRawParams(opts)
if err != nil {
return "", err
}
params := getURLParams(*o.bucket.Client.Conn, rawParams)
return getURL(o.um, o.bucket.BucketName, name, params).String(), nil
return o.bucket.SignURL(name, http.MethodGet, int64(expire/time.Second), opts...)
}
@@ -16,24 +16,10 @@ package oss
import (
"net/http"
"net/url"
_ "unsafe"
"github.com/aliyun/aliyun-oss-go-sdk/oss"
)
//go:linkname signHeader github.com/aliyun/aliyun-oss-go-sdk/oss.Conn.signHeader
func signHeader(c oss.Conn, req *http.Request, canonicalizedResource string)
//go:linkname getURLParams github.com/aliyun/aliyun-oss-go-sdk/oss.Conn.getURLParams
func getURLParams(c oss.Conn, params map[string]interface{}) string
//go:linkname getURL github.com/aliyun/aliyun-oss-go-sdk/oss.urlMaker.getURL
func getURL(um urlMaker, bucket, object, params string) *url.URL
type urlMaker struct {
Scheme string
NetLoc string
Type int
IsProxy bool
}
//go:linkname ossSignHeader github.com/aliyun/aliyun-oss-go-sdk/oss.(*Conn).signHeader
func ossSignHeader(c *oss.Conn, req *http.Request, canonicalizedResource string)
@@ -66,6 +66,5 @@ type ConversationModelInterface interface {
GetUserAllHasReadSeqs(ctx context.Context, ownerUserID string) (hashReadSeqs map[string]int64, err error)
GetConversationsByConversationID(ctx context.Context, conversationIDs []string) ([]*ConversationModel, error)
GetConversationIDsNeedDestruct(ctx context.Context) ([]*ConversationModel, error)
GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error)
NewTx(tx any) ConversationModelInterface
}
-25
View File
@@ -1,25 +0,0 @@
package relation
import (
"context"
"time"
)
type Log struct {
LogID string `gorm:"column:log_id;primary_key;type:char(64)"`
Platform string `gorm:"column:platform;type:varchar(32)"`
UserID string `gorm:"column:user_id;type:char(64)"`
CreateTime time.Time `gorm:"index:,sort:desc"`
Url string `gorm:"column:url;type varchar(255)"`
FileName string `gorm:"column:filename;type varchar(255)"`
SystemType string `gorm:"column:system_type;type varchar(255)"`
Version string `gorm:"column:version;type varchar(255)"`
Ex string `gorm:"column:ex;type varchar(255)"`
}
type LogInterface interface {
Create(ctx context.Context, log []*Log) error
Search(ctx context.Context, keyword string, start time.Time, end time.Time, pageNumber int32, showNumber int32) (uint32, []*Log, error)
Delete(ctx context.Context, logID []string, userID string) error
Get(ctx context.Context, logIDs []string, userID string) ([]*Log, error)
}
+5 -4
View File
@@ -25,7 +25,6 @@ source "${OPENIM_ROOT}/scripts/lib/init.sh"
trap 'openim::util::onCtrlC' INT
chmod +x "${OPENIM_ROOT}"/scripts/*.sh
"${OPENIM_ROOT}"/scripts/init-config.sh
openim::util::ensure_docker_daemon_connectivity
@@ -40,10 +39,12 @@ else
fi
pushd "${OPENIM_ROOT}"
${DOCKER_COMPOSE_COMMAND} up -d
${DOCKER_COMPOSE_COMMAND} stop
curl https://gitee.com/openimsdk/openim-docker/raw/main/example/full-openim-server-and-chat.yml -o docker-compose.yml && make init && docker compose up -d
"${OPENIM_ROOT}"/scripts/init-config.sh
${DOCKER_COMPOSE_COMMAND} up --remove-orphans -d
sleep 60
${DOCKER_COMPOSE_COMMAND} logs
${DOCKER_COMPOSE_COMMAND} logs openim-server
${DOCKER_COMPOSE_COMMAND} ps
popd
+1 -5
View File
@@ -176,7 +176,7 @@ readonly API_URL=${API_URL:-"http://${IP}:${API_OPENIM_PORT}"}
def "OBJECT_ENABLE" "minio" # 对象是否启用
# 对象的API地址
readonly OBJECT_APIURL=${OBJECT_APIURL:-"${API_URL}"}
readonly OBJECT_APIURL=${OBJECT_APIURL:-"http://${API_URL}"}
def "MINIO_BUCKET" "openim" # MinIO的存储桶名称
def "MINIO_PORT" "10005" # MinIO的端口
# MinIO的端点URL
@@ -186,21 +186,17 @@ def "MINIO_ACCESS_KEY" "${USER}"
def "MINIO_SECRET_KEY" "${PASSWORD}" # MinIO的密钥
def "MINIO_SESSION_TOKEN" # MinIO的会话令牌
readonly MINIO_SIGN_ENDPOINT=${MINIO_SIGN_ENDPOINT:-"http://${IP}:${MINIO_PORT}"} # signEndpoint为minio公网地址
def "MINIO_PUBLIC_READ" "false" # 公有读
# 腾讯云COS的存储桶URL
def "COS_BUCKET_URL" "https://temp-1252357374.cos.ap-chengdu.myqcloud.com"
def "COS_SECRET_ID" # 腾讯云COS的密钥ID
def "COS_SECRET_KEY" # 腾讯云COS的密钥
def "COS_SESSION_TOKEN" # 腾讯云COS的会话令牌
def "COS_PUBLIC_READ" "false" # 公有读
def "OSS_ENDPOINT" "https://oss-cn-chengdu.aliyuncs.com" # 阿里云OSS的端点URL
def "OSS_BUCKET" "demo-9999999" # 阿里云OSS的存储桶名称
def "OSS_BUCKET_URL" "https://demo-9999999.oss-cn-chengdu.aliyuncs.com" # 阿里云OSS的存储桶URL
def "OSS_ACCESS_KEY_ID" # 阿里云OSS的访问密钥ID
def "OSS_ACCESS_KEY_SECRET" # 阿里云OSS的密钥
def "OSS_SESSION_TOKEN" # 阿里云OSS的会话令牌
def "OSS_PUBLIC_READ" "false" # 公有读
###################### Redis 配置信息 ######################
def "REDIS_PORT" "16379" # Redis的端口
+12 -19
View File
@@ -22,6 +22,7 @@ import (
"net"
"net/url"
"os"
"strings"
"time"
"github.com/minio/minio-go/v7"
@@ -116,24 +117,16 @@ func main() {
os.Exit(1)
}
func checkMinioIP() error {
for _, i := range []string{config.Config.Object.ApiURL, config.Config.Object.Minio.SignEndpoint} {
u, err := url.Parse(i)
if err != nil {
return utils.Wrap(err, "api format error,please check config file apiURL or Minio SignEndpoint")
}
if u.Scheme == "https" {
continue
}
host, _, err := net.SplitHostPort(u.Host)
if err != nil {
host = u.Host
}
if host == "127.0.0.1" {
return ErrConfig.Wrap("apiURL or Minio SignEndpoint endpoint contain 127.0.0.1,please modify it")
}
func exactIP(urll string) string {
u, _ := url.Parse(urll)
host, _, err := net.SplitHostPort(u.Host)
if err != nil {
host = u.Host
}
return nil
if strings.HasSuffix(host, ":") {
host = host[0 : len(host)-1]
}
return host
}
func checkMysql() error {
@@ -212,8 +205,8 @@ func checkMinio() error {
return ErrComponentStart.Wrap("Minio server is offline")
}
}
if checkMinioIP() != nil {
return checkMinioIP()
if exactIP(config.Config.Object.ApiURL) == "127.0.0.1" || exactIP(config.Config.Object.Minio.SignEndpoint) == "127.0.0.1" {
return ErrConfig.Wrap("apiURL or Minio SignEndpoint endpoint contain 127.0.0.1")
}
}
return nil
+2 -3
View File
@@ -1,4 +1,4 @@
module github.com/openimsdk/open-im-server/v3/tools/imctl
module github.com/openimsdk/open-im-server/v3/tools/imctl
go 1.18
@@ -14,6 +14,5 @@ require (
require (
github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect
github.com/inconshreveable/mousetrap v1.1.0 // indirect
golang.org/x/sys v0.10.0 // indirect
k8s.io/kubernetes v1.28.2
golang.org/x/sys v0.1.0 // indirect
)
-3
View File
@@ -20,8 +20,5 @@ github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An
golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.1.0 h1:kunALQeHf1/185U1i0GOB/fy1IPRDDpuoOOqRReG57U=
golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.10.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
k8s.io/kubernetes v1.28.2 h1:GhcnYeNTukeaC0dD5BC+UWBvzQsFEpWj7XBVMQptfYc=
k8s.io/kubernetes v1.28.2/go.mod h1:FmB1Mlp9ua0ezuwQCTGs/y6wj/fVisN2sVxhzjj0WDk=
-20
View File
@@ -1,20 +0,0 @@
module github.com/openimsdk/open-im-server/v3/tools/url2im
go 1.20
require (
github.com/OpenIMSDK/protocol v0.0.21
github.com/kelindar/bitmap v1.5.1
)
require (
github.com/golang/protobuf v1.5.3 // indirect
github.com/kelindar/simd v1.1.2 // indirect
github.com/klauspost/cpuid/v2 v2.2.4 // indirect
golang.org/x/net v0.9.0 // indirect
golang.org/x/sys v0.7.0 // indirect
golang.org/x/text v0.9.0 // indirect
google.golang.org/genproto v0.0.0-20230410155749-daa745c078e1 // indirect
google.golang.org/grpc v1.56.2 // indirect
google.golang.org/protobuf v1.31.0 // indirect
)
-33
View File
@@ -1,33 +0,0 @@
github.com/OpenIMSDK/protocol v0.0.21 h1:5H6H+hJ9d/VgRqttvxD/zfK9Asd+4M8Eknk5swSbUVY=
github.com/OpenIMSDK/protocol v0.0.21/go.mod h1:F25dFrwrIx3lkNoiuf6FkCfxuwf8L4Z8UIsdTHP/r0Y=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg=
github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
github.com/kelindar/bitmap v1.5.1 h1:+ZmZdwHbJ+CGE+q/aAJ74KJSnp0vOlGD7KY5x51mVzk=
github.com/kelindar/bitmap v1.5.1/go.mod h1:j3qZjxH9s4OtvsnFTP2bmPkjqil9Y2xQlxPYHexasEA=
github.com/kelindar/simd v1.1.2 h1:KduKb+M9cMY2HIH8S/cdJyD+5n5EGgq+Aeeleos55To=
github.com/kelindar/simd v1.1.2/go.mod h1:inq4DFudC7W8L5fhxoeZflLRNpWSs0GNx6MlWFvuvr0=
github.com/klauspost/cpuid/v2 v2.2.4 h1:acbojRNwl3o09bUq+yDCtZFc1aiwaAAxtcn8YkZXnvk=
github.com/klauspost/cpuid/v2 v2.2.4/go.mod h1:RVVoqg1df56z8g3pUjL/3lE5UfnlrJX8tyFgg4nqhuY=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/stretchr/testify v1.8.2 h1:+h33VjcLVPDHtOdpUCuF+7gSuG3yGIftsP1YvFihtJ8=
golang.org/x/net v0.9.0 h1:aWJ/m6xSmxWBx+V0XRHTlrYrPG56jKsLdTFmsSsCzOM=
golang.org/x/net v0.9.0/go.mod h1:d48xBJpPfHeWQsugry2m+kC02ZBRGRgulfHnEXEuWns=
golang.org/x/sys v0.0.0-20220704084225-05e143d24a9e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.7.0 h1:3jlCCIQZPdOYu1h8BkNvLz8Kgwtae2cagcG/VamtZRU=
golang.org/x/sys v0.7.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/text v0.9.0 h1:2sjJmO8cDvYveuX97RDLsxlyUxLl+GHoLxBiRdHllBE=
golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
google.golang.org/genproto v0.0.0-20230410155749-daa745c078e1 h1:KpwkzHKEF7B9Zxg18WzOa7djJ+Ha5DzthMyZYQfEn2A=
google.golang.org/genproto v0.0.0-20230410155749-daa745c078e1/go.mod h1:nKE/iIaLqn2bQwXBg8f1g2Ylh6r5MN5CmZvuzZCgsCU=
google.golang.org/grpc v1.56.2 h1:fVRFRnXvU+x6C4IlHZewvJOVHoOv1TUuQyoRsYnB4bI=
google.golang.org/grpc v1.56.2/go.mod h1:I9bI3vqKfayGqPUAwGdOSu7kt6oIJLixfffKrpXqQ9s=
google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
google.golang.org/protobuf v1.31.0 h1:g0LDEJHgrBl9N9r17Ru3sqWhkIx2NB67okBHPwC7hs8=
google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
-84
View File
@@ -1,84 +0,0 @@
package main
import (
"flag"
"log"
"os"
"path/filepath"
"time"
"github.com/openimsdk/open-im-server/v3/tools/url2im/pkg"
)
/*take.txt
{"url":"http://xxx/xxxx","name":"xxxx","contentType":"image/jpeg"}
{"url":"http://xxx/xxxx","name":"xxxx","contentType":"image/jpeg"}
{"url":"http://xxx/xxxx","name":"xxxx","contentType":"image/jpeg"}
*/
func main() {
var conf pkg.Config // 后面带*的为必填项
flag.StringVar(&conf.TaskPath, "task", "take.txt", "task path") // 任务日志文件*
flag.StringVar(&conf.ProgressPath, "progress", "", "progress path") // 进度日志文件
flag.IntVar(&conf.Concurrency, "concurrency", 1, "concurrency num") // 并发数
flag.IntVar(&conf.Retry, "retry", 1, "retry num") // 重试次数
flag.StringVar(&conf.TempDir, "temp", "", "temp dir") // 临时文件夹
flag.Int64Var(&conf.CacheSize, "cache", 1024*1024*100, "cache size") // 缓存大小(超过时,下载到磁盘)
flag.Int64Var((*int64)(&conf.Timeout), "timeout", 5000, "timeout") // 请求超时时间(毫秒)
flag.StringVar(&conf.Api, "api", "http://127.0.0.1:10002", "api") // im地址*
flag.StringVar(&conf.UserID, "userID", "openIM123456", "userID") // im管理员
flag.StringVar(&conf.Secret, "secret", "openIM123", "secret") // im config secret
flag.Parse()
if !filepath.IsAbs(conf.TaskPath) {
var err error
conf.TaskPath, err = filepath.Abs(conf.TaskPath)
if err != nil {
log.Println("get abs path err:", err)
return
}
}
if conf.ProgressPath == "" {
conf.ProgressPath = conf.TaskPath + ".progress.txt"
} else if !filepath.IsAbs(conf.ProgressPath) {
var err error
conf.ProgressPath, err = filepath.Abs(conf.ProgressPath)
if err != nil {
log.Println("get abs path err:", err)
return
}
}
if conf.TempDir == "" {
conf.TempDir = conf.TaskPath + ".temp"
}
if info, err := os.Stat(conf.TempDir); err == nil {
if !info.IsDir() {
log.Printf("temp dir %s is not dir\n", err)
return
}
} else if os.IsNotExist(err) {
if err := os.MkdirAll(conf.TempDir, os.ModePerm); err != nil {
log.Printf("mkdir temp dir %s err %+v\n", conf.TempDir, err)
return
}
defer os.RemoveAll(conf.TempDir)
} else {
log.Println("get temp dir err:", err)
return
}
if conf.Concurrency <= 0 {
conf.Concurrency = 1
}
if conf.Retry <= 0 {
conf.Retry = 1
}
if conf.CacheSize <= 0 {
conf.CacheSize = 1024 * 1024 * 100 // 100M
}
if conf.Timeout <= 0 {
conf.Timeout = 5000
}
conf.Timeout = conf.Timeout * time.Millisecond
if err := pkg.Run(conf); err != nil {
log.Println("main err:", err)
}
}
-112
View File
@@ -1,112 +0,0 @@
package pkg
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"github.com/OpenIMSDK/protocol/auth"
"github.com/OpenIMSDK/protocol/constant"
"github.com/OpenIMSDK/protocol/third"
)
type Api struct {
Api string
UserID string
Secret string
Token string
Client *http.Client
}
func (a *Api) apiPost(ctx context.Context, path string, req any, resp any) error {
operationID, _ := ctx.Value("operationID").(string)
if operationID == "" {
return errors.New("call api operationID is empty")
}
reqBody, err := json.Marshal(req)
if err != nil {
return err
}
request, err := http.NewRequestWithContext(ctx, http.MethodPost, a.Api+path, bytes.NewReader(reqBody))
if err != nil {
return err
}
DefaultRequestHeader(request.Header)
request.ContentLength = int64(len(reqBody))
request.Header.Set("Content-Type", "application/json")
request.Header.Set("operationID", operationID)
if a.Token != "" {
request.Header.Set("token", a.Token)
}
response, err := a.Client.Do(request)
if err != nil {
return err
}
defer response.Body.Close()
body, err := io.ReadAll(response.Body)
if err != nil {
return err
}
if response.StatusCode != http.StatusOK {
return fmt.Errorf("api %s status %s body %s", path, response.Status, body)
}
var baseResponse struct {
ErrCode int `json:"errCode"`
ErrMsg string `json:"errMsg"`
ErrDlt string `json:"errDlt"`
Data json.RawMessage `json:"data"`
}
if err := json.Unmarshal(body, &baseResponse); err != nil {
return err
}
if baseResponse.ErrCode != 0 {
return fmt.Errorf("api %s errCode %d errMsg %s errDlt %s", path, baseResponse.ErrCode, baseResponse.ErrMsg, baseResponse.ErrDlt)
}
if resp != nil {
if err := json.Unmarshal(baseResponse.Data, resp); err != nil {
return err
}
}
return nil
}
func (a *Api) GetToken(ctx context.Context) (string, error) {
req := auth.UserTokenReq{
UserID: a.UserID,
Secret: a.Secret,
PlatformID: constant.AdminPlatformID,
}
var resp auth.UserTokenResp
if err := a.apiPost(ctx, "/auth/user_token", &req, &resp); err != nil {
return "", err
}
return resp.Token, nil
}
func (a *Api) GetPartLimit(ctx context.Context) (*third.PartLimitResp, error) {
var resp third.PartLimitResp
if err := a.apiPost(ctx, "/object/part_limit", &third.PartLimitReq{}, &resp); err != nil {
return nil, err
}
return &resp, nil
}
func (a *Api) InitiateMultipartUpload(ctx context.Context, req *third.InitiateMultipartUploadReq) (*third.InitiateMultipartUploadResp, error) {
var resp third.InitiateMultipartUploadResp
if err := a.apiPost(ctx, "/object/initiate_multipart_upload", req, &resp); err != nil {
return nil, err
}
return &resp, nil
}
func (a *Api) CompleteMultipartUpload(ctx context.Context, req *third.CompleteMultipartUploadReq) (string, error) {
var resp third.CompleteMultipartUploadResp
if err := a.apiPost(ctx, "/object/complete_multipart_upload", req, &resp); err != nil {
return "", err
}
return resp.Url, nil
}
-96
View File
@@ -1,96 +0,0 @@
package pkg
import (
"bytes"
"io"
"os"
)
type ReadSeekSizeCloser interface {
io.ReadSeekCloser
Size() int64
}
func NewReader(r io.Reader, max int64, path string) (ReadSeekSizeCloser, error) {
buf := make([]byte, max+1)
n, err := io.ReadFull(r, buf)
if err == nil {
f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0666)
if err != nil {
return nil, err
}
var ok bool
defer func() {
if !ok {
_ = f.Close()
_ = os.Remove(path)
}
}()
if _, err := f.Write(buf[:n]); err != nil {
return nil, err
}
cn, err := io.Copy(f, r)
if err != nil {
return nil, err
}
if _, err := f.Seek(0, io.SeekStart); err != nil {
return nil, err
}
ok = true
return &fileBuffer{
f: f,
n: cn + int64(n),
}, nil
} else if err == io.EOF || err == io.ErrUnexpectedEOF {
return &memoryBuffer{
r: bytes.NewReader(buf[:n]),
}, nil
} else {
return nil, err
}
}
type fileBuffer struct {
n int64
f *os.File
}
func (r *fileBuffer) Read(p []byte) (n int, err error) {
return r.f.Read(p)
}
func (r *fileBuffer) Seek(offset int64, whence int) (int64, error) {
return r.f.Seek(offset, whence)
}
func (r *fileBuffer) Size() int64 {
return r.n
}
func (r *fileBuffer) Close() error {
name := r.f.Name()
if err := r.f.Close(); err != nil {
return err
}
return os.Remove(name)
}
type memoryBuffer struct {
r *bytes.Reader
}
func (r *memoryBuffer) Read(p []byte) (n int, err error) {
return r.r.Read(p)
}
func (r *memoryBuffer) Seek(offset int64, whence int) (int64, error) {
return r.r.Seek(offset, whence)
}
func (r *memoryBuffer) Close() error {
return nil
}
func (r *memoryBuffer) Size() int64 {
return r.r.Size()
}
-16
View File
@@ -1,16 +0,0 @@
package pkg
import "time"
type Config struct {
TaskPath string
ProgressPath string
Concurrency int
Retry int
Timeout time.Duration
Api string
UserID string
Secret string
TempDir string
CacheSize int64
}
-7
View File
@@ -1,7 +0,0 @@
package pkg
import "net/http"
func DefaultRequestHeader(header http.Header) {
header.Set("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36")
}
-385
View File
@@ -1,385 +0,0 @@
package pkg
import (
"bufio"
"context"
"crypto/md5"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"io"
"log"
"net/http"
"net/url"
"os"
"path/filepath"
"strconv"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/OpenIMSDK/protocol/third"
)
type Upload struct {
URL string `json:"url"`
Name string `json:"name"`
ContentType string `json:"contentType"`
}
type Task struct {
Index int
Upload Upload
}
type PartInfo struct {
ContentType string
PartSize int64
PartNum int
FileMd5 string
PartMd5 string
PartSizes []int64
PartMd5s []string
}
func Run(conf Config) error {
m := &Manage{
prefix: time.Now().Format("20060102150405"),
conf: &conf,
ctx: context.Background(),
}
return m.Run()
}
type Manage struct {
conf *Config
ctx context.Context
api *Api
partLimit *third.PartLimitResp
prefix string
tasks chan Task
id uint64
success int64
failed int64
}
func (m *Manage) tempFilePath() string {
return filepath.Join(m.conf.TempDir, fmt.Sprintf("%s_%d", m.prefix, atomic.AddUint64(&m.id, 1)))
}
func (m *Manage) Run() error {
defer func(start time.Time) {
log.Printf("run time %s\n", time.Since(start))
}(time.Now())
m.api = &Api{
Api: m.conf.Api,
UserID: m.conf.UserID,
Secret: m.conf.Secret,
Client: &http.Client{Timeout: m.conf.Timeout},
}
var err error
ctx := context.WithValue(m.ctx, "operationID", fmt.Sprintf("%s_init", m.prefix))
m.api.Token, err = m.api.GetToken(ctx)
if err != nil {
return err
}
m.partLimit, err = m.api.GetPartLimit(ctx)
if err != nil {
return err
}
progress, err := ReadProgress(m.conf.ProgressPath)
if err != nil {
return err
}
progressFile, err := os.OpenFile(m.conf.ProgressPath, os.O_CREATE|os.O_RDWR|os.O_APPEND, 0666)
if err != nil {
return err
}
var mutex sync.Mutex
writeSuccessIndex := func(index int) {
mutex.Lock()
defer mutex.Unlock()
if _, err := progressFile.Write([]byte(strconv.Itoa(index) + "\n")); err != nil {
log.Printf("write progress err: %v\n", err)
}
}
file, err := os.Open(m.conf.TaskPath)
if err != nil {
return err
}
m.tasks = make(chan Task, m.conf.Concurrency*2)
go func() {
defer file.Close()
defer close(m.tasks)
scanner := bufio.NewScanner(file)
var (
index int
num int
)
for scanner.Scan() {
line := strings.TrimSpace(scanner.Text())
if line == "" {
continue
}
index++
if progress.IsUploaded(index) {
log.Printf("index: %d already uploaded %s\n", index, line)
continue
}
var upload Upload
if err := json.Unmarshal([]byte(line), &upload); err != nil {
log.Printf("index: %d json.Unmarshal(%s) err: %v", index, line, err)
continue
}
num++
m.tasks <- Task{
Index: index,
Upload: upload,
}
}
if num == 0 {
log.Println("mark all completed")
}
}()
var wg sync.WaitGroup
wg.Add(m.conf.Concurrency)
for i := 0; i < m.conf.Concurrency; i++ {
go func(tid int) {
defer wg.Done()
for task := range m.tasks {
var success bool
for n := 0; n < m.conf.Retry; n++ {
ctx := context.WithValue(m.ctx, "operationID", fmt.Sprintf("%s_%d_%d_%d", m.prefix, tid, task.Index, n+1))
if urlRaw, err := m.RunTask(ctx, task); err == nil {
writeSuccessIndex(task.Index)
log.Println("index:", task.Index, "upload success", "urlRaw", urlRaw)
success = true
break
} else {
log.Printf("index: %d upload: %+v err: %v", task.Index, task.Upload, err)
}
}
if success {
atomic.AddInt64(&m.success, 1)
} else {
atomic.AddInt64(&m.failed, 1)
log.Printf("index: %d upload: %+v failed", task.Index, task.Upload)
}
}
}(i + 1)
}
wg.Wait()
log.Printf("execution completed success %d failed %d\n", m.success, m.failed)
return nil
}
func (m *Manage) RunTask(ctx context.Context, task Task) (string, error) {
resp, err := m.HttpGet(ctx, task.Upload.URL)
if err != nil {
return "", err
}
defer resp.Body.Close()
reader, err := NewReader(resp.Body, m.conf.CacheSize, m.tempFilePath())
if err != nil {
return "", err
}
defer reader.Close()
part, err := m.getPartInfo(ctx, reader, reader.Size())
if err != nil {
return "", err
}
var contentType string
if task.Upload.ContentType == "" {
contentType = part.ContentType
} else {
contentType = task.Upload.ContentType
}
initiateMultipartUploadResp, err := m.api.InitiateMultipartUpload(ctx, &third.InitiateMultipartUploadReq{
Hash: part.PartMd5,
Size: reader.Size(),
PartSize: part.PartSize,
MaxParts: -1,
Cause: "batch-import",
Name: task.Upload.Name,
ContentType: contentType,
})
if err != nil {
return "", err
}
if initiateMultipartUploadResp.Upload == nil {
return initiateMultipartUploadResp.Url, nil
}
if _, err := reader.Seek(0, io.SeekStart); err != nil {
return "", err
}
uploadParts := make([]*third.SignPart, part.PartNum)
for _, part := range initiateMultipartUploadResp.Upload.Sign.Parts {
uploadParts[part.PartNumber-1] = part
}
for i, currentPartSize := range part.PartSizes {
md5Reader := NewMd5Reader(io.LimitReader(reader, currentPartSize))
if m.doPut(ctx, m.api.Client, initiateMultipartUploadResp.Upload.Sign, uploadParts[i], md5Reader, currentPartSize); err != nil {
return "", err
}
if md5val := md5Reader.Md5(); md5val != part.PartMd5s[i] {
return "", fmt.Errorf("upload part %d failed, md5 not match, expect %s, got %s", i, part.PartMd5s[i], md5val)
}
}
urlRaw, err := m.api.CompleteMultipartUpload(ctx, &third.CompleteMultipartUploadReq{
UploadID: initiateMultipartUploadResp.Upload.UploadID,
Parts: part.PartMd5s,
Name: task.Upload.Name,
ContentType: contentType,
Cause: "batch-import",
})
if err != nil {
return "", err
}
return urlRaw, nil
}
func (m *Manage) partSize(size int64) (int64, error) {
if size <= 0 {
return 0, errors.New("size must be greater than 0")
}
if size > m.partLimit.MaxPartSize*int64(m.partLimit.MaxNumSize) {
return 0, fmt.Errorf("size must be less than %db", m.partLimit.MaxPartSize*int64(m.partLimit.MaxNumSize))
}
if size <= m.partLimit.MinPartSize*int64(m.partLimit.MaxNumSize) {
return m.partLimit.MinPartSize, nil
}
partSize := size / int64(m.partLimit.MaxNumSize)
if size%int64(m.partLimit.MaxNumSize) != 0 {
partSize++
}
return partSize, nil
}
func (m *Manage) partMD5(parts []string) string {
s := strings.Join(parts, ",")
md5Sum := md5.Sum([]byte(s))
return hex.EncodeToString(md5Sum[:])
}
func (m *Manage) getPartInfo(ctx context.Context, r io.Reader, fileSize int64) (*PartInfo, error) {
partSize, err := m.partSize(fileSize)
if err != nil {
return nil, err
}
partNum := int(fileSize / partSize)
if fileSize%partSize != 0 {
partNum++
}
partSizes := make([]int64, partNum)
for i := 0; i < partNum; i++ {
partSizes[i] = partSize
}
partSizes[partNum-1] = fileSize - partSize*(int64(partNum)-1)
partMd5s := make([]string, partNum)
buf := make([]byte, 1024*8)
fileMd5 := md5.New()
var contentType string
for i := 0; i < partNum; i++ {
h := md5.New()
r := io.LimitReader(r, partSize)
for {
if n, err := r.Read(buf); err == nil {
if contentType == "" {
contentType = http.DetectContentType(buf[:n])
}
h.Write(buf[:n])
fileMd5.Write(buf[:n])
} else if err == io.EOF {
break
} else {
return nil, err
}
}
partMd5s[i] = hex.EncodeToString(h.Sum(nil))
}
partMd5Val := m.partMD5(partMd5s)
fileMd5val := hex.EncodeToString(fileMd5.Sum(nil))
return &PartInfo{
ContentType: contentType,
PartSize: partSize,
PartNum: partNum,
FileMd5: fileMd5val,
PartMd5: partMd5Val,
PartSizes: partSizes,
PartMd5s: partMd5s,
}, nil
}
func (m *Manage) doPut(ctx context.Context, client *http.Client, sign *third.AuthSignParts, part *third.SignPart, reader io.Reader, size int64) error {
rawURL := part.Url
if rawURL == "" {
rawURL = sign.Url
}
if len(sign.Query)+len(part.Query) > 0 {
u, err := url.Parse(rawURL)
if err != nil {
return err
}
query := u.Query()
for i := range sign.Query {
v := sign.Query[i]
query[v.Key] = v.Values
}
for i := range part.Query {
v := part.Query[i]
query[v.Key] = v.Values
}
u.RawQuery = query.Encode()
rawURL = u.String()
}
req, err := http.NewRequestWithContext(ctx, http.MethodPut, rawURL, reader)
if err != nil {
return err
}
for i := range sign.Header {
v := sign.Header[i]
req.Header[v.Key] = v.Values
}
for i := range part.Header {
v := part.Header[i]
req.Header[v.Key] = v.Values
}
req.ContentLength = size
resp, err := client.Do(req)
if err != nil {
return err
}
defer func() {
_ = resp.Body.Close()
}()
body, err := io.ReadAll(resp.Body)
if err != nil {
return err
}
if resp.StatusCode/200 != 1 {
return fmt.Errorf("PUT %s part %d failed, status code %d, body %s", rawURL, part.PartNumber, resp.StatusCode, string(body))
}
return nil
}
func (m *Manage) HttpGet(ctx context.Context, url string) (*http.Response, error) {
reqUrl := url
for {
request, err := http.NewRequestWithContext(ctx, http.MethodGet, reqUrl, nil)
if err != nil {
return nil, err
}
DefaultRequestHeader(request.Header)
response, err := m.api.Client.Do(request)
if err != nil {
return nil, err
}
if response.StatusCode != http.StatusOK {
_ = response.Body.Close()
return nil, fmt.Errorf("http get %s status %s", url, response.Status)
}
return response, nil
}
}
-29
View File
@@ -1,29 +0,0 @@
package pkg
import (
"crypto/md5"
"encoding/hex"
"hash"
"io"
)
func NewMd5Reader(r io.Reader) *Md5Reader {
return &Md5Reader{h: md5.New(), r: r}
}
type Md5Reader struct {
h hash.Hash
r io.Reader
}
func (r *Md5Reader) Read(p []byte) (n int, err error) {
n, err = r.r.Read(p)
if err == nil && n > 0 {
r.h.Write(p[:n])
}
return
}
func (r *Md5Reader) Md5() string {
return hex.EncodeToString(r.h.Sum(nil))
}
-41
View File
@@ -1,41 +0,0 @@
package pkg
import (
"bufio"
"os"
"strconv"
"github.com/kelindar/bitmap"
)
func ReadProgress(path string) (*Progress, error) {
file, err := os.Open(path)
if err != nil {
if os.IsNotExist(err) {
return &Progress{}, nil
}
return nil, err
}
defer file.Close()
scanner := bufio.NewScanner(file)
var upload bitmap.Bitmap
for scanner.Scan() {
index, err := strconv.Atoi(scanner.Text())
if err != nil || index < 0 {
continue
}
upload.Set(uint32(index))
}
return &Progress{upload: upload}, nil
}
type Progress struct {
upload bitmap.Bitmap
}
func (p *Progress) IsUploaded(index int) bool {
if p == nil {
return false
}
return p.upload.Contains(uint32(index))
}