mirror of
https://github.com/openimsdk/open-im-server.git
synced 2026-05-15 06:25:58 +08:00
merge: update release-v3.8 with main changes. (#2708)
* chore: revert tool pkg version. (#2476)
* feat: update go mod pkg to latest.
* revert tool pkg version.
* feat: update grafana template (#2484)
* feat: update web front images (#2487)
* fix: import del cache (#2492)
* fix:update recive name to same (#2493)
* fix: fix-validate message. (#2499)
* h (#2501)
* refactor: refactor workflows structure. (#2511)
* refactor: refactor workflows contents.
* add tool workflows.
* update field.
* fix: remove chat error.
* Fix err.
* fix error.
* remove cn comment.
* update workflows files.
* update infra config.
* move workflows.
* fix: solve uncorrect outdated msg get. (#2513)
* refactor: refactor workflows contents.
* add tool workflows.
* update field.
* fix: remove chat error.
* Fix err.
* fix error.
* remove cn comment.
* update workflows files.
* update infra config.
* move workflows.
* feat: update bot.
* fix: solve uncorrect outdated msg get.
* update get docIDs logic.
* update
* update skip logic.
* fix
* update.
* feat: update issue translator in workflows (#2521)
* refactor: refactor workflows contents.
* add tool workflows.
* update field.
* fix: remove chat error.
* Fix err.
* fix error.
* remove cn comment.
* update workflows files.
* update infra config.
* move workflows.
* feat: update bot.
* fix: solve uncorrect outdated msg get.
* update get docIDs logic.
* update
* update skip logic.
* fix
* update.
* feat: update issue translator.
* fix: pass getMinioImageThumbnailKey error. (#2532)
* refactor: refactor workflows contents.
* add tool workflows.
* update field.
* fix: remove chat error.
* Fix err.
* fix error.
* remove cn comment.
* update workflows files.
* update infra config.
* move workflows.
* feat: update bot.
* fix: solve uncorrect outdated msg get.
* update get docIDs logic.
* update
* update skip logic.
* fix
* update.
* feat: update issue translator.
* fix: pass getMinioImageThumbnailKey error.
* docs: update CLA comments contents. (#2534)
* refactor: refactor workflows contents.
* add tool workflows.
* update field.
* fix: remove chat error.
* Fix err.
* fix error.
* remove cn comment.
* update workflows files.
* update infra config.
* move workflows.
* feat: update bot.
* fix: solve uncorrect outdated msg get.
* update get docIDs logic.
* update
* update skip logic.
* fix
* update.
* docs: update CLA comments contents.
* update workflow file.
* fix: the log key value is not aligned (#2527)
* fix: GroupApplicationAcceptedNotification
* fix: GroupApplicationAcceptedNotification
* fix: NotificationUserInfoUpdate
* cicd: robot automated Change
* fix: component
* fix: getConversationInfo
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* fix: minio config url recognition error
* update gomake version
* update gomake version
* fix: seq conversion bug
* fix: redis pipe exec
* fix: ImportFriends
* fix: A large number of logs keysAndValues length is not even
---------
Co-authored-by: withchao <withchao@users.noreply.github.com>
* fix 细节 (#2525)
1、统一结构体方法 receiver，都用 pointer
2、使用 errors.Is 来做错误判断
3、修复单词拼写的错误
* Fix config (#2541)
* feat: remove double quotation marks
* feat: test
* feat: default close prometheus
* Fix: solve conversation blocking in private chat when non friendship. (#2542)
* refactor: refactor workflows contents.
* add tool workflows.
* update field.
* fix: remove chat error.
* Fix err.
* fix error.
* remove cn comment.
* update workflows files.
* update infra config.
* move workflows.
* feat: update bot.
* fix: solve uncorrect outdated msg get.
* update get docIDs logic.
* update
* update skip logic.
* fix
* update.
* fix: update contribute cla.
* fix: solve conversations bug.
* fix:get msg error (#2494)
* fix:mgo delete err (#2496)
* fix:doPut error (#2495)
* feat: update set conversation logic. (#2544)
* refactor: refactor workflows contents.
* add tool workflows.
* update field.
* fix: remove chat error.
* Fix err.
* fix error.
* remove cn comment.
* update workflows files.
* update infra config.
* move workflows.
* feat: update bot.
* fix: solve uncorrect outdated msg get.
* update get docIDs logic.
* update
* update skip logic.
* fix
* update.
* feat: update set conversation logic.
* content format.
* update setConversation logic.
* update conversation logic.
* update conversation logic.
* update logic.
* update logic.
* Groupmsg (#2548)
* feat: add EnableHistoryForNewMembers
* feat: change name
* refactor: move first create conversation
* fix: set min seq
* feat: implement `SetGroupInfoEX` interface. (#2552)
* refactor: refactor workflows contents.
* add tool workflows.
* update field.
* fix: remove chat error.
* Fix err.
* fix error.
* remove cn comment.
* update workflows files.
* update infra config.
* move workflows.
* feat: update bot.
* fix: solve uncorrect outdated msg get.
* update get docIDs logic.
* update
* update skip logic.
* fix
* update.
* feat: implement `SetGroupInfoEX` interface.
* remove null interface.
* update router.
* fix: set min seq (#2556)
* fix:log
* fix: set min seq
* Fix push (#2559)
* fix:log
* fix: add log
* fix: del return
* fix: push config
* feat: add push err log and extend push wait time
* feat: group config
* feat: add log in write binary msg
* feat: Modify Prometheus data scraping ports and reserve port space.
* feat: change group rpc num
* feat: change log
* fix: remove quotation mark
* fix: read seq is written to mongo, online status redis cluster is supported (#2558)
* fix: GroupApplicationAcceptedNotification
* fix: GroupApplicationAcceptedNotification
* fix: NotificationUserInfoUpdate
* cicd: robot automated Change
* fix: component
* fix: getConversationInfo
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* fix: minio config url recognition error
* update gomake version
* update gomake version
* fix: seq conversion bug
* fix: redis pipe exec
* fix: ImportFriends
* fix: A large number of logs keysAndValues length is not even
* feat: mark read aggregate write
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* merge
* merge
* read seq is written to mongo
* read seq is written to mongo
---------
Co-authored-by: withchao <withchao@users.noreply.github.com>
* fix: invitation to join group notification opuser is null (#2562)
* fix: GroupApplicationAcceptedNotification
* fix: GroupApplicationAcceptedNotification
* fix: NotificationUserInfoUpdate
* cicd: robot automated Change
* fix: component
* fix: getConversationInfo
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* fix: minio config url recognition error
* update gomake version
* update gomake version
* fix: seq conversion bug
* fix: redis pipe exec
* fix: ImportFriends
* fix: A large number of logs keysAndValues length is not even
* feat: mark read aggregate write
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* merge
* merge
* read seq is written to mongo
* read seq is written to mongo
* fix: invitation to join group notification
---------
Co-authored-by: withchao <withchao@users.noreply.github.com>
* Fix set convsation (#2564)
* fix: set conversation unequal check
* fix: check update list
* fix: delay deleteObject func. (#2566)
* refactor: refactor workflows contents.
* add tool workflows.
* update field.
* fix: remove chat error.
* Fix err.
* fix error.
* remove cn comment.
* update workflows files.
* update infra config.
* move workflows.
* feat: update bot.
* fix: solve uncorrect outdated msg get.
* update get docIDs logic.
* update
* update skip logic.
* fix
* update.
* fix: delay deleteObject func.
* remove unused content.
* fix: memory queue optimization (#2568)
* fix: GroupApplicationAcceptedNotification
* fix: GroupApplicationAcceptedNotification
* fix: NotificationUserInfoUpdate
* cicd: robot automated Change
* fix: component
* fix: getConversationInfo
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* fix: minio config url recognition error
* update gomake version
* update gomake version
* fix: seq conversion bug
* fix: redis pipe exec
* fix: ImportFriends
* fix: A large number of logs keysAndValues length is not even
* feat: mark read aggregate write
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* merge
* merge
* read seq is written to mongo
* read seq is written to mongo
* fix: invitation to join group notification
* fix: friend op_user_id
* feat: optimizing asynchronous context
* feat: optimizing memamq size
---------
Co-authored-by: withchao <withchao@users.noreply.github.com>
* fix: fill opUser in invite tips (#2578)
* fix: fill opUser in invite tips
* fix: del code
* feat: update group notification when set to null. (#2590)
* refactor: refactor workflows contents.
* add tool workflows.
* update field.
* fix: remove chat error.
* Fix err.
* fix error.
* remove cn comment.
* update workflows files.
* update infra config.
* move workflows.
* feat: update bot.
* fix: solve uncorrect outdated msg get.
* update get docIDs logic.
* update
* update skip logic.
* fix
* update.
* fix: delay deleteObject func.
* remove unused content.
* feat: update group notification when set to null.
* update log standard.
* feat: add long time push msg in prometheus (#2584)
* feat: add long time push msg in prometheus
* fix: log print
* fix: go mod
* fix: log msg
* fix: log init
* feat: push msg
* feat: go mod ,remove cgo package
* feat: remove error log
* feat: test dummy push
* feat:redis pool config
* feat: push to kafka log
* feat: supports getting messages based on session ID and seq (#2582)
* fix: GroupApplicationAcceptedNotification
* fix: GroupApplicationAcceptedNotification
* fix: NotificationUserInfoUpdate
* cicd: robot automated Change
* fix: component
* fix: getConversationInfo
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* fix: minio config url recognition error
* update gomake version
* update gomake version
* fix: seq conversion bug
* fix: redis pipe exec
* fix: ImportFriends
* fix: A large number of logs keysAndValues length is not even
* feat: mark read aggregate write
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* merge
* merge
* read seq is written to mongo
* read seq is written to mongo
* fix: invitation to join group notification
* fix: friend op_user_id
* feat: optimizing asynchronous context
* feat: optimizing memamq size
* feat: add GetSeqMessage
* feat: GroupApplicationAgreeMemberEnterNotification
* feat: GroupApplicationAgreeMemberEnterNotification
* feat: go.mod
* feat: go.mod
* feat: join group notification and get seq
---------
Co-authored-by: withchao <withchao@users.noreply.github.com>
* feat: implement request batch count limit. (#2591)
* refactor: refactor workflows contents.
* add tool workflows.
* update field.
* fix: remove chat error.
* Fix err.
* fix error.
* remove cn comment.
* update workflows files.
* update infra config.
* move workflows.
* feat: update bot.
* fix: solve uncorrect outdated msg get.
* update get docIDs logic.
* update
* update skip logic.
* fix
* update.
* fix: delay deleteObject func.
* remove unused content.
* update log type.
* feat: implement request batch count limit.
* update
* update
* fix: getting messages based on session ID and seq (#2595)
* fix: GroupApplicationAcceptedNotification
* fix: GroupApplicationAcceptedNotification
* fix: NotificationUserInfoUpdate
* cicd: robot automated Change
* fix: component
* fix: getConversationInfo
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* fix: minio config url recognition error
* update gomake version
* update gomake version
* fix: seq conversion bug
* fix: redis pipe exec
* fix: ImportFriends
* fix: A large number of logs keysAndValues length is not even
* feat: mark read aggregate write
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* merge
* merge
* read seq is written to mongo
* read seq is written to mongo
* fix: invitation to join group notification
* fix: friend op_user_id
* feat: optimizing asynchronous context
* feat: optimizing memamq size
* feat: add GetSeqMessage
* feat: GroupApplicationAgreeMemberEnterNotification
* feat: GroupApplicationAgreeMemberEnterNotification
* feat: go.mod
* feat: go.mod
* feat: join group notification and get seq
* feat: join group notification and get seq
---------
Co-authored-by: withchao <withchao@users.noreply.github.com>
* feat: avoid pulling messages from sessions with a large number of max seq values of 0 (#2602)
* fix: GroupApplicationAcceptedNotification
* fix: GroupApplicationAcceptedNotification
* fix: NotificationUserInfoUpdate
* cicd: robot automated Change
* fix: component
* fix: getConversationInfo
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* fix: minio config url recognition error
* update gomake version
* update gomake version
* fix: seq conversion bug
* fix: redis pipe exec
* fix: ImportFriends
* fix: A large number of logs keysAndValues length is not even
* feat: mark read aggregate write
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* merge
* merge
* read seq is written to mongo
* read seq is written to mongo
* fix: invitation to join group notification
* fix: friend op_user_id
* feat: optimizing asynchronous context
* feat: optimizing memamq size
* feat: add GetSeqMessage
* feat: GroupApplicationAgreeMemberEnterNotification
* feat: GroupApplicationAgreeMemberEnterNotification
* feat: go.mod
* feat: go.mod
* feat: join group notification and get seq
* feat: join group notification and get seq
* feat: avoid pulling messages from sessions with a large number of max seq values of 0
---------
Co-authored-by: withchao <withchao@users.noreply.github.com>
* refactor: improve db structure in `storage/controller` (#2604)
* refactor: refactor workflows contents.
* add tool workflows.
* update field.
* fix: remove chat error.
* Fix err.
* fix error.
* remove cn comment.
* update workflows files.
* update infra config.
* move workflows.
* feat: update bot.
* fix: solve uncorrect outdated msg get.
* update get docIDs logic.
* update
* update skip logic.
* fix
* update.
* fix: delay deleteObject func.
* remove unused content.
* update log type.
* feat: implement request batch count limit.
* update
* update
* refactor: improve db structure in `storage/controller`
* feat: implement offline push using kafka (#2600)
* refactor: refactor workflows contents.
* add tool workflows.
* update field.
* fix: remove chat error.
* Fix err.
* fix error.
* remove cn comment.
* update workflows files.
* update infra config.
* move workflows.
* feat: update bot.
* fix: solve uncorrect outdated msg get.
* update get docIDs logic.
* update
* update skip logic.
* fix
* update.
* fix: delay deleteObject func.
* remove unused content.
* update log type.
* feat: implement request batch count limit.
* update
* update
* feat: implement offline push.
* feat: implement batch Push spilt
* update go mod
* feat: implement kafka producer and consumer.
* update format,
* add PushMQ log.
* feat: update Handler logic.
* update MQ logic.
* update
* update
* fix: update OfflinePushConsumerHandler.
* feat: API supports gzip (#2609)
* fix: GroupApplicationAcceptedNotification
* fix: GroupApplicationAcceptedNotification
* fix: NotificationUserInfoUpdate
* cicd: robot automated Change
* fix: component
* fix: getConversationInfo
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* fix: minio config url recognition error
* update gomake version
* update gomake version
* fix: seq conversion bug
* fix: redis pipe exec
* fix: ImportFriends
* fix: A large number of logs keysAndValues length is not even
* feat: mark read aggregate write
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* merge
* merge
* read seq is written to mongo
* read seq is written to mongo
* fix: invitation to join group notification
* fix: friend op_user_id
* feat: optimizing asynchronous context
* feat: optimizing memamq size
* feat: add GetSeqMessage
* feat: GroupApplicationAgreeMemberEnterNotification
* feat: GroupApplicationAgreeMemberEnterNotification
* feat: go.mod
* feat: go.mod
* feat: join group notification and get seq
* feat: join group notification and get seq
* feat: avoid pulling messages from sessions with a large number of max seq values of 0
* feat: API supports gzip
---------
Co-authored-by: withchao <withchao@users.noreply.github.com>
* Fix err (#2608)
* refactor: refactor workflows contents.
* add tool workflows.
* update field.
* fix: remove chat error.
* Fix err.
* fix error.
* remove cn comment.
* update workflows files.
* update infra config.
* move workflows.
* feat: update bot.
* fix: solve uncorrect outdated msg get.
* update get docIDs logic.
* update
* update skip logic.
* fix
* update.
* fix: delay deleteObject func.
* remove unused content.
* update log type.
* feat: implement request batch count limit.
* update
* update
* feat: add rocksTimeout
* feat: wrap logs
* feat: add logs
* feat: listen config
* feat: enable listen TIME_WAIT port
* feat: add logs
* feat: cache batch
* chore: enable fullUserCache
* feat: push rpc num
* feat: push err
* feat: with operationID
* feat: sleep
* feat: change 1s
* feat: change log
* feat: implement Getbatch in rpcCache.
* feat: print getOnline cost
* feat: change log
* feat: change kafka and push config
* feat: del interface
* feat: fix err
* feat: change config
* feat: go mod
* feat: change config
* feat: change config
* feat: add sleep in push
* feat: warn logs
* feat: logs
* feat: logs
* feat: change port
* feat: start config
* feat: remove port reuse
* feat: prometheus config
* feat: prometheus config
* feat: prometheus config
* feat: add long time send msg to grafana
* feat: init
* feat: init
* feat: implement offline push.
* feat: batch get user online
* feat: implement batch Push spilt
* update go mod
* Revert "feat: change port"
This reverts commit 06d5e944
* feat: change port
* feat: change config
* feat: implement kafka producer and consumer.
* update format,
* add PushMQ log.
* feat: get all online users and init push
* feat: lock in online cache
* feat: config
* fix: init online status
* fix: add logs
* fix: userIDs
* fix: add logs
* feat: update Handler logic.
* update MQ logic.
* update
* update
* fix: method name
* fix: update OfflinePushConsumerHandler.
* fix: prommetrics
* fix: add logs
* fix: ctx
* fix: log
* fix: config
* feat: change port
* fix: atomic online cache status
---------
Co-authored-by: Monet Lee <monet_lee@163.com>
* feature: add GetConversationsHasReadAndMaxSeq interface to the WebSocket API. (#2611)
* fix: lru lock (#2613)
* fix: lru lock
* fix: lru lock
* fix: lru lock
* fix: nil pointer error on close (#2618)
* fix: GroupApplicationAcceptedNotification
* fix: GroupApplicationAcceptedNotification
* fix: NotificationUserInfoUpdate
* cicd: robot automated Change
* fix: component
* fix: getConversationInfo
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* fix: minio config url recognition error
* update gomake version
* update gomake version
* fix: seq conversion bug
* fix: redis pipe exec
* fix: ImportFriends
* fix: A large number of logs keysAndValues length is not even
* feat: mark read aggregate write
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* merge
* merge
* read seq is written to mongo
* read seq is written to mongo
* fix: invitation to join group notification
* fix: friend op_user_id
* feat: optimizing asynchronous context
* feat: optimizing memamq size
* feat: add GetSeqMessage
* feat: GroupApplicationAgreeMemberEnterNotification
* feat: GroupApplicationAgreeMemberEnterNotification
* feat: go.mod
* feat: go.mod
* feat: join group notification and get seq
* feat: join group notification and get seq
* feat: avoid pulling messages from sessions with a large number of max seq values of 0
* feat: API supports gzip
* go.mod
* fix: nil pointer error on close
---------
Co-authored-by: withchao <withchao@users.noreply.github.com>
* feat: create group can push notification (#2617)
* fix: blockage caused by listen error (#2620)
* fix: GroupApplicationAcceptedNotification
* fix: GroupApplicationAcceptedNotification
* fix: NotificationUserInfoUpdate
* cicd: robot automated Change
* fix: component
* fix: getConversationInfo
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* fix: minio config url recognition error
* update gomake version
* update gomake version
* fix: seq conversion bug
* fix: redis pipe exec
* fix: ImportFriends
* fix: A large number of logs keysAndValues length is not even
* feat: mark read aggregate write
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* merge
* merge
* read seq is written to mongo
* read seq is written to mongo
* fix: invitation to join group notification
* fix: friend op_user_id
* feat: optimizing asynchronous context
* feat: optimizing memamq size
* feat: add GetSeqMessage
* feat: GroupApplicationAgreeMemberEnterNotification
* feat: GroupApplicationAgreeMemberEnterNotification
* feat: go.mod
* feat: go.mod
* feat: join group notification and get seq
* feat: join group notification and get seq
* feat: avoid pulling messages from sessions with a large number of max seq values of 0
* feat: API supports gzip
* go.mod
* fix: nil pointer error on close
* fix: listen error
---------
Co-authored-by: withchao <withchao@users.noreply.github.com>
* fix: go.mod (#2621)
* fix: GroupApplicationAcceptedNotification
* fix: GroupApplicationAcceptedNotification
* fix: NotificationUserInfoUpdate
* cicd: robot automated Change
* fix: component
* fix: getConversationInfo
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* fix: minio config url recognition error
* update gomake version
* update gomake version
* fix: seq conversion bug
* fix: redis pipe exec
* fix: ImportFriends
* fix: A large number of logs keysAndValues length is not even
* feat: mark read aggregate write
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* merge
* merge
* read seq is written to mongo
* read seq is written to mongo
* fix: invitation to join group notification
* fix: friend op_user_id
* feat: optimizing asynchronous context
* feat: optimizing memamq size
* feat: add GetSeqMessage
* feat: GroupApplicationAgreeMemberEnterNotification
* feat: GroupApplicationAgreeMemberEnterNotification
* feat: go.mod
* feat: go.mod
* feat: join group notification and get seq
* feat: join group notification and get seq
* feat: avoid pulling messages from sessions with a large number of max seq values of 0
* feat: API supports gzip
* go.mod
* fix: nil pointer error on close
* fix: listen error
* fix: listen error
* update go.mod
---------
Co-authored-by: withchao <withchao@users.noreply.github.com>
* feat: improve searchMsg implement. (#2614)
* refactor: refactor workflows contents.
* add tool workflows.
* update field.
* fix: remove chat error.
* Fix err.
* fix error.
* remove cn comment.
* update workflows files.
* update infra config.
* move workflows.
* feat: update bot.
* fix: solve uncorrect outdated msg get.
* update get docIDs logic.
* update
* update skip logic.
* fix
* update.
* fix: delay deleteObject func.
* remove unused content.
* update log type.
* feat: implement request batch count limit.
* update
* update
* remove unused script.
* feat: improve searchMsg implement.
* update mongo config.
* Fix lock (#2622)
* fix:log
* fix: lock
* fix: update setGroupInfoEX field name. (#2625)
* refactor: refactor workflows contents.
* add tool workflows.
* update field.
* fix: remove chat error.
* Fix err.
* fix error.
* remove cn comment.
* update workflows files.
* update infra config.
* move workflows.
* feat: update bot.
* fix: solve uncorrect outdated msg get.
* update get docIDs logic.
* update
* update skip logic.
* fix
* update.
* fix: delay deleteObject func.
* remove unused content.
* update log type.
* feat: implement request batch count limit.
* update
* update
* fix: update setGroupInfoEX field name.
* fix: update setGroupInfoEX field name (#2626)
* refactor: refactor workflows contents.
* add tool workflows.
* update field.
* fix: remove chat error.
* Fix err.
* fix error.
* remove cn comment.
* update workflows files.
* update infra config.
* move workflows.
* feat: update bot.
* fix: solve uncorrect outdated msg get.
* update get docIDs logic.
* update
* update skip logic.
* fix
* update.
* fix: delay deleteObject func.
* remove unused content.
* update log type.
* feat: implement request batch count limit.
* update
* update
* fix: update setGroupInfoEX field name.
* fix: update setGroupInfoEX field name
* feat: msg gateway add log (#2631)
* fix: GroupApplicationAcceptedNotification
* fix: GroupApplicationAcceptedNotification
* fix: NotificationUserInfoUpdate
* cicd: robot automated Change
* fix: component
* fix: getConversationInfo
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* fix: minio config url recognition error
* update gomake version
* update gomake version
* fix: seq conversion bug
* fix: redis pipe exec
* fix: ImportFriends
* fix: A large number of logs keysAndValues length is not even
* feat: mark read aggregate write
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* merge
* merge
* read seq is written to mongo
* read seq is written to mongo
* fix: invitation to join group notification
* fix: friend op_user_id
* feat: optimizing asynchronous context
* feat: optimizing memamq size
* feat: add GetSeqMessage
* feat: GroupApplicationAgreeMemberEnterNotification
* feat: GroupApplicationAgreeMemberEnterNotification
* feat: go.mod
* feat: go.mod
* feat: join group notification and get seq
* feat: join group notification and get seq
* feat: avoid pulling messages from sessions with a large number of max seq values of 0
* feat: API supports gzip
* go.mod
* fix: nil pointer error on close
* fix: listen error
* fix: listen error
* update go.mod
* feat: add log
---------
Co-authored-by: withchao <withchao@users.noreply.github.com>
* fix: update setGroupInfoEx func name and field. (#2634)
* refactor: refactor workflows contents.
* add tool workflows.
* update field.
* fix: remove chat error.
* Fix err.
* fix error.
* remove cn comment.
* update workflows files.
* update infra config.
* move workflows.
* feat: update bot.
* fix: solve uncorrect outdated msg get.
* update get docIDs logic.
* update
* update skip logic.
* fix
* update.
* fix: delay deleteObject func.
* remove unused content.
* update log type.
* feat: implement request batch count limit.
* update
* update
* fix: update setGroupInfoEx func name and field.
* refactor: update groupinfoEx field.
* refactor: update database name in mongodb.yml
* add groupName Condition
* fix: fix setConversations req fill. (#2645)
* refactor: refactor workflows contents.
* add tool workflows.
* update field.
* fix: remove chat error.
* Fix err.
* fix error.
* remove cn comment.
* update workflows files.
* update infra config.
* move workflows.
* feat: update bot.
* fix: solve uncorrect outdated msg get.
* update get docIDs logic.
* update
* update skip logic.
* fix
* update.
* fix: delay deleteObject func.
* remove unused content.
* update log type.
* feat: implement request batch count limit.
* update
* update
* fix: fix setConversations req fill.
* fix: GetMsgBySeqs boundary issues (#2647)
* fix: GroupApplicationAcceptedNotification
* fix: GroupApplicationAcceptedNotification
* fix: NotificationUserInfoUpdate
* cicd: robot automated Change
* fix: component
* fix: getConversationInfo
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* fix: minio config url recognition error
* update gomake version
* update gomake version
* fix: seq conversion bug
* fix: redis pipe exec
* fix: ImportFriends
* fix: A large number of logs keysAndValues length is not even
* feat: mark read aggregate write
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* merge
* merge
* read seq is written to mongo
* read seq is written to mongo
* fix: invitation to join group notification
* fix: friend op_user_id
* feat: optimizing asynchronous context
* feat: optimizing memamq size
* feat: add GetSeqMessage
* feat: GroupApplicationAgreeMemberEnterNotification
* feat: GroupApplicationAgreeMemberEnterNotification
* feat: go.mod
* feat: go.mod
* feat: join group notification and get seq
* feat: join group notification and get seq
* feat: avoid pulling messages from sessions with a large number of max seq values of 0
* feat: API supports gzip
* go.mod
* fix: nil pointer error on close
* fix: listen error
* fix: listen error
* update go.mod
* feat: add log
* fix: token parse token value
* fix: GetMsgBySeqs boundary issues
---------
Co-authored-by: withchao <withchao@users.noreply.github.com>
* fix: the attribute version is obsolete, remove it (#2644)
* refactor: update Userregister request field. (#2650)
* Test Workflow (#2640)
* feat: cicd
* feat: cicd
* fix: cicd
* fix: cicd
* fix: cicd
* fix: cicd
* fix: kick group member callback (#2643)
* fix: kill group member callback
* fix: change port
* fix: change port
* fix: route (#2654)
* feat: add GetSpecifiedBlacks interface. (#2656)
* Upgrade the Google Firebase version. (#2638)
* Fix token (#2653)
* fix: kick token
* fix: kick token
* fix: change config
* feat: get not notify conversationIDs (#2658)
* feat: get not notify conversationIDs
* feat: api
* fix: database
* fix: change name
* feat: GetPinnedConversationIDs (#2660)
* feat: GetPinnedConversationIDs
* feat: api
* Upgrade the FCM SDK to version 4, and use the SendEach method instead of the SendAll method. (#2633)
Co-authored-by: Monet Lee <monet_lee@163.com>
* feat: implement GetSpecifiedUserGroupRequestInfo interface. (#2661)
* feat: implement GetSpecifiedUserGroupRequestInfo interface.
* update mongo config.
* feat: provide the interface required by js sdk (#2664)
* fix: redis support acquisition time
* fix: GetActiveConversation
* feat: jssdk GetConversations, GetActiveConversation
* feat: jssdk GetConversations, GetActiveConversation
* feat: jssdk GetConversations, GetActiveConversation
* feat: jssdk GetConversations, GetActiveConversation
* feat: jssdk GetConversations, GetActiveConversation
* feat: improve get admin role memberIDs implement. (#2666)
* feat: implement GetSpecifiedUserGroupRequestInfo interface.
* update mongo config.
* feat: improve get admin role memberIDs implement.
* remove unused contents.
* remove unused contents.
* remove todo implement.
* fix: fix update groupName invalid. (#2673)
* refactor: change platform to platformID (#2670)
* feat: don`t return nil data (#2675)
Co-authored-by: Monet Lee <monet_lee@163.com>
* refactor: update fields type in userStatus and check registered. (#2676)
* fix: usertoken auth. (#2677)
* refactor: update fields type in userStatus and check registered.
* fix: usertoken auth.
* update contents.
* update content.
* update
* fix
* update pb file.
* feat: add friend agree after callback (#2680)
* fix: sn not sort (#2682)
* fix: GroupApplicationAcceptedNotification
* fix: GroupApplicationAcceptedNotification
* fix: NotificationUserInfoUpdate
* cicd: robot automated Change
* fix: component
* fix: getConversationInfo
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* feat: cron task
* fix: minio config url recognition error
* update gomake version
* update gomake version
* fix: seq conversion bug
* fix: redis pipe exec
* fix: ImportFriends
* fix: A large number of logs keysAndValues length is not even
* feat: mark read aggregate write
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* feat: online status supports redis cluster
* merge
* merge
* read seq is written to mongo
* read seq is written to mongo
* fix: invitation to join group notification
* fix: friend op_user_id
* feat: optimizing asynchronous context
* feat: optimizing memamq size
* feat: add GetSeqMessage
* feat: GroupApplicationAgreeMemberEnterNotification
* feat: GroupApplicationAgreeMemberEnterNotification
* feat: go.mod
* feat: go.mod
* feat: join group notification and get seq
* feat: join group notification and get seq
* feat: avoid pulling messages from sessions with a large number of max seq values of 0
* feat: API supports gzip
* go.mod
* fix: nil pointer error on close
* fix: listen error
* fix: listen error
* update go.mod
* feat: add log
* fix: token parse token value
* fix: GetMsgBySeqs boundary issues
* fix: sn_ not sort
* fix: sn_ not sort
* fix: sn_ not sort
---------
Co-authored-by: withchao <withchao@users.noreply.github.com>
* refactor: add GetAdminToken interface. (#2684)
* refactor: add GetAdminToken interface.
* update config.
* fix: admin token (#2686)
* fix: update workflows logic. (#2688)
* refactor: add GetAdminToken interface.
* update config.
* update workflows logic.
* fix: admin token (#2687)
* update the front image (#2692)
* update the front image
* update version
* feat: improve publish docker image workflows (#2697)
* refactor: add GetAdminToken interface.
* update config.
* update workflows logic.
* feat: improve publish docker image workflows
* update condition logic.
* fix: update load file logic. (#2700)
* refactor: add GetAdminToken interface.
* update config.
* update workflows logic.
* feat: improve publish docker image workflows
* update condition logic.
* fix: update load file logic.
* feat: Msg filter (#2703)
* feat: msg filter
* feat: msg filter
* feat: msg filter
---------
Co-authored-by: Monet Lee <monet_lee@163.com>
Co-authored-by: icey-yu <119291641+icey-yu@users.noreply.github.com>
Co-authored-by: Kevin Lee <59052025+lgz5689@users.noreply.github.com>
Co-authored-by: qinguoyi <1532979219@qq.com>
Co-authored-by: chao <48119764+withchao@users.noreply.github.com>
Co-authored-by: withchao <withchao@users.noreply.github.com>
Co-authored-by: Mew151 <dgqypl@gmail.com>
Co-authored-by: 蔡相跃 <caixiangyue007@gmail.com>
Co-authored-by: 暴走的大猩猩 <Hz_testing@163.com>
Co-authored-by: Libo <zhaolibo1989@foxmail.com>
Co-authored-by: skiffer-git <72860476+skiffer-git@users.noreply.github.com>
This commit is contained in:
OpenIM-Gordon
@@ -0,0 +1,29 @@
|
||||
package push
|
||||
|
||||
import (
|
||||
"github.com/openimsdk/protocol/sdkws"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestName(t *testing.T) {
|
||||
var c ConsumerHandler
|
||||
c.readCh = make(chan *sdkws.MarkAsReadTips)
|
||||
|
||||
go c.loopRead()
|
||||
|
||||
go func() {
|
||||
for i := 0; ; i++ {
|
||||
seq := int64(i + 1)
|
||||
if seq%3 == 0 {
|
||||
seq = 1
|
||||
}
|
||||
c.readCh <- &sdkws.MarkAsReadTips{
|
||||
ConversationID: "c100",
|
||||
MarkAsReadUserID: "u100",
|
||||
HasReadSeq: seq,
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
select {}
|
||||
}
|
||||
@@ -17,6 +17,7 @@ package dummy
|
||||
import (
|
||||
"context"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
|
||||
"github.com/openimsdk/tools/log"
|
||||
)
|
||||
|
||||
func NewClient() *Dummy {
|
||||
@@ -27,5 +28,6 @@ type Dummy struct {
|
||||
}
|
||||
|
||||
func (d *Dummy) Push(ctx context.Context, userIDs []string, title, content string, opts *options.Opts) error {
|
||||
log.ZDebug(ctx, "dummy push")
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -22,8 +22,8 @@ import (
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
firebase "firebase.google.com/go"
|
||||
"firebase.google.com/go/messaging"
|
||||
firebase "firebase.google.com/go/v4"
|
||||
"firebase.google.com/go/v4/messaging"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
|
||||
"github.com/openimsdk/protocol/constant"
|
||||
@@ -99,7 +99,7 @@ func (f *Fcm) Push(ctx context.Context, userIDs []string, title, content string,
|
||||
apns := &messaging.APNSConfig{Payload: &messaging.APNSPayload{Aps: &messaging.Aps{Sound: opts.IOSPushSound}}}
|
||||
messageCount := len(messages)
|
||||
if messageCount >= SinglePushCountLimit {
|
||||
response, err := f.fcmMsgCli.SendAll(ctx, messages)
|
||||
response, err := f.fcmMsgCli.SendEach(ctx, messages)
|
||||
if err != nil {
|
||||
Fail = Fail + messageCount
|
||||
// Record push error
|
||||
@@ -154,7 +154,7 @@ func (f *Fcm) Push(ctx context.Context, userIDs []string, title, content string,
|
||||
}
|
||||
messageCount := len(messages)
|
||||
if messageCount > 0 {
|
||||
response, err := f.fcmMsgCli.SendAll(ctx, messages)
|
||||
response, err := f.fcmMsgCli.SendEach(ctx, messages)
|
||||
if err != nil {
|
||||
Fail = Fail + messageCount
|
||||
} else {
|
||||
|
||||
@@ -18,11 +18,11 @@ import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache"
|
||||
"github.com/openimsdk/tools/errs"
|
||||
@@ -91,6 +91,15 @@ func (g *Client) Push(ctx context.Context, userIDs []string, title, content stri
|
||||
for i, v := range s.GetSplitResult() {
|
||||
go func(index int, userIDs []string) {
|
||||
defer wg.Done()
|
||||
for i := 0; i < len(userIDs); i += maxNum {
|
||||
end := i + maxNum
|
||||
if end > len(userIDs) {
|
||||
end = len(userIDs)
|
||||
}
|
||||
if err = g.batchPush(ctx, token, userIDs[i:end], pushReq); err != nil {
|
||||
log.ZError(ctx, "batchPush failed", err, "index", index, "token", token, "req", pushReq)
|
||||
}
|
||||
}
|
||||
if err = g.batchPush(ctx, token, userIDs, pushReq); err != nil {
|
||||
log.ZError(ctx, "batchPush failed", err, "index", index, "token", token, "req", pushReq)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,122 @@
|
||||
package push
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/IBM/sarama"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
|
||||
"github.com/openimsdk/protocol/constant"
|
||||
pbpush "github.com/openimsdk/protocol/push"
|
||||
"github.com/openimsdk/protocol/sdkws"
|
||||
"github.com/openimsdk/tools/errs"
|
||||
"github.com/openimsdk/tools/log"
|
||||
"github.com/openimsdk/tools/mq/kafka"
|
||||
"github.com/openimsdk/tools/utils/jsonutil"
|
||||
"google.golang.org/protobuf/proto"
|
||||
)
|
||||
|
||||
type OfflinePushConsumerHandler struct {
|
||||
OfflinePushConsumerGroup *kafka.MConsumerGroup
|
||||
offlinePusher offlinepush.OfflinePusher
|
||||
}
|
||||
|
||||
func NewOfflinePushConsumerHandler(config *Config, offlinePusher offlinepush.OfflinePusher) (*OfflinePushConsumerHandler, error) {
|
||||
var offlinePushConsumerHandler OfflinePushConsumerHandler
|
||||
var err error
|
||||
offlinePushConsumerHandler.offlinePusher = offlinePusher
|
||||
offlinePushConsumerHandler.OfflinePushConsumerGroup, err = kafka.NewMConsumerGroup(config.KafkaConfig.Build(), config.KafkaConfig.ToOfflineGroupID,
|
||||
[]string{config.KafkaConfig.ToOfflinePushTopic}, true)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &offlinePushConsumerHandler, nil
|
||||
}
|
||||
|
||||
func (*OfflinePushConsumerHandler) Setup(sarama.ConsumerGroupSession) error { return nil }
|
||||
func (*OfflinePushConsumerHandler) Cleanup(sarama.ConsumerGroupSession) error { return nil }
|
||||
func (o *OfflinePushConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
|
||||
for msg := range claim.Messages() {
|
||||
ctx := o.OfflinePushConsumerGroup.GetContextFromMsg(msg)
|
||||
o.handleMsg2OfflinePush(ctx, msg.Value)
|
||||
sess.MarkMessage(msg, "")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (o *OfflinePushConsumerHandler) handleMsg2OfflinePush(ctx context.Context, msg []byte) {
|
||||
offlinePushMsg := pbpush.PushMsgReq{}
|
||||
if err := proto.Unmarshal(msg, &offlinePushMsg); err != nil {
|
||||
log.ZError(ctx, "offline push Unmarshal msg err", err, "msg", string(msg))
|
||||
return
|
||||
}
|
||||
if offlinePushMsg.MsgData == nil || offlinePushMsg.UserIDs == nil {
|
||||
log.ZError(ctx, "offline push msg is empty", errs.New("offlinePushMsg is empty"), "userIDs", offlinePushMsg.UserIDs, "msg", offlinePushMsg.MsgData)
|
||||
return
|
||||
}
|
||||
log.ZInfo(ctx, "receive to OfflinePush MQ", "userIDs", offlinePushMsg.UserIDs, "msg", offlinePushMsg.MsgData)
|
||||
|
||||
err := o.offlinePushMsg(ctx, offlinePushMsg.MsgData, offlinePushMsg.UserIDs)
|
||||
if err != nil {
|
||||
log.ZWarn(ctx, "offline push failed", err, "msg", offlinePushMsg.String())
|
||||
}
|
||||
}
|
||||
|
||||
func (o *OfflinePushConsumerHandler) getOfflinePushInfos(msg *sdkws.MsgData) (title, content string, opts *options.Opts, err error) {
|
||||
type AtTextElem struct {
|
||||
Text string `json:"text,omitempty"`
|
||||
AtUserList []string `json:"atUserList,omitempty"`
|
||||
IsAtSelf bool `json:"isAtSelf"`
|
||||
}
|
||||
|
||||
opts = &options.Opts{Signal: &options.Signal{}}
|
||||
if msg.OfflinePushInfo != nil {
|
||||
opts.IOSBadgeCount = msg.OfflinePushInfo.IOSBadgeCount
|
||||
opts.IOSPushSound = msg.OfflinePushInfo.IOSPushSound
|
||||
opts.Ex = msg.OfflinePushInfo.Ex
|
||||
}
|
||||
|
||||
if msg.OfflinePushInfo != nil {
|
||||
title = msg.OfflinePushInfo.Title
|
||||
content = msg.OfflinePushInfo.Desc
|
||||
}
|
||||
if title == "" {
|
||||
switch msg.ContentType {
|
||||
case constant.Text:
|
||||
fallthrough
|
||||
case constant.Picture:
|
||||
fallthrough
|
||||
case constant.Voice:
|
||||
fallthrough
|
||||
case constant.Video:
|
||||
fallthrough
|
||||
case constant.File:
|
||||
title = constant.ContentType2PushContent[int64(msg.ContentType)]
|
||||
case constant.AtText:
|
||||
ac := AtTextElem{}
|
||||
_ = jsonutil.JsonStringToStruct(string(msg.Content), &ac)
|
||||
case constant.SignalingNotification:
|
||||
title = constant.ContentType2PushContent[constant.SignalMsg]
|
||||
default:
|
||||
title = constant.ContentType2PushContent[constant.Common]
|
||||
}
|
||||
}
|
||||
if content == "" {
|
||||
content = title
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (o *OfflinePushConsumerHandler) offlinePushMsg(ctx context.Context, msg *sdkws.MsgData, offlinePushUserIDs []string) error {
|
||||
title, content, opts, err := o.getOfflinePushInfos(msg)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = o.offlinePusher.Push(ctx, offlinePushUserIDs, title, content, opts)
|
||||
if err != nil {
|
||||
prommetrics.MsgOfflinePushFailedCounter.Inc()
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -19,20 +19,20 @@ type OnlinePusher interface {
|
||||
pushToUserIDs *[]string) []string
|
||||
}
|
||||
|
||||
type emptyOnlinePUsher struct{}
|
||||
type emptyOnlinePusher struct{}
|
||||
|
||||
func newEmptyOnlinePUsher() *emptyOnlinePUsher {
|
||||
return &emptyOnlinePUsher{}
|
||||
func newEmptyOnlinePusher() *emptyOnlinePusher {
|
||||
return &emptyOnlinePusher{}
|
||||
}
|
||||
|
||||
func (emptyOnlinePUsher) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.MsgData,
|
||||
func (emptyOnlinePusher) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.MsgData,
|
||||
pushToUserIDs []string) (wsResults []*msggateway.SingleMsgToUserResults, err error) {
|
||||
log.ZWarn(ctx, "emptyOnlinePUsher GetConnsAndOnlinePush", nil)
|
||||
log.ZInfo(ctx, "emptyOnlinePusher GetConnsAndOnlinePush", nil)
|
||||
return nil, nil
|
||||
}
|
||||
func (u emptyOnlinePUsher) GetOnlinePushFailedUserIDs(ctx context.Context, msg *sdkws.MsgData,
|
||||
func (u emptyOnlinePusher) GetOnlinePushFailedUserIDs(ctx context.Context, msg *sdkws.MsgData,
|
||||
wsResults []*msggateway.SingleMsgToUserResults, pushToUserIDs *[]string) []string {
|
||||
log.ZWarn(ctx, "emptyOnlinePUsher GetOnlinePushFailedUserIDs", nil)
|
||||
log.ZInfo(ctx, "emptyOnlinePusher GetOnlinePushFailedUserIDs", nil)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -45,7 +45,7 @@ func NewOnlinePusher(disCov discovery.SvcDiscoveryRegistry, config *Config) Onli
|
||||
case "etcd":
|
||||
return NewDefaultAllNode(disCov, config)
|
||||
default:
|
||||
return newEmptyOnlinePUsher()
|
||||
return newEmptyOnlinePusher()
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+16
-3
@@ -2,6 +2,7 @@ package push
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis"
|
||||
@@ -17,12 +18,12 @@ type pushServer struct {
|
||||
disCov discovery.SvcDiscoveryRegistry
|
||||
offlinePusher offlinepush.OfflinePusher
|
||||
pushCh *ConsumerHandler
|
||||
offlinePushCh *OfflinePushConsumerHandler
|
||||
}
|
||||
|
||||
type Config struct {
|
||||
RpcConfig config.Push
|
||||
RedisConfig config.Redis
|
||||
MongodbConfig config.Mongo
|
||||
KafkaConfig config.Kafka
|
||||
NotificationConfig config.Notification
|
||||
Share config.Share
|
||||
@@ -55,18 +56,30 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
database := controller.NewPushDatabase(cacheModel)
|
||||
|
||||
consumer, err := NewConsumerHandler(config, offlinePusher, rdb, client)
|
||||
database := controller.NewPushDatabase(cacheModel, &config.KafkaConfig)
|
||||
|
||||
consumer, err := NewConsumerHandler(config, database, offlinePusher, rdb, client)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
offlinePushConsumer, err := NewOfflinePushConsumerHandler(config, offlinePusher)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
pbpush.RegisterPushMsgServiceServer(server, &pushServer{
|
||||
database: database,
|
||||
disCov: client,
|
||||
offlinePusher: offlinePusher,
|
||||
pushCh: consumer,
|
||||
offlinePushCh: offlinePushConsumer,
|
||||
})
|
||||
|
||||
go consumer.pushConsumerGroup.RegisterHandleAndConsumer(ctx, consumer)
|
||||
|
||||
go offlinePushConsumer.OfflinePushConsumerGroup.RegisterHandleAndConsumer(ctx, offlinePushConsumer)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -1,33 +1,20 @@
|
||||
// Copyright © 2023 OpenIM. All rights reserved.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package push
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
|
||||
"github.com/IBM/sarama"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush"
|
||||
"github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/storage/controller"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/webhook"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/rpccache"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/util/conversationutil"
|
||||
"github.com/openimsdk/protocol/constant"
|
||||
pbchat "github.com/openimsdk/protocol/msg"
|
||||
"github.com/openimsdk/protocol/msggateway"
|
||||
pbpush "github.com/openimsdk/protocol/push"
|
||||
"github.com/openimsdk/protocol/sdkws"
|
||||
@@ -40,12 +27,16 @@ import (
|
||||
"github.com/openimsdk/tools/utils/timeutil"
|
||||
"github.com/redis/go-redis/v9"
|
||||
"google.golang.org/protobuf/proto"
|
||||
"math/rand"
|
||||
"strconv"
|
||||
"time"
|
||||
)
|
||||
|
||||
type ConsumerHandler struct {
|
||||
pushConsumerGroup *kafka.MConsumerGroup
|
||||
offlinePusher offlinepush.OfflinePusher
|
||||
onlinePusher OnlinePusher
|
||||
pushDatabase controller.PushDatabase
|
||||
onlineCache *rpccache.OnlineCache
|
||||
groupLocalCache *rpccache.GroupLocalCache
|
||||
conversationLocalCache *rpccache.ConversationLocalCache
|
||||
@@ -56,7 +47,7 @@ type ConsumerHandler struct {
|
||||
config *Config
|
||||
}
|
||||
|
||||
func NewConsumerHandler(config *Config, offlinePusher offlinepush.OfflinePusher, rdb redis.UniversalClient,
|
||||
func NewConsumerHandler(config *Config, database controller.PushDatabase, offlinePusher offlinepush.OfflinePusher, rdb redis.UniversalClient,
|
||||
client discovery.SvcDiscoveryRegistry) (*ConsumerHandler, error) {
|
||||
var consumerHandler ConsumerHandler
|
||||
var err error
|
||||
@@ -65,7 +56,9 @@ func NewConsumerHandler(config *Config, offlinePusher offlinepush.OfflinePusher,
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
userRpcClient := rpcclient.NewUserRpcClient(client, config.Share.RpcRegisterName.User, config.Share.IMAdminUserID)
|
||||
|
||||
consumerHandler.offlinePusher = offlinePusher
|
||||
consumerHandler.onlinePusher = NewOnlinePusher(client, config)
|
||||
consumerHandler.groupRpcClient = rpcclient.NewGroupRpcClient(client, config.Share.RpcRegisterName.Group)
|
||||
@@ -75,42 +68,45 @@ func NewConsumerHandler(config *Config, offlinePusher offlinepush.OfflinePusher,
|
||||
consumerHandler.conversationLocalCache = rpccache.NewConversationLocalCache(consumerHandler.conversationRpcClient, &config.LocalCacheConfig, rdb)
|
||||
consumerHandler.webhookClient = webhook.NewWebhookClient(config.WebhooksConfig.URL)
|
||||
consumerHandler.config = config
|
||||
consumerHandler.onlineCache = rpccache.NewOnlineCache(userRpcClient, consumerHandler.groupLocalCache, rdb, nil)
|
||||
consumerHandler.pushDatabase = database
|
||||
consumerHandler.onlineCache, err = rpccache.NewOnlineCache(userRpcClient, consumerHandler.groupLocalCache, rdb, config.RpcConfig.FullUserCache, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &consumerHandler, nil
|
||||
}
|
||||
|
||||
func (c *ConsumerHandler) handleMs2PsChat(ctx context.Context, msg []byte) {
|
||||
msgFromMQ := pbchat.PushMsgDataToMQ{}
|
||||
msgFromMQ := pbpush.PushMsgReq{}
|
||||
if err := proto.Unmarshal(msg, &msgFromMQ); err != nil {
|
||||
log.ZError(ctx, "push Unmarshal msg err", err, "msg", string(msg))
|
||||
return
|
||||
}
|
||||
pbData := &pbpush.PushMsgReq{
|
||||
MsgData: msgFromMQ.MsgData,
|
||||
ConversationID: msgFromMQ.ConversationID,
|
||||
}
|
||||
|
||||
sec := msgFromMQ.MsgData.SendTime / 1000
|
||||
nowSec := timeutil.GetCurrentTimestampBySecond()
|
||||
log.ZDebug(ctx, "push msg", "msg", pbData.String(), "sec", sec, "nowSec", nowSec)
|
||||
|
||||
if nowSec-sec > 10 {
|
||||
return
|
||||
prommetrics.MsgLoneTimePushCounter.Inc()
|
||||
log.ZWarn(ctx, "it’s been a while since the message was sent", nil, "msg", msgFromMQ.String(), "sec", sec, "nowSec", nowSec, "nowSec-sec", nowSec-sec)
|
||||
}
|
||||
var err error
|
||||
|
||||
switch msgFromMQ.MsgData.SessionType {
|
||||
case constant.ReadGroupChatType:
|
||||
err = c.Push2Group(ctx, pbData.MsgData.GroupID, pbData.MsgData)
|
||||
err = c.Push2Group(ctx, msgFromMQ.MsgData.GroupID, msgFromMQ.MsgData)
|
||||
default:
|
||||
var pushUserIDList []string
|
||||
isSenderSync := datautil.GetSwitchFromOptions(pbData.MsgData.Options, constant.IsSenderSync)
|
||||
if !isSenderSync || pbData.MsgData.SendID == pbData.MsgData.RecvID {
|
||||
pushUserIDList = append(pushUserIDList, pbData.MsgData.RecvID)
|
||||
isSenderSync := datautil.GetSwitchFromOptions(msgFromMQ.MsgData.Options, constant.IsSenderSync)
|
||||
if !isSenderSync || msgFromMQ.MsgData.SendID == msgFromMQ.MsgData.RecvID {
|
||||
pushUserIDList = append(pushUserIDList, msgFromMQ.MsgData.RecvID)
|
||||
} else {
|
||||
pushUserIDList = append(pushUserIDList, pbData.MsgData.RecvID, pbData.MsgData.SendID)
|
||||
pushUserIDList = append(pushUserIDList, msgFromMQ.MsgData.RecvID, msgFromMQ.MsgData.SendID)
|
||||
}
|
||||
err = c.Push2User(ctx, pushUserIDList, pbData.MsgData)
|
||||
err = c.Push2User(ctx, pushUserIDList, msgFromMQ.MsgData)
|
||||
}
|
||||
if err != nil {
|
||||
log.ZWarn(ctx, "push failed", err, "msg", pbData.String())
|
||||
log.ZWarn(ctx, "push failed", err, "msg", msgFromMQ.String())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -119,6 +115,14 @@ func (*ConsumerHandler) Setup(sarama.ConsumerGroupSession) error { return nil }
|
||||
func (*ConsumerHandler) Cleanup(sarama.ConsumerGroupSession) error { return nil }
|
||||
|
||||
func (c *ConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
|
||||
c.onlineCache.Lock.Lock()
|
||||
for c.onlineCache.CurrentPhase.Load() < rpccache.DoSubscribeOver {
|
||||
c.onlineCache.Cond.Wait()
|
||||
}
|
||||
c.onlineCache.Lock.Unlock()
|
||||
ctx := mcontext.SetOperationID(context.TODO(), strconv.FormatInt(time.Now().UnixNano()+int64(rand.Uint32()), 10))
|
||||
log.ZInfo(ctx, "begin consume messages")
|
||||
|
||||
for msg := range claim.Messages() {
|
||||
ctx := c.pushConsumerGroup.GetContextFromMsg(msg)
|
||||
c.handleMs2PsChat(ctx, msg.Value)
|
||||
@@ -129,20 +133,27 @@ func (c *ConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim s
|
||||
|
||||
// Push2User Suitable for two types of conversations, one is SingleChatType and the other is NotificationChatType.
|
||||
func (c *ConsumerHandler) Push2User(ctx context.Context, userIDs []string, msg *sdkws.MsgData) (err error) {
|
||||
log.ZDebug(ctx, "Get msg from msg_transfer And push msg", "userIDs", userIDs, "msg", msg.String())
|
||||
log.ZInfo(ctx, "Get msg from msg_transfer And push msg", "userIDs", userIDs, "msg", msg.String())
|
||||
defer func(duration time.Time) {
|
||||
t := time.Since(duration)
|
||||
log.ZInfo(ctx, "Get msg from msg_transfer And push msg", "msg", msg.String(), "time cost", t)
|
||||
}(time.Now())
|
||||
if err := c.webhookBeforeOnlinePush(ctx, &c.config.WebhooksConfig.BeforeOnlinePush, userIDs, msg); err != nil {
|
||||
return err
|
||||
}
|
||||
log.ZInfo(ctx, "webhookBeforeOnlinePush end")
|
||||
|
||||
wsResults, err := c.GetConnsAndOnlinePush(ctx, msg, userIDs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
log.ZDebug(ctx, "single and notification push result", "result", wsResults, "msg", msg, "push_to_userID", userIDs)
|
||||
log.ZInfo(ctx, "single and notification push result", "result", wsResults, "msg", msg, "push_to_userID", userIDs)
|
||||
|
||||
if !c.shouldPushOffline(ctx, msg) {
|
||||
return nil
|
||||
}
|
||||
log.ZInfo(ctx, "shouldPushOffline end")
|
||||
|
||||
for _, v := range wsResults {
|
||||
//message sender do not need offline push
|
||||
@@ -154,17 +165,17 @@ func (c *ConsumerHandler) Push2User(ctx context.Context, userIDs []string, msg *
|
||||
return nil
|
||||
}
|
||||
}
|
||||
offlinePUshUserID := []string{msg.RecvID}
|
||||
offlinePushUserID := []string{msg.RecvID}
|
||||
|
||||
//receiver offline push
|
||||
if err = c.webhookBeforeOfflinePush(ctx, &c.config.WebhooksConfig.BeforeOfflinePush,
|
||||
offlinePUshUserID, msg, nil); err != nil {
|
||||
offlinePushUserID, msg, nil); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
err = c.offlinePushMsg(ctx, msg, offlinePUshUserID)
|
||||
log.ZInfo(ctx, "webhookBeforeOfflinePush end")
|
||||
err = c.offlinePushMsg(ctx, msg, offlinePushUserID)
|
||||
if err != nil {
|
||||
log.ZWarn(ctx, "offlinePushMsg failed", err, "offlinePUshUserID", offlinePUshUserID, "msg", msg)
|
||||
log.ZWarn(ctx, "offlinePushMsg failed", err, "offlinePushUserID", offlinePushUserID, "msg", msg)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -183,21 +194,11 @@ func (c *ConsumerHandler) shouldPushOffline(_ context.Context, msg *sdkws.MsgDat
|
||||
}
|
||||
|
||||
func (c *ConsumerHandler) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.MsgData, pushToUserIDs []string) ([]*msggateway.SingleMsgToUserResults, error) {
|
||||
var (
|
||||
onlineUserIDs []string
|
||||
offlineUserIDs []string
|
||||
)
|
||||
for _, userID := range pushToUserIDs {
|
||||
online, err := c.onlineCache.GetUserOnline(ctx, userID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if online {
|
||||
onlineUserIDs = append(onlineUserIDs, userID)
|
||||
} else {
|
||||
offlineUserIDs = append(offlineUserIDs, userID)
|
||||
}
|
||||
onlineUserIDs, offlineUserIDs, err := c.onlineCache.GetUsersOnline(ctx, pushToUserIDs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
log.ZDebug(ctx, "GetConnsAndOnlinePush online cache", "sendID", msg.SendID, "recvID", msg.RecvID, "groupID", msg.GroupID, "sessionType", msg.SessionType, "clientMsgID", msg.ClientMsgID, "serverMsgID", msg.ServerMsgID, "offlineUserIDs", offlineUserIDs, "onlineUserIDs", onlineUserIDs)
|
||||
var result []*msggateway.SingleMsgToUserResults
|
||||
if len(onlineUserIDs) > 0 {
|
||||
@@ -216,57 +217,70 @@ func (c *ConsumerHandler) GetConnsAndOnlinePush(ctx context.Context, msg *sdkws.
|
||||
}
|
||||
|
||||
func (c *ConsumerHandler) Push2Group(ctx context.Context, groupID string, msg *sdkws.MsgData) (err error) {
|
||||
log.ZDebug(ctx, "Get group msg from msg_transfer and push msg", "msg", msg.String(), "groupID", groupID)
|
||||
log.ZInfo(ctx, "Get group msg from msg_transfer and push msg", "msg", msg.String(), "groupID", groupID)
|
||||
defer func(duration time.Time) {
|
||||
t := time.Since(duration)
|
||||
log.ZInfo(ctx, "Get group msg from msg_transfer and push msg end", "msg", msg.String(), "groupID", groupID, "time cost", t)
|
||||
}(time.Now())
|
||||
var pushToUserIDs []string
|
||||
if err = c.webhookBeforeGroupOnlinePush(ctx, &c.config.WebhooksConfig.BeforeGroupOnlinePush, groupID, msg,
|
||||
&pushToUserIDs); err != nil {
|
||||
return err
|
||||
}
|
||||
log.ZInfo(ctx, "webhookBeforeGroupOnlinePush end")
|
||||
|
||||
err = c.groupMessagesHandler(ctx, groupID, &pushToUserIDs, msg)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
log.ZInfo(ctx, "groupMessagesHandler end")
|
||||
|
||||
wsResults, err := c.GetConnsAndOnlinePush(ctx, msg, pushToUserIDs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
log.ZDebug(ctx, "group push result", "result", wsResults, "msg", msg)
|
||||
log.ZInfo(ctx, "group push result", "result", wsResults, "msg", msg)
|
||||
|
||||
if !c.shouldPushOffline(ctx, msg) {
|
||||
return nil
|
||||
}
|
||||
needOfflinePushUserIDs := c.onlinePusher.GetOnlinePushFailedUserIDs(ctx, msg, wsResults, &pushToUserIDs)
|
||||
|
||||
log.ZInfo(ctx, "GetOnlinePushFailedUserIDs end")
|
||||
//filter some user, like don not disturb or don't need offline push etc.
|
||||
needOfflinePushUserIDs, err = c.filterGroupMessageOfflinePush(ctx, groupID, msg, needOfflinePushUserIDs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
log.ZInfo(ctx, "filterGroupMessageOfflinePush end")
|
||||
|
||||
// Use offline push messaging
|
||||
if len(needOfflinePushUserIDs) > 0 {
|
||||
var offlinePushUserIDs []string
|
||||
err = c.webhookBeforeOfflinePush(ctx, &c.config.WebhooksConfig.BeforeOfflinePush, needOfflinePushUserIDs, msg, &offlinePushUserIDs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if len(offlinePushUserIDs) > 0 {
|
||||
needOfflinePushUserIDs = offlinePushUserIDs
|
||||
}
|
||||
|
||||
err = c.offlinePushMsg(ctx, msg, needOfflinePushUserIDs)
|
||||
if err != nil {
|
||||
log.ZWarn(ctx, "offlinePushMsg failed", err, "groupID", groupID, "msg", msg)
|
||||
return nil
|
||||
}
|
||||
|
||||
c.asyncOfflinePush(ctx, needOfflinePushUserIDs, msg)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *ConsumerHandler) asyncOfflinePush(ctx context.Context, needOfflinePushUserIDs []string, msg *sdkws.MsgData) {
|
||||
var offlinePushUserIDs []string
|
||||
err := c.webhookBeforeOfflinePush(ctx, &c.config.WebhooksConfig.BeforeOfflinePush, needOfflinePushUserIDs, msg, &offlinePushUserIDs)
|
||||
if err != nil {
|
||||
log.ZWarn(ctx, "webhookBeforeOfflinePush failed", err, "msg", msg)
|
||||
return
|
||||
}
|
||||
|
||||
if len(offlinePushUserIDs) > 0 {
|
||||
needOfflinePushUserIDs = offlinePushUserIDs
|
||||
}
|
||||
if err := c.pushDatabase.MsgToOfflinePushMQ(ctx, conversationutil.GenConversationUniqueKeyForSingle(msg.SendID, msg.RecvID), needOfflinePushUserIDs, msg); err != nil {
|
||||
log.ZError(ctx, "Msg To OfflinePush MQ error", err, "needOfflinePushUserIDs",
|
||||
needOfflinePushUserIDs, "msg", msg)
|
||||
prommetrics.SingleChatMsgProcessFailedCounter.Inc()
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
func (c *ConsumerHandler) groupMessagesHandler(ctx context.Context, groupID string, pushToUserIDs *[]string, msg *sdkws.MsgData) (err error) {
|
||||
if len(*pushToUserIDs) == 0 {
|
||||
*pushToUserIDs, err = c.groupLocalCache.GetGroupMemberIDs(ctx, groupID)
|
||||
@@ -300,7 +314,7 @@ func (c *ConsumerHandler) groupMessagesHandler(ctx context.Context, groupID stri
|
||||
if unmarshalNotificationElem(msg.Content, &tips) != nil {
|
||||
return err
|
||||
}
|
||||
log.ZInfo(ctx, "GroupDismissedNotificationInfo****", "groupID", groupID, "num", len(*pushToUserIDs), "list", pushToUserIDs)
|
||||
log.ZDebug(ctx, "GroupDismissedNotificationInfo****", "groupID", groupID, "num", len(*pushToUserIDs), "list", pushToUserIDs)
|
||||
if len(c.config.Share.IMAdminUserID) > 0 {
|
||||
ctx = mcontext.WithOpUserIDContext(ctx, c.config.Share.IMAdminUserID[0])
|
||||
}
|
||||
@@ -384,6 +398,7 @@ func (c *ConsumerHandler) getOfflinePushInfos(msg *sdkws.MsgData) (title, conten
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (c *ConsumerHandler) DeleteMemberAndSetConversationSeq(ctx context.Context, groupID string, userIDs []string) error {
|
||||
conversationID := msgprocessor.GetConversationIDBySessionType(constant.ReadGroupChatType, groupID)
|
||||
maxSeq, err := c.msgRpcClient.GetConversationMaxSeq(ctx, conversationID)
|
||||
@@ -392,6 +407,7 @@ func (c *ConsumerHandler) DeleteMemberAndSetConversationSeq(ctx context.Context,
|
||||
}
|
||||
return c.conversationRpcClient.SetConversationMaxSeq(ctx, userIDs, conversationID, maxSeq)
|
||||
}
|
||||
|
||||
func unmarshalNotificationElem(bytes []byte, t any) error {
|
||||
var notification sdkws.NotificationElem
|
||||
if err := json.Unmarshal(bytes, ¬ification); err != nil {
|
||||
|
||||
Reference in New Issue
Block a user