QueuePublisher/ConfirmedPublisherSink connecting to non-existent queue about op-rabbit HOT 6 CLOSED

I'll look into this this evening.

from op-rabbit.

1. That solution only works when publishing messages directly to the queue. I fear that having op-rabbit handle such a thing would explode the complexity. I think providing a way to specify that requirement is a better idea.

2. The ack is coming from RabbitMQ itself. See https://www.rabbitmq.com/confirms.html

We can publish the message in mandatory mode, but unfortunately, that causes the message to be routed back to us without the delivery tag, and unless if set the delivery tag in a custom header for every outbound message, I don't see any way to NACK a message that did not get delivered. It does highlight a bigger issue, however. Perhaps the best way to do it would be to have the application allocate a confirmedPublisher per sink, have it publish in mandatory mode, and then have the whole sink fail if a message is unrouteable.

from op-rabbit.

I've been thinking about this... I just switched the publisher strategy from a function that receives a channel, payload, etc. and returns unit. Since a publisher strategy can be bound to a specific queue, you could have it verify the existence of the queue to which it's publishing, and then fail if it doesn't exist.

class VerifyingQueuePublisher(queue: String) extends MessagePublisher {
  private var queueVerified = false
  def apply(c: Channel, data: Array[Byte], properties: BasicProperties): Unit = {
    if (!queueVerified) {
      c.queueDeclarePassive(queueName)
      queueVerified = true
    }
    c.basicPublish("", queue, properties, data)
  }
}

// returns failed Future if target queue does not exist:
rabbitControl ? ConfirmedMessage(VerifyingQueuePublisher(queueName), data)

What do you think of this? Would this solve your problem?

from op-rabbit.

Yeah, that looks pretty good. I was going to suggest adding another function to MessagePublisher that gets called when the stream is materialized, but this might be better as it keeps the interface simple. The only drawback I can see, is that in order to verify the queue exists, it means making the publisher mutable.

About the confirmations, I have no idea how I didn't notice it before, but that's really bad. What's the point of sending an ack if the message wasn't persisted (or, at least, not where the sender intended it to be)?
Anyway, I'm not so sure about the mandatory mode, it seems like a lot of overhead without necessarily solving the problem.

from op-rabbit.

Okay, great. Thanks for the feedback. I'll work towards making this a reality (what's lacking is proper error handling in the publisher strategy, and channeling that exception back from the ConfirmedPublisher) And... it's not mutable... it's "memoized / lazy"! :D

I'm also going to work on simplifying the definition of implementing custom consumer patterns, providing "escape" hatches where you can leave the opinionated / safety of op-rabbit when needed.

from op-rabbit.

Supported; 6b743ba

from op-rabbit.