mirror of
https://github.com/openimsdk/open-im-server.git
synced 2025-04-06 04:15:46 +08:00
9ed6200e45
* pb * fix: Modifying other fields while setting IsPrivateChat does not take effect * fix: quote message error revoke * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * refactoring scheduled tasks * upgrading pkg tools * fix * fix * optimize log output * feat: support GetLastMessage * feat: support GetLastMessage * feat: s3 switch * feat: s3 switch * fix: GetUsersOnline * feat: SendBusinessNotification supported configuration parameters * feat: SendBusinessNotification supported configuration parameters * feat: SendBusinessNotification supported configuration parameters * feat: seq conversion failed without exiting * monolithic * fix: DeleteDoc crash * fix: DeleteDoc crash * fix: monolithic * fix: monolithic * fix: fill send time * fix: fill send time * fix: crash caused by withdrawing messages from users who have left the group * fix: mq * fix: mq * fix: user msg timestamp * fix: mq * 1 * 1 * 1 * 1 * 1 * 1 * 1 * seq read config * seq read config * 1 * 1 * fix: the source message of the reference is withdrawn, and the referenced message is deleted * 1 * 1 * 1 * 1 * 1 * 1 * 1 * 1 * 1 * 1 * 1 * 1 * 1 * 1
61 lines
1.7 KiB
Go
61 lines
1.7 KiB
Go
package mqbuild
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
|
|
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
|
"github.com/openimsdk/tools/mq"
|
|
"github.com/openimsdk/tools/mq/kafka"
|
|
"github.com/openimsdk/tools/mq/simmq"
|
|
)
|
|
|
|
type Builder interface {
|
|
GetTopicProducer(ctx context.Context, topic string) (mq.Producer, error)
|
|
GetTopicConsumer(ctx context.Context, topic string) (mq.Consumer, error)
|
|
}
|
|
|
|
func NewBuilder(kafka *config.Kafka) Builder {
|
|
if config.Standalone() {
|
|
return standaloneBuilder{}
|
|
}
|
|
return &kafkaBuilder{
|
|
addr: kafka.Address,
|
|
config: kafka.Build(),
|
|
topicGroupID: map[string]string{
|
|
kafka.ToRedisTopic: kafka.ToRedisGroupID,
|
|
kafka.ToMongoTopic: kafka.ToMongoGroupID,
|
|
kafka.ToPushTopic: kafka.ToPushGroupID,
|
|
kafka.ToOfflinePushTopic: kafka.ToOfflineGroupID,
|
|
},
|
|
}
|
|
}
|
|
|
|
type standaloneBuilder struct{}
|
|
|
|
func (standaloneBuilder) GetTopicProducer(ctx context.Context, topic string) (mq.Producer, error) {
|
|
return simmq.GetTopicProducer(topic), nil
|
|
}
|
|
|
|
func (standaloneBuilder) GetTopicConsumer(ctx context.Context, topic string) (mq.Consumer, error) {
|
|
return simmq.GetTopicConsumer(topic), nil
|
|
}
|
|
|
|
type kafkaBuilder struct {
|
|
addr []string
|
|
config *kafka.Config
|
|
topicGroupID map[string]string
|
|
}
|
|
|
|
func (x *kafkaBuilder) GetTopicProducer(ctx context.Context, topic string) (mq.Producer, error) {
|
|
return kafka.NewKafkaProducerV2(x.config, x.addr, topic)
|
|
}
|
|
|
|
func (x *kafkaBuilder) GetTopicConsumer(ctx context.Context, topic string) (mq.Consumer, error) {
|
|
groupID, ok := x.topicGroupID[topic]
|
|
if !ok {
|
|
return nil, fmt.Errorf("topic %s groupID not found", topic)
|
|
}
|
|
return kafka.NewMConsumerGroupV2(ctx, x.config, groupID, []string{topic}, true)
|
|
}
|