Fix issue with message payload being disposed while resending

[RobertIndie]

Fixes #281

Motivation

When the producer resending the message, it would throw the following error:

[16:29:52.717 WRN] clientCnx(7, LogicalAddres Unspecified/apps-broker-0-fccfb098-3242-40e4-8d0c-2c15888ed404.usce1-whale.test.g.sn2.dev:6651) Socket was disconnected exceptionally on writing Send
System.ObjectDisposedException: Cannot access a disposed object.
Object name: 'The stream with Id ef744d42-7ea5-4652-9175-eddbd3bd3304 and Tag  is disposed.'.
   at Microsoft.IO.RecyclableMemoryStream.ThrowDisposedException() in /_/src/RecyclableMemoryStream.cs:line 1352
   at Microsoft.IO.RecyclableMemoryStream.Seek(Int64 offset, SeekOrigin loc) in /_/src/RecyclableMemoryStream.cs:line 1164
   at [email protected](PipeWriter output) in /Users/rbt/code/pulsar-client-dotnet/src/Pulsar.Client/Common/Commands.fs:line 84
   at <StartupCode$Pulsar-Client>[email protected]() in /Users/rbt/code/pulsar-client-dotnet/src/Pulsar.Client/Internal/ClientCnx.fs:line 304

The root cause is that the payload would be disposed regardless of whether the send is successful. And if there is some issue when the message is inflight, like a network issue after sending the message. The producer will resend the message. But now the message payload has already been disposed, so the clientCnx will fail to process the payload MemoryStream and throw an exception.

Modification

• Add the payload to the PendingMessage
• Avoid despoing the payload in the SendTask. And only depose the payload after deque the pendingMessage from the queue

Verification

I added a unit test for asserting the payload won't be deposed after serializeDeserializePayloadCommand. But it's hard to reproduce the reconnection in the unit test.

I have also verified that this PR could fix the issue: #281. I tested it by forcing the reconnecting and the producer could resend the messages successfully. Otherwise without this fix, it could sometimes get stuck resending messages due to the serialization error.

[Lanayx]

Thanks! I've released 3.6.1 with this fix