For Josh: AMQP Serialization Testing

amqp-serialization-spike.cs

        private static BufferListStream Serialize(AmqpMessage message, int bufferSizeBytes = 4096)
        {
            var buffers = new List<ArraySegment<byte>>();
            var more = true;

            while (more)
            {
                var messageBuffers = message.GetPayload(bufferSizeBytes, out more);

                if (messageBuffers != null)
                {
                    foreach (var segment in messageBuffers)
                    {
                        message.CompletePayload(segment.Count);
                        buffers.Add(segment);
                    }
                }
            }

            return new BufferListStream(buffers);
        }


        [Test]
        public void JESSE_TEST_SERIALIZATION_SINGLE()
        {
            var converter = new AmqpMessageConverter();

            var data = EventHubsModelFactory.EventData(
                new BinaryData("this is a body"),
                new Dictionary<string, object> { { "appKey", "appValue" }},
                null,
                "part-key",
                9989,
                1121,
                new DateTimeOffset(2015, 10, 27, 0, 0, 0, TimeSpan.Zero));

            data.MessageId = "123";
            data.ContentType = "text/jesse";
            data.CorrelationId = "ABCDEF";

            using var amqpMessage = converter.CreateMessageFromEvent(data);
            using var serializedMessageStream = Serialize(amqpMessage);

            using var newMessage = AmqpMessage.CreateAmqpStreamMessage(serializedMessageStream);
            var newData = converter.CreateEventFromMessage(newMessage);

            Assert.That(amqpMessage.SerializedMessageSize, Is.EqualTo(newMessage.SerializedMessageSize), "The serialized size should match.");
            Assert.That(data.IsEquivalentTo(newData, false), Is.True, "The events should be the same");

            using var streamBytes = new MemoryStream();
            serializedMessageStream.CopyTo(streamBytes);
            serializedMessageStream.Dispose();

            using var bufferStream = BufferListStream.Create(streamBytes, 4096);
            using var byteMessage = AmqpMessage.CreateAmqpStreamMessage(bufferStream);
            var byteData = converter.CreateEventFromMessage(byteMessage);

            Assert.That(amqpMessage.SerializedMessageSize, Is.EqualTo(byteMessage.SerializedMessageSize), "The serialized size should match.");
            Assert.That(data.IsEquivalentTo(byteData, false), Is.True, "The events should be the same");
        }


        [Test]
        public void JESSE_TEST_SERIALIZATION_SINGLE_FILE()
        {
            var converter = new AmqpMessageConverter();

            var data = EventHubsModelFactory.EventData(
                new BinaryData("this is a body"),
                new Dictionary<string, object> { { "appKey", "appValue" }},
                null,
                "part-key",
                9989,
                1121,
                new DateTimeOffset(2015, 10, 27, 0, 0, 0, TimeSpan.Zero));

            data.MessageId = "123";
            data.ContentType = "text/plain";
            data.CorrelationId = "ABCDEF";

            var outName = Path.GetTempFileName();
            try
            {
                using var amqpMessage = converter.CreateMessageFromEvent(data);
                using var serializedMessageStream = amqpMessage.ToStream();
                using var outFile = File.OpenWrite(outName);

                serializedMessageStream.CopyTo(outFile);
                outFile.Close();
                outFile.Dispose();

                using var streamBytes = File.OpenRead(outName);
                using var bufferStream = BufferListStream.Create(streamBytes, 4096);
                using var byteMessage = AmqpMessage.CreateAmqpStreamMessage(bufferStream);

                var byteData = converter.CreateEventFromMessage(byteMessage);

                streamBytes.Close();
                streamBytes.Dispose();

                Assert.That(amqpMessage.SerializedMessageSize, Is.EqualTo(byteMessage.SerializedMessageSize), "The serialized size should match.");
                Assert.That(data.IsEquivalentTo(byteData, false), Is.True, "The events should be the same");
            }
            finally
            {
                try { File.Delete(outName); }  catch {}
            }
        }