使用go mod tidy时，默认的github.com/tx7d0/kratos-transport/transport/kafka版本是v1.0.2
此版本kafka.WithCodec()接受参数类型为encoding.Codec，而不是string，而kratos-cqrs的范例代码中传入string，会报错。

手动在go.mod里指定版本为v0.0.0-20221220110912-9db1b79d385c可以解决这个问题。请问能否让go mod tidy自动指定的版本为正确的版本？

没有实现 add 方法。

# github.com/tx7do/kratos-transport/transport/pulsar
/Users/caster/go/pkg/mod/github.com/tx7do/kratos-transport/transport/[email protected]/transport.go:40:9: cannot use tr.reqHeader (variable of type headerCarrier) as transport.Header value in return statement: headerCarrier does not implement transport.Header (missing method Add)
/Users/caster/go/pkg/mod/github.com/tx7do/kratos-transport/transport/[email protected]/transport.go:45:9: cannot use tr.replyHeader (variable of type headerCarrier) as transport.Header value in return statement: headerCarrier does not implement transport.Header (missing method Add)

相关 issue：#45

是否支持nats jetstream？

尝试使用 默认Codec nil
然后 注册订阅时 binder 返回也为 nil

上图中outValue 不会为nil 导致无法获取
还是我得姿势不对
求解

使用 go get -u github.com/tx7do/kratos-transport/broker/rabbitmq命令，无法更新 rabbitmq 这个子包的版本，代码编译时会报错。

请问一下大佬，加了 pulsar server 之后，consul 的 health check 会失败。是这个用法不对吗？

你好，
这个库好像无法指定一些rabbitmq的高级参数，比如交换机的名字，优先级等。 在代码中我看到了相关的封装，但是外层无法把这个参数注入进去。这个要这么解决呢。

transport/asynq/server.go
Server.init
When using custom ServerOption like redis address, it will not apply to asynq.Server
Because in init() function, it will crate a kratos.Server instance and asynq.Server first, then it will apply ServerOption to kratos.Server, but won't apply to asynq.Server. It will always using default redis address '127.0.0.1:6379'

func NewServer(opts ...ServerOption) *Server {
	srv := &Server{
		baseCtx: context.Background(),
		started: false,
		redisOpt: asynq.RedisClientOpt{
			Addr: defaultRedisAddress,
			DB:   0,
		},
		asynqConfig: asynq.Config{
			Concurrency: 10,
			Logger:      newLogger(),
		},
		mux: asynq.NewServeMux(),
	}

	srv.init(opts...)

	return srv
}

func (s *Server) init(opts ...ServerOption) {
	_ = s.createAsynqServer()

	for _, o := range opts {
		o(s)
	}
}

func (s *Server) createAsynqServer() error {
	if s.asynqServer != nil {
		log.Errorf("[asynq] asynq server already created")
		return errors.New("asynq server already created")
	}

	s.asynqServer = asynq.NewServer(s.redisOpt, s.asynqConfig)
	if s.asynqServer == nil {
		log.Errorf("[asynq] create asynq server failed")
		return errors.New("create asynq server failed")
	}
	return nil
}

能加一个kitex的transport吗，来替换kratos的grpc

最新修改的代码中
https://github.com/tx7do/kratos-transport/blob/5cbe362109/broker/rabbitmq/connection.go#L251
新增了 autoDel 参数，但是调用端没有使用 这里 7 个参数

func (r *rabbitConnection) Consume(queueName, routingKey string, bindArgs amqp.Table, qArgs amqp.Table, autoAck, durableQueue, autoDel bool) (*rabbitChannel, <-chan amqp.Delivery, error) {
	consumerChannel, err := newRabbitChannel(r.Connection, r.qos)
	if err != nil {
		return nil, nil, err
	}

	if err = consumerChannel.DeclareQueue(queueName, qArgs, durableQueue, autoDel); err != nil {
		return nil, nil, err
	}

	deliveries, err := consumerChannel.ConsumeQueue(queueName, autoAck)
	if err != nil {
		return nil, nil, err
	}

	if err = consumerChannel.BindQueue(queueName, routingKey, r.exchange.Name, bindArgs); err != nil {
		return nil, nil, err
	}

	return consumerChannel, deliveries, nil
}

调用端
参数不足，直接报错，只传递 6 个参数
https://github.com/tx7do/kratos-transport/blob/5cbe362109/broker/rabbitmq/subscriber.go#L75

		ch, sub, err := s.r.conn.Consume(
			s.opts.Queue,
			s.topic,
			s.headers,
			s.queueArgs,
			s.opts.AutoAck,
			s.durableQueue,
		)

是这样写的不?

srv := kafka.NewServer(
		kafka.WithAddress(c.Kafka.Addrs),
		kafka.WithCodec(encoding.GetCodec("json")),
		kafka.WithTracerProvider(tp, "kafka-tracer"),
	)

Hi，
我在kratos中，使用kafka的server，需要配置一下Borker的readerConfig的QueueCapacity。现在支持吗？

server: docker
redis version: 6.2.7
err msg: === RUN TestServer
ReadTimeout set error
IdleTimeout set error
ReadTimeout set error
IdleTimeout set error
INFO msg=[redis] server listening on: redis://127.0.0.1:6379
redis recv error: read tcp 127.0.0.1:49996->127.0.0.1:6379: i/o timeout

AsynqServer 首次添加周期性任务时请求会被卡住，看源码应该是在创建并等待 Scheduler 启动完成。

调用代码如下

首次调用

再次调用

问题:

当我循环注册多个queue消费时出现无法消费以及控制面板上的queue未创建的问题

以下是我需要注册的queue

实际的面板queue

使用版本

github.com/tx7do/kratos-transport v1.0.5 // indirect
github.com/tx7do/kratos-transport/broker/rabbitmq v0.0.0-20221220110912-9db1b79d385c // indirect
github.com/tx7do/kratos-transport/transport/rabbitmq v0.0.0-20221220110912-9db1b79d385c // indirect

以下是实际的代码

func NewMqServer(
	c *conf.Data,
	logger log.Logger,
	bk broker.Broker,
	executors *sender.TaskExecutor,
	hs *sender.Handle,
) *rabbitmq.Server {


	srv := rabbitmq.NewServer(
		rabbitmq.WithAddress([]string{c.Rabbitmq.URL}),
		rabbitmq.WithExchange("austin.biz.exchange", true),
		rabbitmq.WithCodec("json"),
	)
	logic := NewMqHandler(logger, bk, executors, hs)

	for _, groupId := range groups.GetAllGroupIds() {
		fmt.Println(`subscriber`,fmt.Sprintf("austin.biz.%s", groupId))
		_ = srv.RegisterSubscriber(context.Background(),
			"austin.biz.routing",
			logic.registerMessageHandler(logic.onMassage),
			func() broker.Any {
				return []*types.TaskInfo{}
			},
			broker.WithQueueName(fmt.Sprintf("austin.biz.%s", groupId)),
			brokermq.WithDurableQueue(),
			brokermq.WithAckOnSuccess(),
		)
	}

	return srv
}



type MessageHandler func(_ context.Context, topic string, headers broker.Headers, msg []*types.TaskInfo) error

type MqHandler struct {
	logger   log.Logger
	broker   broker.Broker
	executor *sender.TaskExecutor
	hs       *sender.Handle
}

func NewMqHandler(
	logger log.Logger,
	broker broker.Broker,
	executor *sender.TaskExecutor,
	hs *sender.Handle,
) *MqHandler {
	return &MqHandler{
		logger:   logger,
		broker:   broker,
		executor: executor,
		hs:       hs,
	}
}

func (m *MqHandler) onMassage(ctx context.Context, topic string, headers broker.Headers, taskList []*types.TaskInfo) error {
	for _, taskInfo := range taskList {
		fmt.Println(taskInfo)
		channel := channelType.TypeCodeEn[taskInfo.SendChannel]
		msgType := messageType.TypeCodeEn[taskInfo.MsgType]
		err := m.executor.Submit(ctx, fmt.Sprintf("%s.%s", channel, msgType), sender.NewTask(taskInfo, m.hs, m.logger))
		if err != nil {
			l := log.NewHelper(log.With(m.logger, "module", "MqHandler/onMassage"))
			l.Error("err", err, "task_list", taskList)
		}
	}
	return nil
}

func (m *MqHandler) registerMessageHandler(fnc MessageHandler) broker.Handler {
	return func(ctx context.Context, event broker.Event) error {
		switch t := event.Message().Body.(type) {
		case []*types.TaskInfo:
			if err := fnc(ctx, event.Topic(), event.Message().Headers, t); err != nil {
				return err
			}
		default:
			return fmt.Errorf("unsupported type: %T", t)
		}
		return nil
	}
}

在最新的代码提交中，ackSuccess 是取反 ackSuccessKey 的值
https://github.com/tx7do/kratos-transport/blob/main/broker/rabbitmq/rabbitmq.go#L197

	var ackSuccess = false
	if val, ok := options.Context.Value(ackSuccessKey{}).(bool); ok {
		options.AutoAck = val
		ackSuccess = !val
	}

但是，声明服务中的选项永远都是 true，导致没有错误时，手动 ack 无效
https://github.com/tx7do/kratos-transport/blob/main/broker/rabbitmq/options.go#L88

func WithAckOnSuccess() broker.SubscribeOption {
	return broker.SubscribeContextWithValue(ackSuccessKey{}, true)
}

下面 第一个 if 不会生效
https://github.com/tx7do/kratos-transport/blob/main/broker/rabbitmq/rabbitmq.go#L221

		p.err = handler(r.opts.Context, p)
		if p.err == nil && ackSuccess && !options.AutoAck {
			_ = msg.Ack(false)
		} else if p.err != nil && !options.AutoAck {
			_ = msg.Nack(false, requeueOnError)
		}

问题描述： MQTT断线再次重连后，之前订阅过的主题，无法收到消息
期望：MQTT断线后重连，之前订阅的消息能够正常收到消息
代码如下，

func NewMQTTServer(
	c *conf.Bootstrap, svc *service.OperationStatService,
) *mqtt.Server {
	ctx := context.Background()
	var opts = []mqtt.ServerOption{
		mqtt.WithAddress([]string{c.Server.Mqtt.Addr}),
		mqtt.WithCodec("json"),
		mqtt.WithAuth(c.Server.Mqtt.Username, c.Server.Mqtt.Password),
		mqtt.WithClientId(c.Server.Mqtt.ClientId),
		mqtt.WithCleanSession(true),
	}
	srv := mqtt.NewServer(opts...)
	_ = srv.RegisterSubscriber(
		ctx,
		c.OperationStat.SubTopic,
		biz.RegisterDeviceAckJsonHandler(svc.DeviceAckHandler),
		biz.DeviceAckCreator,
	)
	svc.SetMqttBroker(srv)
	return srv
}

rabbitmq或者 rocketmq 能支持日志吗

请问 rabbitmq 可以支持 opentelemetry 链路追踪吗？？

如题

作者你好，我通过go get github.com/tx7do/kratos-transport/transport/websocket@latest安装依赖包，版本号是日期哈希值
github.com/tx7do/kratos-transport/transport/websocket v0.0.0-20230803222306-0904a461c75e
在日志记录caller时非常长
INFO ts=2023-08-14T10:28:18+08:00 [email protected]/server.go:267 service.id=xxx service.name= service.version= trace.id= span.id= msg=[websocket] server listening on: [::]:9002
能否优化加个tag。
ps：对go包发布不是特别了解，不确定能不能实现子模块的版本管理。

有考虑后续给nats加入 jetstream的计划么？

目前sdk仅支持RocketMQ4.x, 还没法支持RocketMQ5.x

go get github.com/tx7do/kratos-transport/codec/proto
go: module github.com/golang/protobuf is deprecated: Use the "google.golang.org/protobuf" module instead.

按照Kraots示例跑了下，出现这个问题，一般是什么原因导致

看到最新的提交，部分mq已经支持 链路追踪 ，但是实测无效

应该是 content 脱离，导致不一致，我试了下，手动放入 options.Context 是可以记录的

span := b.startConsumerSpan(options.Context, options.Queue, &msg)
defer span.End()

最新提交的代码中，声明队列时，autoDelete = true ，导致就算队列设置 持久化 就没有意义啦，因为如果没有消费者后，队列会自动删除，导致未消费的消息丢失，请问这样的设计原意是什么？

https://github.com/tx7do/kratos-transport/blob/main/broker/rabbitmq/channel.go#L69

	_, err := r.channel.QueueDeclare(
		queueName,
		durable,
		true,
		false,
		false,
		args,
	)

由于 rabbitmq 创建消费者服务时使用 routingKey 进行分组，并且使用 topic 模式， 如果一个 routingKey 下绑定多个队列，应该如何区分？？
示例：

Hi，
我现在使用Kratos框架，server使用的是 kratos-transport的kafka。
有如下疑问：

1. 优雅关闭消费者时，kratos-transport是通过 context机制来实现的（server.Stop()函数不生效）。subscribe的context是通过kafkaSrv.RegisterSubscriber()来设置的，在与Kratos结合时，subscribe的context应该与Kratos.App.ctx关联起来，在Kratos的layout中，Server是单独一个目录 internal/server/xxx.go，关联起来不是很方便，有什么好的方法吗?
2. subscribe 使用 context来关闭，会出现一个问题。如果一个消息正在消费（处理业务逻辑），此时收到了优雅关闭信号，context被取消，此时可能收到错误 ERROR msg=[kafka]: unable to commit msg: context canceled， 待服务重启后，会出现二次消费的问题。

kafka是否可以支持像NSQ那样WithDeferredPublish的配置，实现延迟队列呢？

Related issue：rabbitmq/amqp091-go#18

客户端如何投递任务呢？大佬出个实践示例可否？

看了下broker/kafka中的代码，调用实际上里面封装了segmentio/kafka-go这个包的方法，但是在WriteMessages方法中并没有根据传来字段的hash值指定分区，是现在的版本还不支持吗？

目前声明或绑定交换机与队列都只能在消费端，生产端没有选项，没法实现 ttl 延时队列的功能，消费端绑定就会消费掉数据，希望可以添加选项，生产端可以选择绑定，实现死信与延迟消息功能

pulsar.NewServer 地址不加 pulsar:// 前缀时，解析错误。

srv := pulsar.NewServer(
    pulsar.WithAddress([]string{"0.0.0.0:6650"}),
)

error="parse \"0.0.0.0:6650\": first path segment in URL cannot contain colon"

kafka server adds 不支持多个地址，borker是支持多个地址的

file: transport/asynq/server.go
Sever.runAsynqServer
if asynqServer is not nil which is actually normal in this case, but it will return error

func (s *Server) runAsynqServer() error {
	if s.asynqServer != nil {
		log.Errorf("[asynq] asynq server is nil")
		return errors.New("asynq server is nil")
	}

	if err := s.asynqServer.Run(s.mux); err != nil {
		log.Errorf("[asynq] asynq server run failed: %s", err.Error())
		return err
	}
	return nil
}

C:\Users\xxx\go\pkg\mod\github.com\tx7do\kratos-transport\transport\[email protected]\transport.go:41:9: cannot use tr.reqHeader (variable o
f type headerCarrier) as transport.Header value in return statement: headerCarrier does not implement transport.Header (missing method Add)
C:\Users\xxx\go\pkg\mod\github.com\tx7do\kratos-transport\transport\[email protected]\transport.go:46:9: cannot use tr.replyHeader (variable
of type headerCarrier) as transport.Header value in return statement: headerCarrier does not implement transport.Header (missing method Add)

github.com/go-kratos/kratos/v2 v2.6.2
github.com/tx7do/kratos-transport v1.0.5

kratos-transport 用kafka 有个报错
cannot use tr.reqHeader (variable of type headerCarrier) as transport.Header value in return statement: headerCarrier does not implement transport.Header (missing method Add)

希望可以自定义 SubscriptionName 和 SubscriptionInitialPosition 等配置。
请问现在有办法实现吗？

if credentials.AccessKey != "" && credentials.SecretKey != "" {
producer.WithCredentials(*credentials)
} 为啥不用加opts = append(opts, producer.WithCredentials(*credentials))

mqtt连接断线后，broker自动重新建立连接，但是已经订阅的主题没有重新订阅。