| /* |
| * 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); |
| } |
| } |