Add send overload with option to explicitly set batchable flag

[Havret]

In nms amqp we need option to explicitly set Delivery batchable flag. This PR adds Send overload with this option.

The reasoning behind this change:

apache/activemq-nms-amqp#8 (comment)

MessageProducer rate are 1 msg/sec on brokers that respect the batchable flag for amqp transfers. This has been a long standing issue with amqpnetlite until it was fixed for synchronous sends, version 2.1.8. The provider AmqpProducer uses async sends with ManualResetEvent to block. This is bad in term of amqp protocol as the transfer sent as batchable indicating to the broker that disposition should not be sent right away so it waits however the client is also waiting for a response leading to terrible message rates. Amqpnetlite fixed this issue at least for synchronous sends from a sender link where the batchable flag is set to false when a send is synchronous however that means that AmqpProducer must use a different send (see senderLink.send(msg) and senderLink.send(msg, timeout)) method then its currently using.

@xinchen10 Any chance that you could take a look at this?

[michaelandrepearce]

@xinchen10 any chance this could be merged and release, we're a bit blocked without it.

[xinchen10]

In general I am trying to avoid adding overloads like this to control a specific behavior of the protocol. I understand they are important to some scenarios, and I think we need a better way for it. The handler model in Proton clients is a nice way to observe or change the protocol behavior. We could have something similar registered to a connection and the library will invoke the handler for important events. Changing the value of batchable or even the delivery tag would be done in the OnDelivery event.

[michaelandrepearce]

At the connection level i dont think will work. It is needed to be at the send level as some events cant be batched and these will interleave. @Havret can you share the issue/code so it gives @xinchen10 more context.

[Havret]

To support transactions apache/activemq-nms-amqp#8 we need to pass Transactional State to send method. Unfortunately, the available overload is constructed in such a way, that to support both async (fire and forget) and sync sends we needed to rewrite current implementation of amqpnetlite sync send method and pass our transactional state. As a result we have no more the option to set batchable flag, is it is being set internally by default Send overload:

amqpnetlite/src/SenderLink.cs

Lines 153 to 156 in abf0504

	public void Send(Message message, DeliveryState deliveryState, OutcomeCallback callback, object state)
	{
	this.SendInternal(message, deliveryState, callback, state, false);
	}

Our current implementation is as following:

https://github.com/HavretGC/activemq-nms-amqp/blob/c535a2e4b2e63e1b8741305d6780f6f201c20223/src/NMS.AMQP/Provider/Amqp/AmqpProducer.cs#L144-L178

[cjwmorgan-sol]

In general I am trying to avoid adding overloads like this to control a specific behavior of the protocol. I understand they are important to some scenarios, and I think we need a better way for it. The handler model in Proton clients is a nice way to observe or change the protocol behavior. We could have something similar registered to a connection and the library will invoke the handler for important events. Changing the value of batchable or even the delivery tag would be done in the OnDelivery event.

@xinchen10
How would a potential OnDelivery event be defined. Would such an event include context from the senderLink.Send method? Like the "state" parameter? Deliveries already have a userToken for the outcomeCallbacks would re-purposing this state parameter cause issues? If so then a new argument must be passed anyways to be referred by the new event, still introducing a new overloaded send method. Also what would be the interface for modifying the Delivery?

The Message Class already has some access for Delivery fields, might it be easier to just add a boolean property to set the batchable flag on the Message avoiding adding a Event to modify what is currently internal fields to amqpnetlite? This would hopefully avoid having to add a new send overload and provide a per message send batchable flag assignment.

Being able to assign the batchable flag per message would greatly improve a long standing performance issue (now isolated to amqp transactions) with amqp brokers that support the batchable flag in the amqp protocol.

At the connection level i dont think will work. It is needed to be at the send level as some events cant be batched and these will interleave. @Havret can you share the issue/code so it gives @xinchen10 more context.

As long as the "state" object from the SenderLink.Send method is past (or some equivalent context reference) to the potential OnDelivery event to provide per message send specific context to apply changes to the outgoing Delivery having a Connection (or maybe Session?) event.

[xinchen10]

Something similar to https://github.com/apache/qpid-proton-j/blob/master/proton-j/src/main/java/org/apache/qpid/proton/engine/Handler.java but much simpler. It won't have nested handlers nor event dispatching. Just an method the library calls on certain important events. It needs to be extensible so we can start with a few and add more if needed.

[michaelandrepearce]

I think the handler approach will be difficult as it will be too late, meaning the sender will be needing to track per message a handler (e.g. a handler per send) this wouldnt be that great. Also that would be some breaking change if i understand in the current way AMQPNetlite is working.

Maybe a halfway house between the two extremes would be as @cjwmorgan-sol suggested it be a message level access to delivery fields, that could be honoured? Would that be exceptable @xinchen10 , this actually could be quite powerful to have message level delivery fields, as could also support link re-attachment this way also, something that is missing currently as well.

[xinchen10]

I added the Handler. On the SendDelivery event you can inspect or change any field of IDelivery. Depending on your code layout, I am not sure how easy it is to set up the handler on the connection.

[Havret]

@xinchen10 Any ETA on new release with the Handler included?

[xinchen10]

Have you tried it? If it works fine I can release it soon.

[cjwmorgan-sol]

Have you tried it? If it works fine I can release it soon.

I've tried to use a build using the master branch and it produces invalid amqp on the target on the receiver link with 25 NULL amqp bytes. Not sure if that's related to your IHandler changes though.

@Havret I used a quickly hacked up implementation of the new IHandler with the Helloworld sample in nms amqp have you tried the changes that are on amqpnetlite master yet?

[cjwmorgan-sol]

Found my issue the Method DescribedList.EncodeValue counts the number of non-null fields of a DescribedList then writes the number of nulls around as padding. However the the calculation for counting the number of field does not handle the case when all fields are null. Like when a receiverLink gives a new Target() instance, in the amqp nms provider. Having all field be null will cause the "count" var to become the max field count of 32 and on a DescribedList of Fixed length 7 aka Target this is invalid amqp. Although I found apache artemis seemed to handle this with some warning logs. And some other amqp brokers do not.

Here is the code snippet with a suggested fix from DescribedList.cs (about line 402)

internal override void EncodeValue(ByteBuffer buffer)
{
int count = 0;
int temp = this.fields;
while (temp > 0)
{
count++;
temp <<= 1;
}
count = (count == 0) ? 0 : 32 - count; // fix
count = 32 - count; // original
if (count == 0)
{
AmqpBitConverter.WriteUByte(buffer, FormatCode.List0);
}
else
{
int pos = buffer.WritePos;
AmqpBitConverter.WriteUByte(buffer, 0);
....

Without the amqp protocol issue I found the IHandler interface could set batchable per message.

[xinchen10]

Nice find. It should be fixed now.

[Havret]

@xinchen10 I used it and it seems to work (apache/activemq-nms-amqp#20), but in my humble opinion, it would make better sense to have handler exposed on connection factory CreateAsync method (passed as an optional argument), rather than on ConnectionFactory itself. Current implementation results in reusing handler instance between all connections created by a single factory, which for me is kind of a counter-intuitive.

The other thing is that while using Handler API we do not have the context of the invocation any more. Imagine I would like to extend IDelivery interface Havret@7c646a0 to be able to tweak settled field conditionally. With current api it would be really hard to do achieve this. I would have to really work my way around, to pass the context, to be able to react in handler accordingly.

As we are taking design ideas from Java Proton implementation, maybe it would be worth considering to apply their approach to delivery object? In Proton we have full control over the delivery object, and we can tune it as much as we want before sending the frame over the wire.

[xinchen10]

The handler was set on the settings so that Listener can use it as well, but I agree that for it makes more sense to have it in CreateAsync call for client factory. I have refactored the code.

For IDelivery, what else do you need for this scenario? If you delay the decision on setting delivery properties, you would have to pass the context from the send call. Though now you could do everything in the handler callback, it is still better to set most of the properties, if possible, when you send.

[Havret]

The thing is I cannot set most of the properties (in this particular case the Settled field of the Delivery) when I send, because Delivery object is internal. On the other hand, when I use handler callback I don't know what I want to set, because I don't have context of the invocation any more.

[xinchen10]

Are you calling the overload that takes a user's state object? Can you use that to make decision later? It seems that you are looking for a method like void Send(Delivery delivery). It looks very powerful but I think it can easily cause problems if it is set correctly.

[Havret]

Hmm. I think you just provided me what I needed. User state can be used to preserve the context and to extract it later in the handler. I totally missed that one. It should do the trick. Nice!

Could you please merge this #380 one?

[Havret]

@xinchen10 Any chance that we have a release any time soon?