Send Message Batching

[devigned]

Related to: #35

After digging into send batching in Azure Event Hubs, I found the current implementation does not actually accomplish send batching as Event Hubs expects. Unfortunately, the structure that Event Hubs expects also can't be built easily via the publicly exposed structures.

Here's the Java code to implement this feature via Proton messages.

Basically, the first message is marshaled with an empty data body, then each data body is used to create an inner message, which is marshaled and added to a wrapper message. Finally, the wrapped messages are marshaled at the end of the first message.

I'll add a PR to better explain the process.

[devigned]

@vcabbage also related to this, when sending an improperly encode batch message, I'm seeing a corresponding disposition with an encoding error, but send does not return an error. Is this expected?

RX(Session): Disposition{Role: Receiver, First: 0, Last: <nil>, Settled: true, 
State: Rejected{Error: *Error{Condition: amqp:decode-error, Description: 
The format code '0' at frame buffer offset '136' is invalid or unexpected., 
Info: map[]}}, Batchable: false}

[vcabbage]

I'm confused. The code you linked to says:

// proton-j doesn't support multiple dataSections to be part of AmqpMessage
// here's the alternate approach provided by them: https://github.com/apache/qpid-proton/pull/54

The code in your PR shows using multiple data sections. Can you explain how they're related?

[vcabbage]

I'm not sure about the disposition error off the top of my head. I'll look at it in more detail after work.

[devigned]

The difference in multiple data sections in the PR is that the data sections are double encoded, first as a message with properties, and then as a data section / binary. This paired with message-format = 0x80013700 causes Event Hubs to handle the message as a batch and parse each of the encoded messages into individual messages in the hub.

When running the code as it is originally:

	for _, data := range m.Data {
		writeDescriptor(wr, typeCodeApplicationData)
		err := writeBinary(wr, data)
		if err != nil {
			return err
		}
	}

You would receive the following decode-error from Event Hub.

RX(Session): Disposition{Role: Receiver, First: 0, Last: <nil>, Settled: true, State: Rejected{Error: *Error{Condition: amqp:decode-error, Description: The format code '0x68' at frame buffer offset '141' is invalid or unexpected., Info: map[]}}, Batchable: false}

When doing the double encoding (in the PR), the messages are parsed properly.

As I said in the PR, I'm not sure this is how batching works across broker implementations and thus, not sure if it belongs in this library or somewhere more vendor specific.

[vcabbage]

Ok, if I'm understanding correctly, it seems that to support this use case we could export a function or method to encode an amqp.Message, then you can encoded and add the messages as data sections to the "container" message before sending. Does that sound right to you?

[devigned]

Yes, I think that is probably the best way to support this.

[vcabbage]

Regarding the rejected disposition, that is a bug. I've opened #48 to track it.