Merge TopicMessages by topic in sendBatch

[Nevon]

Fixes #622

I'm not entirely satisfied with the testing here. Basically what I'm asserting is that the expected number of messages are produced, but I'm not asserting message ordering.

Unfortunately I can't really validate that without consuming the topic, which means that the producer test depends on the consumer. Maybe that's okay? On the other hand, the logic is pretty trivial, but still.

[tulios]

surprisingly enough, the spread operation is faster than concat.

topicBatch.messages = [...topicBatch.messages, ...messages]

[JaapRood]

Totally bike shedding here, but I wonder whether that's because concatenation to any empty array is faster 🧐. Doesn't matter, move on, nothing to see here 😇.

[JaapRood]

Implementation looks fine to me. Test wise, I think it's probably worth consuming the messages we've produced in batch to verify their order has been preserved.

[JaapRood]

Totally bike shedding here, but I wonder whether that's because concatenation to any empty array is faster 🧐. Doesn't matter, move on, nothing to see here 😇.

[Nevon]

Implementation looks fine to me. Test wise, I think it's probably worth consuming the messages we've produced in batch to verify their order has been preserved.

I should learn to trust my gut, because of course we missed something and the implementation didn't actually work. I was using merged.find to find the TopicMessages for each topic, and was relying on mutating that to add the messages, but that didn't actually work:

const topicBatch = merged.find(({ topic: batchTopic }) => batchTopic === topic)

if (!topicBatch) {
  merged.push({ topic, messages })
} else {
  // this doesn't actually mutate `merged`
  topicBatch.messages.concat(messages)
}

return merged

Noticed when I updated the tests to actually consume the produced messages.

[JaapRood]

👍 LGTM. +1 for integration tests 😅