src/DotPulsar/Internal/MessageFactory.cs

/*
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

namespace DotPulsar.Internal
{
    using DotPulsar.Abstractions;
    using DotPulsar.Internal.Abstractions;
    using DotPulsar.Internal.PulsarApi;
    using System.Buffers;
    using System.Collections.Generic;

    public sealed class MessageFactory<TValue> : IMessageFactory<TValue>
    {
        private static readonly Dictionary<string, string> _empty;

        static MessageFactory() => _empty = new Dictionary<string, string>();

        private static IReadOnlyDictionary<string, string> FromKeyValueList(List<KeyValue> keyValues)
        {
            if (keyValues.Count == 0)
                return _empty;

            var dictionary = new Dictionary<string, string>(keyValues.Count);

            for (var i = 0; i < keyValues.Count; ++i)
            {
                var keyValue = keyValues[i];
                dictionary.Add(keyValue.Key, keyValue.Value);
            }

            return dictionary;
        }

        private readonly ISchema<TValue> _schema;

        public MessageFactory(ISchema<TValue> schema)
            => _schema = schema;

        public IMessage<TValue> Create(MessageId messageId, uint redeliveryCount, ReadOnlySequence<byte> data, MessageMetadata metadata, SingleMessageMetadata? singleMetadata = null)
        {
            if (singleMetadata is null)
                return Create(messageId, redeliveryCount, metadata, data);

            return Create(messageId, redeliveryCount, metadata, singleMetadata, data);
        }

        private Message<TValue> Create(
            MessageId messageId,
            uint redeliveryCount,
            MessageMetadata metadata,
            ReadOnlySequence<byte> data)
        {
            return new Message<TValue>(
                messageId: messageId,
                data: data,
                producerName: metadata.ProducerName,
                sequenceId: metadata.SequenceId,
                redeliveryCount: redeliveryCount,
                eventTime: metadata.EventTime,
                publishTime: metadata.PublishTime,
                properties: FromKeyValueList(metadata.Properties),
                hasBase64EncodedKey: metadata.PartitionKeyB64Encoded,
                key: metadata.PartitionKey,
                orderingKey: metadata.OrderingKey,
                schemaVersion: metadata.SchemaVersion,
                _schema);
        }

        private Message<TValue> Create(
            MessageId messageId,
            uint redeliveryCount,
            MessageMetadata metadata,
            SingleMessageMetadata singleMetadata,
            ReadOnlySequence<byte> data)
        {
            return new Message<TValue>(
                messageId: messageId,
                data: data,
                producerName: metadata.ProducerName,
                sequenceId: singleMetadata.SequenceId,
                redeliveryCount: redeliveryCount,
                eventTime: singleMetadata.EventTime,
                publishTime: metadata.PublishTime,
                properties: FromKeyValueList(singleMetadata.Properties),
                hasBase64EncodedKey: singleMetadata.PartitionKeyB64Encoded,
                key: singleMetadata.PartitionKey,
                orderingKey: singleMetadata.OrderingKey,
                schemaVersion: metadata.SchemaVersion,
                _schema);
        }
    }
}