mirror of
https://github.com/openimsdk/open-im-server.git
synced 2026-07-14 18:11:12 +08:00
Compare commits
17 Commits
37d869a74f
...
06abc58b50
| Author | SHA1 | Date | |
|---|---|---|---|
|
06abc58b50 | ||
|
dd981b976d | ||
|
91db18168d | ||
|
bed81af75c | ||
|
|
986237af2e | ||
|
|
c54b948f1a | ||
|
fa3d251dcb | ||
|
|
58994095a5 | ||
|
|
97b8c07d57 | ||
|
|
045afd74b9 | ||
|
|
8e824c7e8e | ||
|
|
52bd5e8be9 | ||
|
|
1697651100 | ||
|
|
d385fdd0aa | ||
|
|
aca0eac955 | ||
|
|
3b710fdfdb | ||
|
|
3be343be57 |
@ -28,6 +28,8 @@ run:
|
||||
# - util
|
||||
# - .*~
|
||||
# - api/swagger/docs
|
||||
|
||||
|
||||
# - server/docs
|
||||
# - components/mnt/config/certs
|
||||
# - logs
|
||||
|
||||
4
CHANGELOG/CHANGELOG-3.8.md
Normal file
4
CHANGELOG/CHANGELOG-3.8.md
Normal file
@ -0,0 +1,4 @@
|
||||
## [v3.8.3](https://github.com/openimsdk/open-im-server/releases/tag/v3.8.3) (2025-01-14)
|
||||
|
||||
**Full Changelog**: [v3.8.2...v3.8.3](https://github.com/openimsdk/open-im-server/compare/v3.8.2...v3.8.3)
|
||||
|
||||
4
CHANGELOG/README.md
Normal file
4
CHANGELOG/README.md
Normal file
@ -0,0 +1,4 @@
|
||||
# CHANGELOGs
|
||||
|
||||
- [CHANGELOG-3.8.md](./CHANGELOG-3.8.md)
|
||||
|
||||
210
LICENSE
210
LICENSE
@ -1,35 +1,201 @@
|
||||
# Open Source License
|
||||
Apache License
|
||||
Version 2.0, January 2004
|
||||
http://www.apache.org/licenses/
|
||||
|
||||
OpenIM is licensed under the Apache License 2.0, with the following additional conditions:
|
||||
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
|
||||
|
||||
1. OpenIM may be utilized commercially, including as a backend service for other applications or as an application development platform for enterprises.
|
||||
A commercial license must be obtained from the producer if:
|
||||
- Under no circumstances may you operate a multi-tenant or multi-business environment using the OpenIM source code, whether or not you have modified the repository code. In other words, a single instance of OpenIM may not simultaneously serve multiple enterprises, nor may it serve multiple lines of business within the same enterprise.
|
||||
- If you intend to operate in such a multi-tenant or multi-business manner, you must obtain a commercial license from the producer in advance.
|
||||
1. Definitions.
|
||||
|
||||
2. As a contributor, you should agree that:
|
||||
"License" shall mean the terms and conditions for use, reproduction,
|
||||
and distribution as defined by Sections 1 through 9 of this document.
|
||||
|
||||
a. The producer can adjust the open-source agreement to be more strict or more relaxed as deemed necessary.
|
||||
b. Your contributed code may be used for commercial purposes, including but not limited to its cloud business operations.
|
||||
"Licensor" shall mean the copyright owner or entity authorized by
|
||||
the copyright owner that is granting the License.
|
||||
|
||||
Apart from the specific conditions mentioned above, all other rights and restrictions follow the Apache License 2.0. Detailed information about the Apache License 2.0 can be found at http://www.apache.org/licenses/LICENSE-2.0.
|
||||
"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.
|
||||
|
||||
For any licensing-related questions or to obtain a commercial license, please contact contact@openim.io.
|
||||
"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.
|
||||
|
||||
© 2024 OpenIMSDK
|
||||
"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.
|
||||
|
||||
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
|
||||
"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."
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
"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.
|
||||
|
||||
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.
|
||||
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.
|
||||
|
||||
@ -132,7 +132,7 @@ Thank you for contributing to building a powerful instant messaging solution!
|
||||
|
||||
## :closed_book: License
|
||||
|
||||
For more details, please refer to [here](./LICENSE).
|
||||
This software is licensed under the Apache License 2.0
|
||||
|
||||
|
||||
|
||||
|
||||
@ -131,9 +131,10 @@
|
||||
|
||||
感谢您的贡献，一起来打造强大的即时通讯解决方案！
|
||||
|
||||
## :closed_book: 许可证
|
||||
## :closed_book: 开源许可证 License
|
||||
|
||||
This software is licensed under the Apache License 2.0
|
||||
|
||||
OpenIMSDK 在 Apache License 2.0 许可下可用。查看[LICENSE 文件](https://github.com/openimsdk/open-im-server/blob/main/LICENSE)了解更多信息。
|
||||
|
||||
## 🔮 Thanks to our contributors!
|
||||
|
||||
|
||||
@ -7,3 +7,7 @@ multiLogin:
|
||||
policy: 1
|
||||
# max num of tokens in one end
|
||||
maxNumOneEnd: 30
|
||||
|
||||
rpcMaxBodySize:
|
||||
requestMaxBodySize: 8388608
|
||||
responseMaxBodySize: 8388608
|
||||
|
||||
4
go.mod
4
go.mod
@ -12,8 +12,8 @@ require (
|
||||
github.com/gorilla/websocket v1.5.1
|
||||
github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
|
||||
github.com/mitchellh/mapstructure v1.5.0
|
||||
github.com/openimsdk/protocol v0.0.72-alpha.81
|
||||
github.com/openimsdk/tools v0.0.50-alpha.74
|
||||
github.com/openimsdk/protocol v0.0.73-alpha.6
|
||||
github.com/openimsdk/tools v0.0.50-alpha.79
|
||||
github.com/pkg/errors v0.9.1 // indirect
|
||||
github.com/prometheus/client_golang v1.18.0
|
||||
github.com/stretchr/testify v1.9.0
|
||||
|
||||
8
go.sum
8
go.sum
@ -347,10 +347,10 @@ github.com/onsi/gomega v1.25.0 h1:Vw7br2PCDYijJHSfBOWhov+8cAnUf8MfMaIOV323l6Y=
|
||||
github.com/onsi/gomega v1.25.0/go.mod h1:r+zV744Re+DiYCIPRlYOTxn0YkOLcAnW8k1xXdMPGhM=
|
||||
github.com/openimsdk/gomake v0.0.15-alpha.2 h1:5Q8yl8ezy2yx+q8/ucU/t4kJnDfCzNOrkXcDACCqtyM=
|
||||
github.com/openimsdk/gomake v0.0.15-alpha.2/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
|
||||
github.com/openimsdk/protocol v0.0.72-alpha.81 h1:6tDuZ3Anfi1uhX/V5mWxITqJnGQPnvgeaxeqJlEHIVE=
|
||||
github.com/openimsdk/protocol v0.0.72-alpha.81/go.mod h1:WF7EuE55vQvpyUAzDXcqg+B+446xQyEba0X35lTINmw=
|
||||
github.com/openimsdk/tools v0.0.50-alpha.74 h1:yh10SiMiivMEjicEQg+QAsH4pvaO+4noMPdlw+ew0Kc=
|
||||
github.com/openimsdk/tools v0.0.50-alpha.74/go.mod h1:n2poR3asX1e1XZce4O+MOWAp+X02QJRFvhcLCXZdzRo=
|
||||
github.com/openimsdk/protocol v0.0.73-alpha.6 h1:sna9coWG7HN1zObBPtvG0Ki/vzqHXiB4qKbA5P3w7kc=
|
||||
github.com/openimsdk/protocol v0.0.73-alpha.6/go.mod h1:WF7EuE55vQvpyUAzDXcqg+B+446xQyEba0X35lTINmw=
|
||||
github.com/openimsdk/tools v0.0.50-alpha.79 h1:jxYEbrzaze4Z2r4NrKad816buZ690ix0L9MTOOOH3ik=
|
||||
github.com/openimsdk/tools v0.0.50-alpha.79/go.mod h1:n2poR3asX1e1XZce4O+MOWAp+X02QJRFvhcLCXZdzRo=
|
||||
github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6Wq+LM=
|
||||
github.com/pelletier/go-toml/v2 v2.2.2/go.mod h1:1t835xjRzz80PqgE6HHgN2JOsmgYu/h4qDAS4n929Rs=
|
||||
github.com/pierrec/lz4/v4 v4.1.21 h1:yOVMLb6qSIDP67pl/5F7RepeKYu/VmTyEXvuMI5d9mQ=
|
||||
|
||||
@ -90,7 +90,7 @@ func Start(ctx context.Context, config *Config, client discovery.Conn, service g
|
||||
//case <-ctx.Done():
|
||||
//}
|
||||
<-apiCtx.Done()
|
||||
exitCause := context.Cause(ctx)
|
||||
exitCause := context.Cause(apiCtx)
|
||||
log.ZWarn(ctx, "api server exit", exitCause)
|
||||
timer := time.NewTimer(time.Second * 15)
|
||||
defer timer.Stop()
|
||||
|
||||
@ -125,6 +125,11 @@ func newGinRouter(ctx context.Context, client discovery.Conn, cfg *Config) (*gin
|
||||
userRouterGroup.POST("/add_notification_account", u.AddNotificationAccount)
|
||||
userRouterGroup.POST("/update_notification_account", u.UpdateNotificationAccountInfo)
|
||||
userRouterGroup.POST("/search_notification_account", u.SearchNotificationAccount)
|
||||
|
||||
userRouterGroup.POST("/get_user_client_config", u.GetUserClientConfig)
|
||||
userRouterGroup.POST("/set_user_client_config", u.SetUserClientConfig)
|
||||
userRouterGroup.POST("/del_user_client_config", u.DelUserClientConfig)
|
||||
userRouterGroup.POST("/page_user_client_config", u.PageUserClientConfig)
|
||||
}
|
||||
// friend routing group
|
||||
{
|
||||
|
||||
@ -242,3 +242,19 @@ func (u *UserApi) UpdateNotificationAccountInfo(c *gin.Context) {
|
||||
func (u *UserApi) SearchNotificationAccount(c *gin.Context) {
|
||||
a2r.Call(c, user.UserClient.SearchNotificationAccount, u.Client)
|
||||
}
|
||||
|
||||
func (u *UserApi) GetUserClientConfig(c *gin.Context) {
|
||||
a2r.Call(c, user.UserClient.GetUserClientConfig, u.Client)
|
||||
}
|
||||
|
||||
func (u *UserApi) SetUserClientConfig(c *gin.Context) {
|
||||
a2r.Call(c, user.UserClient.SetUserClientConfig, u.Client)
|
||||
}
|
||||
|
||||
func (u *UserApi) DelUserClientConfig(c *gin.Context) {
|
||||
a2r.Call(c, user.UserClient.DelUserClientConfig, u.Client)
|
||||
}
|
||||
|
||||
func (u *UserApi) PageUserClientConfig(c *gin.Context) {
|
||||
a2r.Call(c, user.UserClient.PageUserClientConfig, u.Client)
|
||||
}
|
||||
|
||||
@ -448,12 +448,25 @@ func (g *groupServer) InviteUserToGroup(ctx context.Context, req *pbgroup.Invite
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if err := g.db.CreateGroup(ctx, nil, groupMembers); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
const singleQuantity = 50
|
||||
for start := 0; start < len(groupMembers); start += singleQuantity {
|
||||
end := start + singleQuantity
|
||||
if end > len(groupMembers) {
|
||||
end = len(groupMembers)
|
||||
}
|
||||
currentMembers := groupMembers[start:end]
|
||||
|
||||
if err = g.notification.GroupApplicationAgreeMemberEnterNotification(ctx, req.GroupID, req.SendMessage, opUserID, req.InvitedUserIDs...); err != nil {
|
||||
return nil, err
|
||||
if err := g.db.CreateGroup(ctx, nil, currentMembers); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
userIDs := datautil.Slice(currentMembers, func(e *model.GroupMember) string {
|
||||
return e.UserID
|
||||
})
|
||||
|
||||
if err = g.notification.GroupApplicationAgreeMemberEnterNotification(ctx, req.GroupID, req.SendMessage, opUserID, userIDs...); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return &pbgroup.InviteUserToGroupResp{}, nil
|
||||
}
|
||||
@ -1370,6 +1383,7 @@ func (g *groupServer) DismissGroup(ctx context.Context, req *pbgroup.DismissGrou
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
group.Status = constant.GroupStatusDismissed
|
||||
tips := &sdkws.GroupDismissedTips{
|
||||
Group: g.groupDB2PB(group, owner.UserID, num),
|
||||
OpUser: &sdkws.GroupMemberFullInfo{},
|
||||
|
||||
@ -284,7 +284,8 @@ func (g *NotificationSender) fillUserByUserID(ctx context.Context, userID string
|
||||
|
||||
func (g *NotificationSender) setVersion(ctx context.Context, version *uint64, versionID *string, collName string, id string) {
|
||||
versions := versionctx.GetVersionLog(ctx).Get()
|
||||
for _, coll := range versions {
|
||||
for i := len(versions) - 1; i >= 0; i-- {
|
||||
coll := versions[i]
|
||||
if coll.Name == collName && coll.Doc.DID == id {
|
||||
*version = uint64(coll.Doc.Version)
|
||||
*versionID = coll.Doc.ID.Hex()
|
||||
@ -521,6 +522,10 @@ func (g *NotificationSender) MemberKickedNotification(ctx context.Context, tips
|
||||
}
|
||||
|
||||
func (g *NotificationSender) GroupApplicationAgreeMemberEnterNotification(ctx context.Context, groupID string, SendMessage *bool, invitedOpUserID string, entrantUserID ...string) error {
|
||||
return g.groupApplicationAgreeMemberEnterNotification(ctx, groupID, SendMessage, invitedOpUserID, entrantUserID...)
|
||||
}
|
||||
|
||||
func (g *NotificationSender) groupApplicationAgreeMemberEnterNotification(ctx context.Context, groupID string, SendMessage *bool, invitedOpUserID string, entrantUserID ...string) error {
|
||||
var err error
|
||||
defer func() {
|
||||
if err != nil {
|
||||
|
||||
@ -61,6 +61,13 @@ func (m *msgServer) GetConversationsHasReadAndMaxSeq(ctx context.Context, req *m
|
||||
return nil, err
|
||||
}
|
||||
resp := &msg.GetConversationsHasReadAndMaxSeqResp{Seqs: make(map[string]*msg.Seqs)}
|
||||
if req.ReturnPinned {
|
||||
pinnedConversationIDs, err := m.ConversationLocalCache.GetPinnedConversationIDs(ctx, req.UserID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
resp.PinnedConversationIDs = pinnedConversationIDs
|
||||
}
|
||||
for conversationID, maxSeq := range maxSeqs {
|
||||
resp.Seqs[conversationID] = &msg.Seqs{
|
||||
HasReadSeq: hasReadSeqs[conversationID],
|
||||
|
||||
@ -62,7 +62,7 @@ func (t *thirdServer) InitiateMultipartUpload(ctx context.Context, req *third.In
|
||||
return nil, err
|
||||
}
|
||||
expireTime := time.Now().Add(t.defaultExpire)
|
||||
result, err := t.s3dataBase.InitiateMultipartUpload(ctx, req.Hash, req.Size, t.defaultExpire, int(req.MaxParts))
|
||||
result, err := t.s3dataBase.InitiateMultipartUpload(ctx, req.Hash, req.Size, t.defaultExpire, int(req.MaxParts), req.ContentType)
|
||||
if err != nil {
|
||||
if haErr, ok := errs.Unwrap(err).(*cont.HashAlreadyExistsError); ok {
|
||||
obj := &model.Object{
|
||||
|
||||
71
internal/rpc/user/config.go
Normal file
71
internal/rpc/user/config.go
Normal file
@ -0,0 +1,71 @@
|
||||
package user
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/authverify"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
pbuser "github.com/openimsdk/protocol/user"
|
||||
"github.com/openimsdk/tools/utils/datautil"
|
||||
)
|
||||
|
||||
func (s *userServer) GetUserClientConfig(ctx context.Context, req *pbuser.GetUserClientConfigReq) (*pbuser.GetUserClientConfigResp, error) {
|
||||
if req.UserID != "" {
|
||||
if err := authverify.CheckAccessV3(ctx, req.UserID, s.config.Share.IMAdminUserID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if _, err := s.db.GetUserByID(ctx, req.UserID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
res, err := s.clientConfig.GetUserConfig(ctx, req.UserID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &pbuser.GetUserClientConfigResp{Configs: res}, nil
|
||||
}
|
||||
|
||||
func (s *userServer) SetUserClientConfig(ctx context.Context, req *pbuser.SetUserClientConfigReq) (*pbuser.SetUserClientConfigResp, error) {
|
||||
if err := authverify.CheckAdmin(ctx, s.config.Share.IMAdminUserID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if req.UserID != "" {
|
||||
if _, err := s.db.GetUserByID(ctx, req.UserID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
if err := s.clientConfig.SetUserConfig(ctx, req.UserID, req.Configs); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &pbuser.SetUserClientConfigResp{}, nil
|
||||
}
|
||||
|
||||
func (s *userServer) DelUserClientConfig(ctx context.Context, req *pbuser.DelUserClientConfigReq) (*pbuser.DelUserClientConfigResp, error) {
|
||||
if err := authverify.CheckAdmin(ctx, s.config.Share.IMAdminUserID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := s.clientConfig.DelUserConfig(ctx, req.UserID, req.Keys); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &pbuser.DelUserClientConfigResp{}, nil
|
||||
}
|
||||
|
||||
func (s *userServer) PageUserClientConfig(ctx context.Context, req *pbuser.PageUserClientConfigReq) (*pbuser.PageUserClientConfigResp, error) {
|
||||
if err := authverify.CheckAdmin(ctx, s.config.Share.IMAdminUserID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
total, res, err := s.clientConfig.GetUserConfigPage(ctx, req.UserID, req.Key, req.Pagination)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &pbuser.PageUserClientConfigResp{
|
||||
Total: total,
|
||||
Configs: datautil.Slice(res, func(e *model.ClientConfig) *pbuser.ClientConfig {
|
||||
return &pbuser.ClientConfig{
|
||||
UserID: e.UserID,
|
||||
Key: e.Key,
|
||||
Value: e.Value,
|
||||
}
|
||||
}),
|
||||
}, nil
|
||||
}
|
||||
@ -64,6 +64,7 @@ type userServer struct {
|
||||
webhookClient *webhook.Client
|
||||
groupClient *rpcli.GroupClient
|
||||
relationClient *rpcli.RelationClient
|
||||
clientConfig controller.ClientConfigDatabase
|
||||
}
|
||||
|
||||
type Config struct {
|
||||
@ -98,6 +99,10 @@ func Start(ctx context.Context, config *Config, client discovery.Conn, server gr
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
clientConfigDB, err := mgo.NewClientConfig(mgocli.GetDB())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
msgConn, err := client.GetConn(ctx, config.Discovery.RpcService.Msg)
|
||||
if err != nil {
|
||||
return err
|
||||
@ -122,9 +127,9 @@ func Start(ctx context.Context, config *Config, client discovery.Conn, server gr
|
||||
userNotificationSender: NewUserNotificationSender(config, msgClient, WithUserFunc(database.FindWithError)),
|
||||
config: config,
|
||||
webhookClient: webhook.NewWebhookClient(config.WebhooksConfig.URL),
|
||||
|
||||
groupClient: rpcli.NewGroupClient(groupConn),
|
||||
relationClient: rpcli.NewRelationClient(friendConn),
|
||||
clientConfig: controller.NewClientConfigDatabase(clientConfigDB, redis.NewClientConfigCache(rdb, clientConfigDB), mgocli.GetTx()),
|
||||
groupClient: rpcli.NewGroupClient(groupConn),
|
||||
relationClient: rpcli.NewRelationClient(friendConn),
|
||||
}
|
||||
pbuser.RegisterUserServer(server, u)
|
||||
return u.db.InitOnce(context.Background(), users)
|
||||
|
||||
@ -348,9 +348,15 @@ type AfterConfig struct {
|
||||
}
|
||||
|
||||
type Share struct {
|
||||
Secret string `yaml:"secret"`
|
||||
IMAdminUserID []string `yaml:"imAdminUserID"`
|
||||
MultiLogin MultiLogin `yaml:"multiLogin"`
|
||||
Secret string `yaml:"secret"`
|
||||
IMAdminUserID []string `yaml:"imAdminUserID"`
|
||||
MultiLogin MultiLogin `yaml:"multiLogin"`
|
||||
RPCMaxBodySize MaxRequestBody `yaml:"rpcMaxBodySize"`
|
||||
}
|
||||
|
||||
type MaxRequestBody struct {
|
||||
RequestMaxBodySize int `yaml:"requestMaxBodySize"`
|
||||
ResponseMaxBodySize int `yaml:"responseMaxBodySize"`
|
||||
}
|
||||
|
||||
type MultiLogin struct {
|
||||
|
||||
@ -21,6 +21,7 @@ import (
|
||||
"net"
|
||||
"os"
|
||||
"os/signal"
|
||||
"reflect"
|
||||
"strconv"
|
||||
"syscall"
|
||||
"time"
|
||||
@ -45,6 +46,36 @@ func init() {
|
||||
prommetrics.RegistryAll()
|
||||
}
|
||||
|
||||
func getConfigRpcMaxRequestBody(value reflect.Value) *conf.MaxRequestBody {
|
||||
for value.Kind() == reflect.Pointer {
|
||||
value = value.Elem()
|
||||
}
|
||||
if value.Kind() == reflect.Struct {
|
||||
num := value.NumField()
|
||||
for i := 0; i < num; i++ {
|
||||
field := value.Field(i)
|
||||
if !field.CanInterface() {
|
||||
continue
|
||||
}
|
||||
for field.Kind() == reflect.Pointer {
|
||||
field = field.Elem()
|
||||
}
|
||||
switch elem := field.Interface().(type) {
|
||||
case conf.Share:
|
||||
return &elem.RPCMaxBodySize
|
||||
case conf.MaxRequestBody:
|
||||
return &elem
|
||||
}
|
||||
if field.Kind() == reflect.Struct {
|
||||
if elem := getConfigRpcMaxRequestBody(field); elem != nil {
|
||||
return elem
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func Start[T any](ctx context.Context, disc *conf.Discovery, prometheusConfig *conf.Prometheus, listenIP,
|
||||
registerIP string, autoSetPorts bool, rpcPorts []int, index int, rpcRegisterName string, notification *conf.Notification, config T,
|
||||
watchConfigNames []string, watchServiceNames []string,
|
||||
@ -55,7 +86,24 @@ func Start[T any](ctx context.Context, disc *conf.Discovery, prometheusConfig *c
|
||||
conf.InitNotification(notification)
|
||||
}
|
||||
|
||||
options = append(options, mw.GrpcServer())
|
||||
maxRequestBody := getConfigRpcMaxRequestBody(reflect.ValueOf(config))
|
||||
|
||||
log.ZDebug(ctx, "rpc start", "rpcMaxRequestBody", maxRequestBody, "rpcRegisterName", rpcRegisterName, "registerIP", registerIP, "listenIP", listenIP)
|
||||
|
||||
options = append(options,
|
||||
mw.GrpcServer(),
|
||||
)
|
||||
var clientOptions []grpc.DialOption
|
||||
if maxRequestBody != nil {
|
||||
if maxRequestBody.RequestMaxBodySize > 0 {
|
||||
options = append(options, grpc.MaxRecvMsgSize(maxRequestBody.RequestMaxBodySize))
|
||||
clientOptions = append(clientOptions, grpc.WithDefaultCallOptions(grpc.MaxCallSendMsgSize(maxRequestBody.RequestMaxBodySize)))
|
||||
}
|
||||
if maxRequestBody.ResponseMaxBodySize > 0 {
|
||||
options = append(options, grpc.MaxSendMsgSize(maxRequestBody.ResponseMaxBodySize))
|
||||
clientOptions = append(clientOptions, grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxRequestBody.ResponseMaxBodySize)))
|
||||
}
|
||||
}
|
||||
|
||||
registerIP, err := network.GetRpcRegisterIP(registerIP)
|
||||
if err != nil {
|
||||
@ -84,6 +132,9 @@ func Start[T any](ctx context.Context, disc *conf.Discovery, prometheusConfig *c
|
||||
mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()),
|
||||
grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")),
|
||||
)
|
||||
if len(clientOptions) > 0 {
|
||||
client.AddOption(clientOptions...)
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancelCause(ctx)
|
||||
|
||||
|
||||
10
pkg/common/storage/cache/cachekey/client_config.go
vendored
Normal file
10
pkg/common/storage/cache/cachekey/client_config.go
vendored
Normal file
@ -0,0 +1,10 @@
|
||||
package cachekey
|
||||
|
||||
const ClientConfig = "CLIENT_CONFIG"
|
||||
|
||||
func GetClientConfigKey(userID string) string {
|
||||
if userID == "" {
|
||||
return ClientConfig
|
||||
}
|
||||
return ClientConfig + ":" + userID
|
||||
}
|
||||
8
pkg/common/storage/cache/client_config.go
vendored
Normal file
8
pkg/common/storage/cache/client_config.go
vendored
Normal file
@ -0,0 +1,8 @@
|
||||
package cache
|
||||
|
||||
import "context"
|
||||
|
||||
type ClientConfigCache interface {
|
||||
DeleteUserCache(ctx context.Context, userIDs []string) error
|
||||
GetUserConfig(ctx context.Context, userID string) (map[string]string, error)
|
||||
}
|
||||
69
pkg/common/storage/cache/redis/client_config.go
vendored
Normal file
69
pkg/common/storage/cache/redis/client_config.go
vendored
Normal file
@ -0,0 +1,69 @@
|
||||
package redis
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
func NewClientConfigCache(rdb redis.UniversalClient, mgo database.ClientConfig) cache.ClientConfigCache {
|
||||
rc := newRocksCacheClient(rdb)
|
||||
return &ClientConfigCache{
|
||||
mgo: mgo,
|
||||
rcClient: rc,
|
||||
delete: rc.GetBatchDeleter(),
|
||||
}
|
||||
}
|
||||
|
||||
type ClientConfigCache struct {
|
||||
mgo database.ClientConfig
|
||||
rcClient *rocksCacheClient
|
||||
delete cache.BatchDeleter
|
||||
}
|
||||
|
||||
func (x *ClientConfigCache) getExpireTime(userID string) time.Duration {
|
||||
if userID == "" {
|
||||
return time.Hour * 24
|
||||
} else {
|
||||
return time.Hour
|
||||
}
|
||||
}
|
||||
|
||||
func (x *ClientConfigCache) getClientConfigKey(userID string) string {
|
||||
return cachekey.GetClientConfigKey(userID)
|
||||
}
|
||||
|
||||
func (x *ClientConfigCache) GetConfig(ctx context.Context, userID string) (map[string]string, error) {
|
||||
return getCache(ctx, x.rcClient, x.getClientConfigKey(userID), x.getExpireTime(userID), func(ctx context.Context) (map[string]string, error) {
|
||||
return x.mgo.Get(ctx, userID)
|
||||
})
|
||||
}
|
||||
|
||||
func (x *ClientConfigCache) DeleteUserCache(ctx context.Context, userIDs []string) error {
|
||||
keys := make([]string, 0, len(userIDs))
|
||||
for _, userID := range userIDs {
|
||||
keys = append(keys, x.getClientConfigKey(userID))
|
||||
}
|
||||
return x.delete.ExecDelWithKeys(ctx, keys)
|
||||
}
|
||||
|
||||
func (x *ClientConfigCache) GetUserConfig(ctx context.Context, userID string) (map[string]string, error) {
|
||||
config, err := x.GetConfig(ctx, "")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if userID != "" {
|
||||
userConfig, err := x.GetConfig(ctx, userID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for k, v := range userConfig {
|
||||
config[k] = v
|
||||
}
|
||||
}
|
||||
return config, nil
|
||||
}
|
||||
58
pkg/common/storage/controller/client_config.go
Normal file
58
pkg/common/storage/controller/client_config.go
Normal file
@ -0,0 +1,58 @@
|
||||
package controller
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
"github.com/openimsdk/tools/db/pagination"
|
||||
"github.com/openimsdk/tools/db/tx"
|
||||
)
|
||||
|
||||
type ClientConfigDatabase interface {
|
||||
SetUserConfig(ctx context.Context, userID string, config map[string]string) error
|
||||
GetUserConfig(ctx context.Context, userID string) (map[string]string, error)
|
||||
DelUserConfig(ctx context.Context, userID string, keys []string) error
|
||||
GetUserConfigPage(ctx context.Context, userID string, key string, pagination pagination.Pagination) (int64, []*model.ClientConfig, error)
|
||||
}
|
||||
|
||||
func NewClientConfigDatabase(db database.ClientConfig, cache cache.ClientConfigCache, tx tx.Tx) ClientConfigDatabase {
|
||||
return &clientConfigDatabase{
|
||||
tx: tx,
|
||||
db: db,
|
||||
cache: cache,
|
||||
}
|
||||
}
|
||||
|
||||
type clientConfigDatabase struct {
|
||||
tx tx.Tx
|
||||
db database.ClientConfig
|
||||
cache cache.ClientConfigCache
|
||||
}
|
||||
|
||||
func (x *clientConfigDatabase) SetUserConfig(ctx context.Context, userID string, config map[string]string) error {
|
||||
return x.tx.Transaction(ctx, func(ctx context.Context) error {
|
||||
if err := x.db.Set(ctx, userID, config); err != nil {
|
||||
return err
|
||||
}
|
||||
return x.cache.DeleteUserCache(ctx, []string{userID})
|
||||
})
|
||||
}
|
||||
|
||||
func (x *clientConfigDatabase) GetUserConfig(ctx context.Context, userID string) (map[string]string, error) {
|
||||
return x.cache.GetUserConfig(ctx, userID)
|
||||
}
|
||||
|
||||
func (x *clientConfigDatabase) DelUserConfig(ctx context.Context, userID string, keys []string) error {
|
||||
return x.tx.Transaction(ctx, func(ctx context.Context) error {
|
||||
if err := x.db.Del(ctx, userID, keys); err != nil {
|
||||
return err
|
||||
}
|
||||
return x.cache.DeleteUserCache(ctx, []string{userID})
|
||||
})
|
||||
}
|
||||
|
||||
func (x *clientConfigDatabase) GetUserConfigPage(ctx context.Context, userID string, key string, pagination pagination.Pagination) (int64, []*model.ClientConfig, error) {
|
||||
return x.db.GetPage(ctx, userID, key, pagination)
|
||||
}
|
||||
@ -33,7 +33,7 @@ type S3Database interface {
|
||||
PartLimit() (*s3.PartLimit, error)
|
||||
PartSize(ctx context.Context, size int64) (int64, error)
|
||||
AuthSign(ctx context.Context, uploadID string, partNumbers []int) (*s3.AuthSignResult, error)
|
||||
InitiateMultipartUpload(ctx context.Context, hash string, size int64, expire time.Duration, maxParts int) (*cont.InitiateUploadResult, error)
|
||||
InitiateMultipartUpload(ctx context.Context, hash string, size int64, expire time.Duration, maxParts int, contentType string) (*cont.InitiateUploadResult, error)
|
||||
CompleteMultipartUpload(ctx context.Context, uploadID string, parts []string) (*cont.UploadResult, error)
|
||||
AccessURL(ctx context.Context, name string, expire time.Duration, opt *s3.AccessURLOption) (time.Time, string, error)
|
||||
SetObject(ctx context.Context, info *model.Object) error
|
||||
@ -73,8 +73,8 @@ func (s *s3Database) AuthSign(ctx context.Context, uploadID string, partNumbers
|
||||
return s.s3.AuthSign(ctx, uploadID, partNumbers)
|
||||
}
|
||||
|
||||
func (s *s3Database) InitiateMultipartUpload(ctx context.Context, hash string, size int64, expire time.Duration, maxParts int) (*cont.InitiateUploadResult, error) {
|
||||
return s.s3.InitiateUpload(ctx, hash, size, expire, maxParts)
|
||||
func (s *s3Database) InitiateMultipartUpload(ctx context.Context, hash string, size int64, expire time.Duration, maxParts int, contentType string) (*cont.InitiateUploadResult, error) {
|
||||
return s.s3.InitiateUploadContentType(ctx, hash, size, expire, maxParts, contentType)
|
||||
}
|
||||
|
||||
func (s *s3Database) CompleteMultipartUpload(ctx context.Context, uploadID string, parts []string) (*cont.UploadResult, error) {
|
||||
|
||||
15
pkg/common/storage/database/client_config.go
Normal file
15
pkg/common/storage/database/client_config.go
Normal file
@ -0,0 +1,15 @@
|
||||
package database
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
"github.com/openimsdk/tools/db/pagination"
|
||||
)
|
||||
|
||||
type ClientConfig interface {
|
||||
Set(ctx context.Context, userID string, config map[string]string) error
|
||||
Get(ctx context.Context, userID string) (map[string]string, error)
|
||||
Del(ctx context.Context, userID string, keys []string) error
|
||||
GetPage(ctx context.Context, userID string, key string, pagination pagination.Pagination) (int64, []*model.ClientConfig, error)
|
||||
}
|
||||
99
pkg/common/storage/database/mgo/client_config.go
Normal file
99
pkg/common/storage/database/mgo/client_config.go
Normal file
@ -0,0 +1,99 @@
|
||||
// Copyright © 2023 OpenIM open source community. 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 mgo
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/database"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/model"
|
||||
"github.com/openimsdk/tools/db/mongoutil"
|
||||
"github.com/openimsdk/tools/db/pagination"
|
||||
"go.mongodb.org/mongo-driver/bson"
|
||||
"go.mongodb.org/mongo-driver/mongo"
|
||||
"go.mongodb.org/mongo-driver/mongo/options"
|
||||
|
||||
"github.com/openimsdk/tools/errs"
|
||||
)
|
||||
|
||||
func NewClientConfig(db *mongo.Database) (database.ClientConfig, error) {
|
||||
coll := db.Collection("config")
|
||||
_, err := coll.Indexes().CreateMany(context.Background(), []mongo.IndexModel{
|
||||
{
|
||||
Keys: bson.D{
|
||||
{Key: "key", Value: 1},
|
||||
{Key: "user_id", Value: 1},
|
||||
},
|
||||
Options: options.Index().SetUnique(true),
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
return nil, errs.Wrap(err)
|
||||
}
|
||||
return &ClientConfig{
|
||||
coll: coll,
|
||||
}, nil
|
||||
}
|
||||
|
||||
type ClientConfig struct {
|
||||
coll *mongo.Collection
|
||||
}
|
||||
|
||||
func (x *ClientConfig) Set(ctx context.Context, userID string, config map[string]string) error {
|
||||
if len(config) == 0 {
|
||||
return nil
|
||||
}
|
||||
for key, value := range config {
|
||||
filter := bson.M{"key": key, "user_id": userID}
|
||||
update := bson.M{
|
||||
"value": value,
|
||||
}
|
||||
err := mongoutil.UpdateOne(ctx, x.coll, filter, bson.M{"$set": update}, false, options.Update().SetUpsert(true))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (x *ClientConfig) Get(ctx context.Context, userID string) (map[string]string, error) {
|
||||
cs, err := mongoutil.Find[*model.ClientConfig](ctx, x.coll, bson.M{"user_id": userID})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
cm := make(map[string]string)
|
||||
for _, config := range cs {
|
||||
cm[config.Key] = config.Value
|
||||
}
|
||||
return cm, nil
|
||||
}
|
||||
|
||||
func (x *ClientConfig) Del(ctx context.Context, userID string, keys []string) error {
|
||||
if len(keys) == 0 {
|
||||
return nil
|
||||
}
|
||||
return mongoutil.DeleteMany(ctx, x.coll, bson.M{"key": bson.M{"$in": keys}, "user_id": userID})
|
||||
}
|
||||
|
||||
func (x *ClientConfig) GetPage(ctx context.Context, userID string, key string, pagination pagination.Pagination) (int64, []*model.ClientConfig, error) {
|
||||
filter := bson.M{}
|
||||
if userID != "" {
|
||||
filter["user_id"] = userID
|
||||
}
|
||||
if key != "" {
|
||||
filter["key"] = key
|
||||
}
|
||||
return mongoutil.FindPage[*model.ClientConfig](ctx, x.coll, filter, pagination)
|
||||
}
|
||||
7
pkg/common/storage/model/client_config.go
Normal file
7
pkg/common/storage/model/client_config.go
Normal file
@ -0,0 +1,7 @@
|
||||
package model
|
||||
|
||||
type ClientConfig struct {
|
||||
Key string `bson:"key"`
|
||||
UserID string `bson:"user_id"`
|
||||
Value string `bson:"value"`
|
||||
}
|
||||
@ -16,6 +16,7 @@ package rpccache
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/localcache"
|
||||
@ -153,6 +154,26 @@ func (c *ConversationLocalCache) getConversationNotReceiveMessageUserIDs(ctx con
|
||||
}))
|
||||
}
|
||||
|
||||
func (c *ConversationLocalCache) getPinnedConversationIDs(ctx context.Context, userID string) (val []string, err error) {
|
||||
log.ZDebug(ctx, "ConversationLocalCache getPinnedConversations req", "userID", userID)
|
||||
defer func() {
|
||||
if err == nil {
|
||||
log.ZDebug(ctx, "ConversationLocalCache getPinnedConversations return", "userID", userID, "value", val)
|
||||
} else {
|
||||
log.ZError(ctx, "ConversationLocalCache getPinnedConversations return", err, "userID", userID)
|
||||
}
|
||||
}()
|
||||
var cache cacheProto[pbconversation.GetPinnedConversationIDsResp]
|
||||
resp, err := cache.Unmarshal(c.local.Get(ctx, cachekey.GetPinnedConversationIDs(userID), func(ctx context.Context) ([]byte, error) {
|
||||
log.ZDebug(ctx, "ConversationLocalCache getConversationNotReceiveMessageUserIDs rpc", "userID", userID)
|
||||
return cache.Marshal(c.client.ConversationClient.GetPinnedConversationIDs(ctx, &pbconversation.GetPinnedConversationIDsReq{UserID: userID}))
|
||||
}))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return resp.ConversationIDs, nil
|
||||
}
|
||||
|
||||
func (c *ConversationLocalCache) GetConversationNotReceiveMessageUserIDs(ctx context.Context, conversationID string) ([]string, error) {
|
||||
res, err := c.getConversationNotReceiveMessageUserIDs(ctx, conversationID)
|
||||
if err != nil {
|
||||
@ -168,3 +189,7 @@ func (c *ConversationLocalCache) GetConversationNotReceiveMessageUserIDMap(ctx c
|
||||
}
|
||||
return datautil.SliceSet(res.UserIDs), nil
|
||||
}
|
||||
|
||||
func (c *ConversationLocalCache) GetPinnedConversationIDs(ctx context.Context, userID string) ([]string, error) {
|
||||
return c.getPinnedConversationIDs(ctx, userID)
|
||||
}
|
||||
|
||||
19
test/stress-test-v2/README.md
Normal file
19
test/stress-test-v2/README.md
Normal file
@ -0,0 +1,19 @@
|
||||
# Stress Test V2
|
||||
|
||||
## Usage
|
||||
|
||||
You need set `TestTargetUserList` variables.
|
||||
|
||||
### Build
|
||||
|
||||
```bash
|
||||
|
||||
go build -o test/stress-test-v2/stress-test-v2 test/stress-test-v2/main.go
|
||||
```
|
||||
|
||||
### Excute
|
||||
|
||||
```bash
|
||||
|
||||
tools/stress-test-v2/stress-test-v2 -c config/
|
||||
```
|
||||
759
test/stress-test-v2/main.go
Normal file
759
test/stress-test-v2/main.go
Normal file
@ -0,0 +1,759 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"flag"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/signal"
|
||||
"sync"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/apistruct"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||
"github.com/openimsdk/protocol/auth"
|
||||
"github.com/openimsdk/protocol/constant"
|
||||
"github.com/openimsdk/protocol/group"
|
||||
"github.com/openimsdk/protocol/sdkws"
|
||||
pbuser "github.com/openimsdk/protocol/user"
|
||||
"github.com/openimsdk/tools/log"
|
||||
"github.com/openimsdk/tools/system/program"
|
||||
)
|
||||
|
||||
// 1. Create 100K New Users
|
||||
// 2. Create 100 100K Groups
|
||||
// 3. Create 1000 999 Groups
|
||||
// 4. Send message to 100K Groups every second
|
||||
// 5. Send message to 999 Groups every minute
|
||||
|
||||
var (
|
||||
// Use default userIDs List for testing, need to be created.
|
||||
TestTargetUserList = []string{
|
||||
// "<need-update-it>",
|
||||
}
|
||||
// DefaultGroupID = "<need-update-it>" // Use default group ID for testing, need to be created.
|
||||
)
|
||||
|
||||
var (
|
||||
ApiAddress string
|
||||
|
||||
// API method
|
||||
GetAdminToken = "/auth/get_admin_token"
|
||||
UserCheck = "/user/account_check"
|
||||
CreateUser = "/user/user_register"
|
||||
ImportFriend = "/friend/import_friend"
|
||||
InviteToGroup = "/group/invite_user_to_group"
|
||||
GetGroupMemberInfo = "/group/get_group_members_info"
|
||||
SendMsg = "/msg/send_msg"
|
||||
CreateGroup = "/group/create_group"
|
||||
GetUserToken = "/auth/user_token"
|
||||
)
|
||||
|
||||
const (
|
||||
MaxUser = 100000
|
||||
Max1kUser = 1000
|
||||
Max100KGroup = 100
|
||||
Max999Group = 1000
|
||||
MaxInviteUserLimit = 999
|
||||
|
||||
CreateUserTicker = 1 * time.Second
|
||||
CreateGroupTicker = 1 * time.Second
|
||||
Create100KGroupTicker = 1 * time.Second
|
||||
Create999GroupTicker = 1 * time.Second
|
||||
SendMsgTo100KGroupTicker = 1 * time.Second
|
||||
SendMsgTo999GroupTicker = 1 * time.Minute
|
||||
)
|
||||
|
||||
type BaseResp struct {
|
||||
ErrCode int `json:"errCode"`
|
||||
ErrMsg string `json:"errMsg"`
|
||||
Data json.RawMessage `json:"data"`
|
||||
}
|
||||
|
||||
type StressTest struct {
|
||||
Conf *conf
|
||||
AdminUserID string
|
||||
AdminToken string
|
||||
DefaultGroupID string
|
||||
DefaultUserID string
|
||||
UserCounter int
|
||||
CreateUserCounter int
|
||||
Create100kGroupCounter int
|
||||
Create999GroupCounter int
|
||||
MsgCounter int
|
||||
CreatedUsers []string
|
||||
CreatedGroups []string
|
||||
Mutex sync.Mutex
|
||||
Ctx context.Context
|
||||
Cancel context.CancelFunc
|
||||
HttpClient *http.Client
|
||||
Wg sync.WaitGroup
|
||||
Once sync.Once
|
||||
}
|
||||
|
||||
type conf struct {
|
||||
Share config.Share
|
||||
Api config.API
|
||||
}
|
||||
|
||||
func initConfig(configDir string) (*config.Share, *config.API, error) {
|
||||
var (
|
||||
share = &config.Share{}
|
||||
apiConfig = &config.API{}
|
||||
)
|
||||
|
||||
err := config.Load(configDir, config.ShareFileName, config.EnvPrefixMap[config.ShareFileName], share)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
err = config.Load(configDir, config.OpenIMAPICfgFileName, config.EnvPrefixMap[config.OpenIMAPICfgFileName], apiConfig)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
return share, apiConfig, nil
|
||||
}
|
||||
|
||||
// Post Request
|
||||
func (st *StressTest) PostRequest(ctx context.Context, url string, reqbody any) ([]byte, error) {
|
||||
// Marshal body
|
||||
jsonBody, err := json.Marshal(reqbody)
|
||||
if err != nil {
|
||||
log.ZError(ctx, "Failed to marshal request body", err, "url", url, "reqbody", reqbody)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(jsonBody))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("operationID", st.AdminUserID)
|
||||
if st.AdminToken != "" {
|
||||
req.Header.Set("token", st.AdminToken)
|
||||
}
|
||||
|
||||
// log.ZInfo(ctx, "Header info is ", "Content-Type", "application/json", "operationID", st.AdminUserID, "token", st.AdminToken)
|
||||
|
||||
resp, err := st.HttpClient.Do(req)
|
||||
if err != nil {
|
||||
log.ZError(ctx, "Failed to send request", err, "url", url, "reqbody", reqbody)
|
||||
return nil, err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
respBody, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
log.ZError(ctx, "Failed to read response body", err, "url", url)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var baseResp BaseResp
|
||||
if err := json.Unmarshal(respBody, &baseResp); err != nil {
|
||||
log.ZError(ctx, "Failed to unmarshal response body", err, "url", url, "respBody", string(respBody))
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if baseResp.ErrCode != 0 {
|
||||
err = fmt.Errorf(baseResp.ErrMsg)
|
||||
// log.ZError(ctx, "Failed to send request", err, "url", url, "reqbody", reqbody, "resp", baseResp)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return baseResp.Data, nil
|
||||
}
|
||||
|
||||
func (st *StressTest) GetAdminToken(ctx context.Context) (string, error) {
|
||||
req := auth.GetAdminTokenReq{
|
||||
Secret: st.Conf.Share.Secret,
|
||||
UserID: st.AdminUserID,
|
||||
}
|
||||
|
||||
resp, err := st.PostRequest(ctx, ApiAddress+GetAdminToken, &req)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
data := &auth.GetAdminTokenResp{}
|
||||
if err := json.Unmarshal(resp, &data); err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
return data.Token, nil
|
||||
}
|
||||
|
||||
func (st *StressTest) CheckUser(ctx context.Context, userIDs []string) ([]string, error) {
|
||||
req := pbuser.AccountCheckReq{
|
||||
CheckUserIDs: userIDs,
|
||||
}
|
||||
|
||||
resp, err := st.PostRequest(ctx, ApiAddress+UserCheck, &req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
data := &pbuser.AccountCheckResp{}
|
||||
if err := json.Unmarshal(resp, &data); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
unRegisteredUserIDs := make([]string, 0)
|
||||
|
||||
for _, res := range data.Results {
|
||||
if res.AccountStatus == constant.UnRegistered {
|
||||
unRegisteredUserIDs = append(unRegisteredUserIDs, res.UserID)
|
||||
}
|
||||
}
|
||||
|
||||
return unRegisteredUserIDs, nil
|
||||
}
|
||||
|
||||
func (st *StressTest) CreateUser(ctx context.Context, userID string) (string, error) {
|
||||
user := &sdkws.UserInfo{
|
||||
UserID: userID,
|
||||
Nickname: userID,
|
||||
}
|
||||
|
||||
req := pbuser.UserRegisterReq{
|
||||
Users: []*sdkws.UserInfo{user},
|
||||
}
|
||||
|
||||
_, err := st.PostRequest(ctx, ApiAddress+CreateUser, &req)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
st.UserCounter++
|
||||
return userID, nil
|
||||
}
|
||||
|
||||
func (st *StressTest) CreateUserBatch(ctx context.Context, userIDs []string) error {
|
||||
// The method can import a large number of users at once.
|
||||
var userList []*sdkws.UserInfo
|
||||
|
||||
defer st.Once.Do(
|
||||
func() {
|
||||
st.DefaultUserID = userIDs[0]
|
||||
fmt.Println("Default Send User Created ID:", st.DefaultUserID)
|
||||
})
|
||||
|
||||
needUserIDs, err := st.CheckUser(ctx, userIDs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, userID := range needUserIDs {
|
||||
user := &sdkws.UserInfo{
|
||||
UserID: userID,
|
||||
Nickname: userID,
|
||||
}
|
||||
userList = append(userList, user)
|
||||
}
|
||||
|
||||
req := pbuser.UserRegisterReq{
|
||||
Users: userList,
|
||||
}
|
||||
|
||||
_, err = st.PostRequest(ctx, ApiAddress+CreateUser, &req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
st.UserCounter += len(userList)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (st *StressTest) GetGroupMembersInfo(ctx context.Context, groupID string, userIDs []string) ([]string, error) {
|
||||
needInviteUserIDs := make([]string, 0)
|
||||
|
||||
const maxBatchSize = 500
|
||||
if len(userIDs) > maxBatchSize {
|
||||
for i := 0; i < len(userIDs); i += maxBatchSize {
|
||||
end := min(i+maxBatchSize, len(userIDs))
|
||||
batchUserIDs := userIDs[i:end]
|
||||
|
||||
// log.ZInfo(ctx, "Processing group members batch", "groupID", groupID, "batch", i/maxBatchSize+1,
|
||||
// "batchUserCount", len(batchUserIDs))
|
||||
|
||||
// Process a single batch
|
||||
batchReq := group.GetGroupMembersInfoReq{
|
||||
GroupID: groupID,
|
||||
UserIDs: batchUserIDs,
|
||||
}
|
||||
|
||||
resp, err := st.PostRequest(ctx, ApiAddress+GetGroupMemberInfo, &batchReq)
|
||||
if err != nil {
|
||||
log.ZError(ctx, "Batch query failed", err, "batch", i/maxBatchSize+1)
|
||||
continue
|
||||
}
|
||||
|
||||
data := &group.GetGroupMembersInfoResp{}
|
||||
if err := json.Unmarshal(resp, &data); err != nil {
|
||||
log.ZError(ctx, "Failed to parse batch response", err, "batch", i/maxBatchSize+1)
|
||||
continue
|
||||
}
|
||||
|
||||
// Process the batch results
|
||||
existingMembers := make(map[string]bool)
|
||||
for _, member := range data.Members {
|
||||
existingMembers[member.UserID] = true
|
||||
}
|
||||
|
||||
for _, userID := range batchUserIDs {
|
||||
if !existingMembers[userID] {
|
||||
needInviteUserIDs = append(needInviteUserIDs, userID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return needInviteUserIDs, nil
|
||||
}
|
||||
|
||||
req := group.GetGroupMembersInfoReq{
|
||||
GroupID: groupID,
|
||||
UserIDs: userIDs,
|
||||
}
|
||||
|
||||
resp, err := st.PostRequest(ctx, ApiAddress+GetGroupMemberInfo, &req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
data := &group.GetGroupMembersInfoResp{}
|
||||
if err := json.Unmarshal(resp, &data); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
existingMembers := make(map[string]bool)
|
||||
for _, member := range data.Members {
|
||||
existingMembers[member.UserID] = true
|
||||
}
|
||||
|
||||
for _, userID := range userIDs {
|
||||
if !existingMembers[userID] {
|
||||
needInviteUserIDs = append(needInviteUserIDs, userID)
|
||||
}
|
||||
}
|
||||
|
||||
return needInviteUserIDs, nil
|
||||
}
|
||||
|
||||
func (st *StressTest) InviteToGroup(ctx context.Context, groupID string, userIDs []string) error {
|
||||
req := group.InviteUserToGroupReq{
|
||||
GroupID: groupID,
|
||||
InvitedUserIDs: userIDs,
|
||||
}
|
||||
_, err := st.PostRequest(ctx, ApiAddress+InviteToGroup, &req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (st *StressTest) SendMsg(ctx context.Context, userID string, groupID string) error {
|
||||
contentObj := map[string]any{
|
||||
// "content": fmt.Sprintf("index %d. The current time is %s", st.MsgCounter, time.Now().Format("2006-01-02 15:04:05.000")),
|
||||
"content": fmt.Sprintf("The current time is %s", time.Now().Format("2006-01-02 15:04:05.000")),
|
||||
}
|
||||
|
||||
req := &apistruct.SendMsgReq{
|
||||
SendMsg: apistruct.SendMsg{
|
||||
SendID: userID,
|
||||
SenderNickname: userID,
|
||||
GroupID: groupID,
|
||||
ContentType: constant.Text,
|
||||
SessionType: constant.ReadGroupChatType,
|
||||
Content: contentObj,
|
||||
},
|
||||
}
|
||||
|
||||
_, err := st.PostRequest(ctx, ApiAddress+SendMsg, &req)
|
||||
if err != nil {
|
||||
log.ZError(ctx, "Failed to send message", err, "userID", userID, "req", &req)
|
||||
return err
|
||||
}
|
||||
|
||||
st.MsgCounter++
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Max userIDs number is 1000
|
||||
func (st *StressTest) CreateGroup(ctx context.Context, groupID string, userID string, userIDsList []string) (string, error) {
|
||||
groupInfo := &sdkws.GroupInfo{
|
||||
GroupID: groupID,
|
||||
GroupName: groupID,
|
||||
GroupType: constant.WorkingGroup,
|
||||
}
|
||||
|
||||
req := group.CreateGroupReq{
|
||||
OwnerUserID: userID,
|
||||
MemberUserIDs: userIDsList,
|
||||
GroupInfo: groupInfo,
|
||||
}
|
||||
|
||||
resp := group.CreateGroupResp{}
|
||||
|
||||
response, err := st.PostRequest(ctx, ApiAddress+CreateGroup, &req)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
if err := json.Unmarshal(response, &resp); err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
// st.GroupCounter++
|
||||
|
||||
return resp.GroupInfo.GroupID, nil
|
||||
}
|
||||
|
||||
func main() {
|
||||
var configPath string
|
||||
// defaultConfigDir := filepath.Join("..", "..", "..", "..", "..", "config")
|
||||
// flag.StringVar(&configPath, "c", defaultConfigDir, "config path")
|
||||
flag.StringVar(&configPath, "c", "", "config path")
|
||||
flag.Parse()
|
||||
|
||||
if configPath == "" {
|
||||
_, _ = fmt.Fprintln(os.Stderr, "config path is empty")
|
||||
os.Exit(1)
|
||||
return
|
||||
}
|
||||
|
||||
fmt.Printf(" Config Path: %s\n", configPath)
|
||||
|
||||
share, apiConfig, err := initConfig(configPath)
|
||||
if err != nil {
|
||||
program.ExitWithError(err)
|
||||
return
|
||||
}
|
||||
|
||||
ApiAddress = fmt.Sprintf("http://%s:%s", "127.0.0.1", fmt.Sprint(apiConfig.Api.Ports[0]))
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
// ch := make(chan struct{})
|
||||
|
||||
st := &StressTest{
|
||||
Conf: &conf{
|
||||
Share: *share,
|
||||
Api: *apiConfig,
|
||||
},
|
||||
AdminUserID: share.IMAdminUserID[0],
|
||||
Ctx: ctx,
|
||||
Cancel: cancel,
|
||||
HttpClient: &http.Client{
|
||||
Timeout: 50 * time.Second,
|
||||
},
|
||||
}
|
||||
|
||||
c := make(chan os.Signal, 1)
|
||||
signal.Notify(c, os.Interrupt, syscall.SIGTERM)
|
||||
go func() {
|
||||
<-c
|
||||
fmt.Println("\nReceived stop signal, stopping...")
|
||||
|
||||
go func() {
|
||||
// time.Sleep(5 * time.Second)
|
||||
fmt.Println("Force exit")
|
||||
os.Exit(0)
|
||||
}()
|
||||
|
||||
st.Cancel()
|
||||
}()
|
||||
|
||||
token, err := st.GetAdminToken(st.Ctx)
|
||||
if err != nil {
|
||||
log.ZError(ctx, "Get Admin Token failed.", err, "AdminUserID", st.AdminUserID)
|
||||
}
|
||||
|
||||
st.AdminToken = token
|
||||
fmt.Println("Admin Token:", st.AdminToken)
|
||||
fmt.Println("ApiAddress:", ApiAddress)
|
||||
|
||||
for i := range MaxUser {
|
||||
userID := fmt.Sprintf("v2_StressTest_User_%d", i)
|
||||
st.CreatedUsers = append(st.CreatedUsers, userID)
|
||||
st.CreateUserCounter++
|
||||
}
|
||||
|
||||
// err = st.CreateUserBatch(st.Ctx, st.CreatedUsers)
|
||||
// if err != nil {
|
||||
// log.ZError(ctx, "Create user failed.", err)
|
||||
// }
|
||||
|
||||
const batchSize = 1000
|
||||
totalUsers := len(st.CreatedUsers)
|
||||
successCount := 0
|
||||
|
||||
if st.DefaultUserID == "" && len(st.CreatedUsers) > 0 {
|
||||
st.DefaultUserID = st.CreatedUsers[0]
|
||||
}
|
||||
|
||||
for i := 0; i < totalUsers; i += batchSize {
|
||||
end := min(i+batchSize, totalUsers)
|
||||
|
||||
userBatch := st.CreatedUsers[i:end]
|
||||
log.ZInfo(st.Ctx, "Creating user batch", "batch", i/batchSize+1, "count", len(userBatch))
|
||||
|
||||
err = st.CreateUserBatch(st.Ctx, userBatch)
|
||||
if err != nil {
|
||||
log.ZError(st.Ctx, "Batch user creation failed", err, "batch", i/batchSize+1)
|
||||
} else {
|
||||
successCount += len(userBatch)
|
||||
log.ZInfo(st.Ctx, "Batch user creation succeeded", "batch", i/batchSize+1,
|
||||
"progress", fmt.Sprintf("%d/%d", successCount, totalUsers))
|
||||
}
|
||||
}
|
||||
|
||||
// Execute create 100k group
|
||||
st.Wg.Add(1)
|
||||
go func() {
|
||||
defer st.Wg.Done()
|
||||
|
||||
create100kGroupTicker := time.NewTicker(Create100KGroupTicker)
|
||||
defer create100kGroupTicker.Stop()
|
||||
|
||||
for i := range Max100KGroup {
|
||||
select {
|
||||
case <-st.Ctx.Done():
|
||||
log.ZInfo(st.Ctx, "Stop Create 100K Group")
|
||||
return
|
||||
|
||||
case <-create100kGroupTicker.C:
|
||||
// Create 100K groups
|
||||
st.Wg.Add(1)
|
||||
go func(idx int) {
|
||||
startTime := time.Now()
|
||||
defer func() {
|
||||
elapsedTime := time.Since(startTime)
|
||||
log.ZInfo(st.Ctx, "100K group creation completed",
|
||||
"groupID", fmt.Sprintf("v2_StressTest_Group_100K_%d", idx),
|
||||
"index", idx,
|
||||
"duration", elapsedTime.String())
|
||||
}()
|
||||
|
||||
defer st.Wg.Done()
|
||||
defer func() {
|
||||
st.Mutex.Lock()
|
||||
st.Create100kGroupCounter++
|
||||
st.Mutex.Unlock()
|
||||
}()
|
||||
|
||||
groupID := fmt.Sprintf("v2_StressTest_Group_100K_%d", idx)
|
||||
|
||||
if _, err = st.CreateGroup(st.Ctx, groupID, st.DefaultUserID, TestTargetUserList); err != nil {
|
||||
log.ZError(st.Ctx, "Create group failed.", err)
|
||||
// continue
|
||||
}
|
||||
|
||||
for i := 0; i <= MaxUser/MaxInviteUserLimit; i++ {
|
||||
InviteUserIDs := make([]string, 0)
|
||||
// ensure TargetUserList is in group
|
||||
InviteUserIDs = append(InviteUserIDs, TestTargetUserList...)
|
||||
|
||||
startIdx := max(i*MaxInviteUserLimit, 1)
|
||||
endIdx := min((i+1)*MaxInviteUserLimit, MaxUser)
|
||||
|
||||
for j := startIdx; j < endIdx; j++ {
|
||||
userCreatedID := fmt.Sprintf("v2_StressTest_User_%d", j)
|
||||
InviteUserIDs = append(InviteUserIDs, userCreatedID)
|
||||
}
|
||||
|
||||
if len(InviteUserIDs) == 0 {
|
||||
// log.ZWarn(st.Ctx, "InviteUserIDs is empty", nil, "groupID", groupID)
|
||||
continue
|
||||
}
|
||||
|
||||
InviteUserIDs, err := st.GetGroupMembersInfo(ctx, groupID, InviteUserIDs)
|
||||
if err != nil {
|
||||
log.ZError(st.Ctx, "GetGroupMembersInfo failed.", err, "groupID", groupID)
|
||||
continue
|
||||
}
|
||||
|
||||
if len(InviteUserIDs) == 0 {
|
||||
// log.ZWarn(st.Ctx, "InviteUserIDs is empty", nil, "groupID", groupID)
|
||||
continue
|
||||
}
|
||||
|
||||
// Invite To Group
|
||||
if err = st.InviteToGroup(st.Ctx, groupID, InviteUserIDs); err != nil {
|
||||
log.ZError(st.Ctx, "Invite To Group failed.", err, "UserID", InviteUserIDs)
|
||||
continue
|
||||
// os.Exit(1)
|
||||
// return
|
||||
}
|
||||
}
|
||||
}(i)
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
// create 999 groups
|
||||
st.Wg.Add(1)
|
||||
go func() {
|
||||
defer st.Wg.Done()
|
||||
|
||||
create999GroupTicker := time.NewTicker(Create999GroupTicker)
|
||||
defer create999GroupTicker.Stop()
|
||||
|
||||
for i := range Max999Group {
|
||||
select {
|
||||
case <-st.Ctx.Done():
|
||||
log.ZInfo(st.Ctx, "Stop Create 999 Group")
|
||||
return
|
||||
|
||||
case <-create999GroupTicker.C:
|
||||
// Create 999 groups
|
||||
st.Wg.Add(1)
|
||||
go func(idx int) {
|
||||
startTime := time.Now()
|
||||
defer func() {
|
||||
elapsedTime := time.Since(startTime)
|
||||
log.ZInfo(st.Ctx, "999 group creation completed",
|
||||
"groupID", fmt.Sprintf("v2_StressTest_Group_1K_%d", idx),
|
||||
"index", idx,
|
||||
"duration", elapsedTime.String())
|
||||
}()
|
||||
|
||||
defer st.Wg.Done()
|
||||
defer func() {
|
||||
st.Mutex.Lock()
|
||||
st.Create999GroupCounter++
|
||||
st.Mutex.Unlock()
|
||||
}()
|
||||
|
||||
groupID := fmt.Sprintf("v2_StressTest_Group_1K_%d", idx)
|
||||
|
||||
if _, err = st.CreateGroup(st.Ctx, groupID, st.DefaultUserID, TestTargetUserList); err != nil {
|
||||
log.ZError(st.Ctx, "Create group failed.", err)
|
||||
// continue
|
||||
}
|
||||
for i := 0; i <= Max1kUser/MaxInviteUserLimit; i++ {
|
||||
InviteUserIDs := make([]string, 0)
|
||||
// ensure TargetUserList is in group
|
||||
InviteUserIDs = append(InviteUserIDs, TestTargetUserList...)
|
||||
|
||||
startIdx := max(i*MaxInviteUserLimit, 1)
|
||||
endIdx := min((i+1)*MaxInviteUserLimit, Max1kUser)
|
||||
|
||||
for j := startIdx; j < endIdx; j++ {
|
||||
userCreatedID := fmt.Sprintf("v2_StressTest_User_%d", j)
|
||||
InviteUserIDs = append(InviteUserIDs, userCreatedID)
|
||||
}
|
||||
|
||||
if len(InviteUserIDs) == 0 {
|
||||
// log.ZWarn(st.Ctx, "InviteUserIDs is empty", nil, "groupID", groupID)
|
||||
continue
|
||||
}
|
||||
|
||||
InviteUserIDs, err := st.GetGroupMembersInfo(ctx, groupID, InviteUserIDs)
|
||||
if err != nil {
|
||||
log.ZError(st.Ctx, "GetGroupMembersInfo failed.", err, "groupID", groupID)
|
||||
continue
|
||||
}
|
||||
|
||||
if len(InviteUserIDs) == 0 {
|
||||
// log.ZWarn(st.Ctx, "InviteUserIDs is empty", nil, "groupID", groupID)
|
||||
continue
|
||||
}
|
||||
|
||||
// Invite To Group
|
||||
if err = st.InviteToGroup(st.Ctx, groupID, InviteUserIDs); err != nil {
|
||||
log.ZError(st.Ctx, "Invite To Group failed.", err, "UserID", InviteUserIDs)
|
||||
continue
|
||||
// os.Exit(1)
|
||||
// return
|
||||
}
|
||||
}
|
||||
}(i)
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
// Send message to 100K groups
|
||||
st.Wg.Wait()
|
||||
fmt.Println("All groups created successfully, starting to send messages...")
|
||||
log.ZInfo(ctx, "All groups created successfully, starting to send messages...")
|
||||
|
||||
var groups100K []string
|
||||
var groups999 []string
|
||||
|
||||
for i := range Max100KGroup {
|
||||
groupID := fmt.Sprintf("v2_StressTest_Group_100K_%d", i)
|
||||
groups100K = append(groups100K, groupID)
|
||||
}
|
||||
|
||||
for i := range Max999Group {
|
||||
groupID := fmt.Sprintf("v2_StressTest_Group_1K_%d", i)
|
||||
groups999 = append(groups999, groupID)
|
||||
}
|
||||
|
||||
send100kGroupLimiter := make(chan struct{}, 20)
|
||||
send999GroupLimiter := make(chan struct{}, 100)
|
||||
|
||||
// execute Send message to 100K groups
|
||||
go func() {
|
||||
ticker := time.NewTicker(SendMsgTo100KGroupTicker)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-st.Ctx.Done():
|
||||
log.ZInfo(st.Ctx, "Stop Send Message to 100K Group")
|
||||
return
|
||||
|
||||
case <-ticker.C:
|
||||
// Send message to 100K groups
|
||||
for _, groupID := range groups100K {
|
||||
send100kGroupLimiter <- struct{}{}
|
||||
go func(groupID string) {
|
||||
defer func() { <-send100kGroupLimiter }()
|
||||
if err := st.SendMsg(st.Ctx, st.DefaultUserID, groupID); err != nil {
|
||||
log.ZError(st.Ctx, "Send message to 100K group failed.", err)
|
||||
}
|
||||
}(groupID)
|
||||
}
|
||||
// log.ZInfo(st.Ctx, "Send message to 100K groups successfully.")
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
// execute Send message to 999 groups
|
||||
go func() {
|
||||
ticker := time.NewTicker(SendMsgTo999GroupTicker)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-st.Ctx.Done():
|
||||
log.ZInfo(st.Ctx, "Stop Send Message to 999 Group")
|
||||
return
|
||||
|
||||
case <-ticker.C:
|
||||
// Send message to 999 groups
|
||||
for _, groupID := range groups999 {
|
||||
send999GroupLimiter <- struct{}{}
|
||||
go func(groupID string) {
|
||||
defer func() { <-send999GroupLimiter }()
|
||||
|
||||
if err := st.SendMsg(st.Ctx, st.DefaultUserID, groupID); err != nil {
|
||||
log.ZError(st.Ctx, "Send message to 999 group failed.", err)
|
||||
}
|
||||
}(groupID)
|
||||
}
|
||||
// log.ZInfo(st.Ctx, "Send message to 999 groups successfully.")
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
<-st.Ctx.Done()
|
||||
fmt.Println("Received signal to exit, shutting down...")
|
||||
}
|
||||
19
test/stress-test/README.md
Normal file
19
test/stress-test/README.md
Normal file
@ -0,0 +1,19 @@
|
||||
# Stress Test
|
||||
|
||||
## Usage
|
||||
|
||||
You need set `TestTargetUserList` and `DefaultGroupID` variables.
|
||||
|
||||
### Build
|
||||
|
||||
```bash
|
||||
|
||||
go build -o test/stress-test/stress-test test/stress-test/main.go
|
||||
```
|
||||
|
||||
### Excute
|
||||
|
||||
```bash
|
||||
|
||||
tools/stress-test/stress-test -c config/
|
||||
```
|
||||
@ -14,6 +14,7 @@ import (
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/apistruct"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||
"github.com/openimsdk/protocol/auth"
|
||||
"github.com/openimsdk/protocol/constant"
|
||||
@ -71,22 +72,22 @@ type BaseResp struct {
|
||||
}
|
||||
|
||||
type StressTest struct {
|
||||
Conf *conf
|
||||
AdminUserID string
|
||||
AdminToken string
|
||||
DefaultGroupID string
|
||||
DefaultSendUserID string
|
||||
UserCounter int
|
||||
GroupCounter int
|
||||
MsgCounter int
|
||||
CreatedUsers []string
|
||||
CreatedGroups []string
|
||||
Mutex sync.Mutex
|
||||
Ctx context.Context
|
||||
Cancel context.CancelFunc
|
||||
HttpClient *http.Client
|
||||
Wg sync.WaitGroup
|
||||
Once sync.Once
|
||||
Conf *conf
|
||||
AdminUserID string
|
||||
AdminToken string
|
||||
DefaultGroupID string
|
||||
DefaultUserID string
|
||||
UserCounter int
|
||||
GroupCounter int
|
||||
MsgCounter int
|
||||
CreatedUsers []string
|
||||
CreatedGroups []string
|
||||
Mutex sync.Mutex
|
||||
Ctx context.Context
|
||||
Cancel context.CancelFunc
|
||||
HttpClient *http.Client
|
||||
Wg sync.WaitGroup
|
||||
Once sync.Once
|
||||
}
|
||||
|
||||
type conf struct {
|
||||
@ -232,12 +233,15 @@ func (st *StressTest) SendMsg(ctx context.Context, userID string) error {
|
||||
"content": fmt.Sprintf("index %d. The current time is %s", st.MsgCounter, time.Now().Format("2006-01-02 15:04:05.000")),
|
||||
}
|
||||
|
||||
req := map[string]any{
|
||||
"sendID": userID,
|
||||
"groupID": st.DefaultGroupID,
|
||||
"contentType": constant.Text,
|
||||
"sessionType": constant.ReadGroupChatType,
|
||||
"content": contentObj,
|
||||
req := &apistruct.SendMsgReq{
|
||||
SendMsg: apistruct.SendMsg{
|
||||
SendID: userID,
|
||||
SenderNickname: userID,
|
||||
GroupID: st.DefaultGroupID,
|
||||
ContentType: constant.Text,
|
||||
SessionType: constant.ReadGroupChatType,
|
||||
Content: contentObj,
|
||||
},
|
||||
}
|
||||
|
||||
_, err := st.PostRequest(ctx, ApiAddress+SendMsg, &req)
|
||||
@ -254,15 +258,18 @@ func (st *StressTest) SendMsg(ctx context.Context, userID string) error {
|
||||
func (st *StressTest) CreateGroup(ctx context.Context, userID string) (string, error) {
|
||||
groupID := fmt.Sprintf("StressTestGroup_%d_%s", st.GroupCounter, time.Now().Format("20060102150405"))
|
||||
|
||||
req := map[string]any{
|
||||
"memberUserIDs": TestTargetUserList,
|
||||
"ownerUserID": userID,
|
||||
"groupInfo": map[string]any{
|
||||
"groupID": groupID,
|
||||
"groupName": groupID,
|
||||
"groupType": constant.WorkingGroup,
|
||||
},
|
||||
groupInfo := &sdkws.GroupInfo{
|
||||
GroupID: groupID,
|
||||
GroupName: groupID,
|
||||
GroupType: constant.WorkingGroup,
|
||||
}
|
||||
|
||||
req := group.CreateGroupReq{
|
||||
OwnerUserID: userID,
|
||||
MemberUserIDs: TestTargetUserList,
|
||||
GroupInfo: groupInfo,
|
||||
}
|
||||
|
||||
resp := group.CreateGroupResp{}
|
||||
|
||||
response, err := st.PostRequest(ctx, ApiAddress+CreateGroup, &req)
|
||||
@ -377,7 +384,6 @@ func main() {
|
||||
os.Exit(1)
|
||||
return
|
||||
}
|
||||
|
||||
// Invite To Group
|
||||
if err = st.InviteToGroup(st.Ctx, userCreatedID); err != nil {
|
||||
log.ZError(st.Ctx, "Invite To Group failed.", err, "UserID", userCreatedID)
|
||||
@ -386,7 +392,7 @@ func main() {
|
||||
}
|
||||
|
||||
st.Once.Do(func() {
|
||||
st.DefaultSendUserID = userCreatedID
|
||||
st.DefaultUserID = userCreatedID
|
||||
fmt.Println("Default Send User Created ID:", userCreatedID)
|
||||
close(ch)
|
||||
})
|
||||
@ -410,8 +416,8 @@ func main() {
|
||||
|
||||
case <-ticker.C:
|
||||
// Send Message
|
||||
if err = st.SendMsg(st.Ctx, st.DefaultSendUserID); err != nil {
|
||||
log.ZError(st.Ctx, "Send Message failed.", err, "UserID", st.DefaultSendUserID)
|
||||
if err = st.SendMsg(st.Ctx, st.DefaultUserID); err != nil {
|
||||
log.ZError(st.Ctx, "Send Message failed.", err, "UserID", st.DefaultUserID)
|
||||
continue
|
||||
}
|
||||
}
|
||||
@ -436,9 +442,9 @@ func main() {
|
||||
case <-ticker.C:
|
||||
|
||||
// Create Group
|
||||
_, err := st.CreateGroup(st.Ctx, st.DefaultSendUserID)
|
||||
_, err := st.CreateGroup(st.Ctx, st.DefaultUserID)
|
||||
if err != nil {
|
||||
log.ZError(st.Ctx, "Create Group failed.", err, "UserID", st.DefaultSendUserID)
|
||||
log.ZError(st.Ctx, "Create Group failed.", err, "UserID", st.DefaultUserID)
|
||||
os.Exit(1)
|
||||
return
|
||||
}
|
||||
@ -4,6 +4,11 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"net/http"
|
||||
"path/filepath"
|
||||
"time"
|
||||
|
||||
"github.com/mitchellh/mapstructure"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis"
|
||||
@ -19,10 +24,6 @@ import (
|
||||
"github.com/openimsdk/tools/s3/oss"
|
||||
"github.com/spf13/viper"
|
||||
"go.mongodb.org/mongo-driver/mongo"
|
||||
"log"
|
||||
"net/http"
|
||||
"path/filepath"
|
||||
"time"
|
||||
)
|
||||
|
||||
const defaultTimeout = time.Second * 10
|
||||
@ -159,7 +160,7 @@ func doObject(db database.ObjectInfo, newS3, oldS3 s3.Interface, skip int) (*Res
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
putURL, err := newS3.PresignedPutObject(ctx, obj.Key, time.Hour)
|
||||
putURL, err := newS3.PresignedPutObject(ctx, obj.Key, time.Hour, &s3.PutOption{ContentType: obj.ContentType})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@ -176,7 +177,7 @@ func doObject(db database.ObjectInfo, newS3, oldS3 s3.Interface, skip int) (*Res
|
||||
return nil, fmt.Errorf("download object failed %s", downloadResp.Status)
|
||||
}
|
||||
log.Printf("file size %d", obj.Size)
|
||||
request, err := http.NewRequest(http.MethodPut, putURL, downloadResp.Body)
|
||||
request, err := http.NewRequest(http.MethodPut, putURL.URL, downloadResp.Body)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@ -337,7 +337,7 @@ func SetVersion(coll *mongo.Collection, key string, version int) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
|
||||
defer cancel()
|
||||
option := options.Update().SetUpsert(true)
|
||||
filter := bson.M{"key": key, "value": strconv.Itoa(version)}
|
||||
filter := bson.M{"key": key}
|
||||
update := bson.M{"$set": bson.M{"key": key, "value": strconv.Itoa(version)}}
|
||||
return mongoutil.UpdateOne(ctx, coll, filter, update, false, option)
|
||||
}
|
||||
|
||||
@ -1,25 +0,0 @@
|
||||
# Stress Test
|
||||
|
||||
## Usage
|
||||
|
||||
You need set `TestTargetUserList` and `DefaultGroupID` variables.
|
||||
|
||||
### Build
|
||||
|
||||
```bash
|
||||
go build -o _output/bin/tools/linux/amd64/stress-test tools/stress-test/main.go
|
||||
|
||||
# or
|
||||
|
||||
go build -o tools/stress-test/stress-test tools/stress-test/main.go
|
||||
```
|
||||
|
||||
### Excute
|
||||
|
||||
```bash
|
||||
_output/bin/tools/linux/amd64/stress-test -c config/
|
||||
|
||||
#or
|
||||
|
||||
tools/stress-test/stress-test -c config/
|
||||
```
|
||||
@ -1,6 +1,14 @@
|
||||
package version
|
||||
|
||||
import _ "embed"
|
||||
import (
|
||||
_ "embed"
|
||||
"strings"
|
||||
)
|
||||
|
||||
//go:embed version
|
||||
var Version string
|
||||
|
||||
func init() {
|
||||
Version = strings.Trim(Version, "\n")
|
||||
Version = strings.TrimSpace(Version)
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user