NotifyPublish ReconnectCount - how is it intended to be used?

[jamoflaw]

Similar to #66 I am trying to enable confirms using the NotifyPublish handler - attempting to manually keep track of the DeliveryTag being set by incrementing each time I successfully publish

However on reconnect we don't seem to have any mechanism for the publisher to know that a reconnect happens so this manual tracking of DeliveryTags falls over when the channels DeliveryTag silently resets to 0

For example:

	// Publisher func
	go func() {

		publisher.NotifyPublish(func(p rabbitmq.Confirmation) {
			publisher_confirms <- p
		})

		// Initialise a delivery counter to keep track of the DeliveryTags assigned to the messages
		deliveryCounter := messagestore.DeliveryTag(1)
		for msg := range messages {

			err := publisher.PublishWithContext(
				context.Background(),
				msg.Body,
				[]string{msg.RoutingKey},

				rabbitmq.WithPublishOptionsExchange("ingress"),
				rabbitmq.WithPublishOptionsContentType(msg.ContentType),
				rabbitmq.WithPublishOptionsContentEncoding(msg.ContentEncoding),
				rabbitmq.WithPublishOptionsAppID(msg.AppId),
				rabbitmq.WithPublishOptionsPersistentDelivery,
			)
			if err != nil {
				logger.Errorf("error publishing message, %s", err)
				continue
			}

			store.AppendMessage(deliveryCounter, msg)
			deliveryCounter++

		}
	}()

	// Confirms func
	go func() {

		for {
			select {
			case confirm := <-publisher_confirms:
				// Try to confirm
				if confirm.Ack {
					/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
					// Whatever I do here  you can't work out the delivery counter since the reconnection count isn't related to what the delivery counter was _before_
					// it was reset
					/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
					store.Ack(messagestore.DeliveryTag(confirm.DeliveryTag + uint64(confirm.ReconnectionCount)))
				} else {
					store.Nack(messagestore.DeliveryTag(confirm.DeliveryTag+uint64(confirm.ReconnectionCount)), true)
				}
			}
		}
	}()

Am I missing something in how it is supposed to be used? Is there any internal delivery tag which could be returned from the publisher object instead to track it?

[jamoflaw]

A potentially naïve fix (assuming I'm not missing how its supposed to be used) would be to expose the GetNextPublishSeqNo() method from the underlying channel

in channel_manager.go

func (chanManager *ChannelManager) GetNextPublishSeqNo() uint64 {
	return chanManager.channel.GetNextPublishSeqNo()
}

in publish.go

func (publisher *Publisher) GetNextPublishSeqNo() uint64 {
	publisher.disablePublishDueToBlockedMux.RLock()
	defer publisher.disablePublishDueToBlockedMux.RUnlock()
	return publisher.chanManager.GetNextPublishSeqNo()
}

however the thread safety of the lib is something I'm not 100% certain about - grabbing the next delivery tag sequence + publishing in two steps opens the possibility of a race condition if the publisher were to be shared between two go routines (is that even something which is possible / supported?)

I've tested this on a simple publisher / consumer app and that now works with a app where the producer is contained within a single go routine (showing changes from the original code)

	// Publisher func
	go func() {

		publisher.NotifyPublish(func(p rabbitmq.Confirmation) {
			publisher_confirms <- p
		})
		publisher.NotifyReturn(func(r rabbitmq.Return) {
			publisher_returns <- &r
		})

		// Initialise a delivery counter to keep track of the DeliveryTags assigned to the messages
		for msg := range messages {

			deliveryTag := messagestore.DeliveryTag(publisher.GetNextPublishSeqNo())
			err := publisher.PublishWithContext(
				context.Background(),
				msg.Body,
				[]string{msg.RoutingKey},

				rabbitmq.WithPublishOptionsExchange("ingress"),
				rabbitmq.WithPublishOptionsContentType(msg.ContentType),
				rabbitmq.WithPublishOptionsContentEncoding(msg.ContentEncoding),
				rabbitmq.WithPublishOptionsAppID(msg.AppId),
				rabbitmq.WithPublishOptionsPersistentDelivery,
			)
			if err != nil {
				logger.Errorf("error publishing message, %s", err)
				continue
			}

			store.AppendMessage(deliveryTag, msg)
		}
	}()

	// Confirms func
	go func() {

		for {
			select {
			case confirm := <-publisher_confirms:
				// Try to confirm
				if confirm.Ack {
					logger.Infof("Attempting to ack DeliveryTag: %d and ReconnectionCount: %d", confirm.DeliveryTag, confirm.ReconnectionCount)
					store.Ack(messagestore.DeliveryTag(confirm.DeliveryTag))
				} else {
					logger.Infof("Attempting to nack DeliveryTag: %d and ReconnectionCount: %d", confirm.DeliveryTag, confirm.ReconnectionCount)
					store.Nack(messagestore.DeliveryTag(confirm.DeliveryTag), true)
				}

			case <-publisher_returns:

			}
		}
	}()

[wagslane]

The reconnect count should let you know when a reconnect happens, does that not help you know when the counter is reset?