add PublishToLocal

[sljeff]

RichardKnop#686

[fenngwd]

@sljeff 是不是有几个consume写掉了

[fenngwd]

@sljeff PublishToLocal 从broker上暴露出来之后，怎么在liveness和readiness 探针里面去访问呢？

[sljeff]

@sljeff 是不是有几个consume写掉了

写漏了？应该只有 gcp_pubsub 的没办法实现就没写。

@sljeff PublishToLocal 从broker上暴露出来之后，怎么在liveness和readiness 探针里面去访问呢？

参考这个描述里面的 checkHealth 方法，在 goaby 或者什么地方实现一下就好 RichardKnop#686

[fenngwd]

@sljeff

machinery/v1/brokers/redis/redis.go

Line 283 in 2c04128

func (b *Broker) consume(deliveries <-chan []byte, concurrency int, taskProcessor iface.TaskProcessor) error {

[fenngwd]

参考这个描述里面的 checkHealth 方法，在 goaby 或者什么地方实现一下就好 RichardKnop#686

我的意思时说，比如你在Worker里实现了收到消息健康检查的Task，

	server.RegisterTask(healthCheckTaskName, func(healthCheckUUID string) error {
		select {
		case healthCheckCompleteChan <- healthCheckUUID: // success and send uuid
			return nil
		case <-time.After(5 * time.Second):
			return fmt.Errorf("send health check result error: %v", healthCheckUUID)
		}
	})

但从liveness和readiness的探针怎么调用了？HTTP？还是别的？

[sljeff]

@sljeff

machinery/v1/brokers/redis/redis.go

Line 283 in 2c04128

func (b *Broker) consume(deliveries <-chan []byte, concurrency int, taskProcessor iface.TaskProcessor) error {

这个逻辑是对的吧，consume 会从 deliveries 拉消息；PublishToLocal 也是往同一个 deliveries 里丢消息。

参考这个描述里面的 checkHealth 方法，在 goaby 或者什么地方实现一下就好 RichardKnop#686

我的意思时说，比如你在Worker里实现了收到消息健康检查的Task，

	server.RegisterTask(healthCheckTaskName, func(healthCheckUUID string) error {
		select {
		case healthCheckCompleteChan <- healthCheckUUID: // success and send uuid
			return nil
		case <-time.After(5 * time.Second):
			return fmt.Errorf("send health check result error: %v", healthCheckUUID)
		}
	})

但从liveness和readiness的探针怎么调用了？HTTP？还是别的？

探针就是有个命令能调用 checkHealth 方法就行，用 HTTP 应该没什么必要。

[fenngwd]

@sljeff 那换个问题，为什么amqp需要一个单独的delivery？

[fenngwd]

@sljeff 命令可以调用吗，不是local的吗？进程间怎么通信？

[sljeff]

@sljeff 那换个问题，为什么amqp需要一个单独的delivery？

应该是 amqp 的 deliveries 对象是个 <-chan Delivery（返回值是 amqp 那个库决定的），不能往里丢消息，所以就额外加了一个 channel。

@sljeff 命令可以调用吗，不是local的吗？进程间怎么通信？

确实，没考虑到这个。可能需要做成一个 http 接口或者什么其他的可以被触发的接口，总之是可以放进一个库里然后探针去触发。