//******************************************************************************************************//  MeasurementExtensions.cs - Gbtc////  Copyright © 2015, Grid Protection Alliance.  All Rights Reserved.////  Licensed to the Grid Protection Alliance (GPA) under one or more contributor license agreements. See//  the NOTICE file distributed with this work for additional information regarding copyright ownership.//  The GPA licenses this file to you under the MIT License (MIT), the "License"; you may not use this//  file except in compliance with the License. You may obtain a copy of the License at:////      http://opensource.org/licenses/MIT////  Unless agreed to in writing, the subject software distributed under the License is distributed on an//  "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Refer to the//  License for the specific language governing permissions and limitations.////  Code Modification History://  ----------------------------------------------------------------------------------------------------//  10/22/2015 - Ritchie Carroll//       Generated original version of source code.////****************************************************************************************************** using System;using System.Runtime.CompilerServices;using GSF;using GSF.TimeSeries;using Misakai.Kafka; namespace KafkaAdapters{    #region [ Enumerations ]     /// <summary>    /// Defines the data quality flags applied to a Kafka serialized measurement.    /// </summary>    [Flags]    public enum DataQualityFlags : byte    {        /// <summary>        /// If no flags are set, measurement quality is normal.        /// </summary>        Normal = (byte)Bits.Nil,        /// <summary>        /// If bit 0 is set, measurement value quality is bad - as reported by source.        /// </summary>        ValueQualityIsBad = (byte)Bits.Bit00,        /// <summary>        /// If bit 1 is set, measurement time quality is bad - as reported by source.        /// </summary>        TimeQualityIsBad = (byte)Bits.Bit01,        /// <summary>        /// Decode version mask - remaining bits used for decoding version, allows 64 possible versions (0 to 63).        /// </summary>        DecodeVersionMask = (byte)(Bits.Bit02 | Bits.Bit03 | Bits.Bit04 | Bits.Bit05 | Bits.Bit06 | Bits.Bit07)    }     #endregion     /// <summary>    /// Defines extension methods for handling Kafka message serialization to and from time-series measurements.    /// </summary>    /// <remarks>    /// The goal of this serialization implementation is to make it very simple to deserialize    /// Kafka time-series messages from other language implementations, such as C++ or Java.    /// </remarks>    public static class SerializationExtensions    {        /// <summary>        /// Defines the size of serialized Kafka measurement key.        /// </summary>        /// <remarks>        /// Key format:<br/>        /// - long Time: 64-bit time of measurement in ticks, i.e., 100-nanosecond intervals since 0/0/0000<br/>        /// - uint ID:   32-bit measurement identifier (temporal index into metadata)        /// </remarks>        public const int KeySize = sizeof(long) + sizeof(uint);                     // 12 bytes         /// <summary>        /// Defines the size of serialized Kafka measurement value.        /// </summary>        /// <remarks>        /// Value format:<br/>        /// - byte Version: 8-bit metadata serialization version<br/>        /// - byte Quality: 8-bit measurement quality - bit 0 = bad value, bit 1 = bad time*<br/>        /// - double Value: 64-bit floating-point measurement value        /// * Remaining bits used to define decoding version.        /// </remarks>        public const int ValueSize = sizeof(byte) + sizeof(byte) + sizeof(double);  // 10 bytes         /// <summary>        /// Defines the current message decoding version.        /// </summary>        public const byte DecodeVersion = 0;         /// <summary>        /// Serializes a time-series <see cref="IMeasurement"/> instance into key-value buffers for a Kafka message.        /// </summary>        /// <param name="measurement"><see cref="IMeasurement"/> instance to serialize.</param>        /// <param name="message">Kafka message to hold serialized key-value buffers.</param>        /// <param name="metadataVersion">Current metadata serialization version.</param>        [MethodImpl(MethodImplOptions.AggressiveInlining)]        public static void KakfaSerialize(this IMeasurement measurement, Message message, byte metadataVersion)        {            if ((object)measurement == null)                throw new ArgumentNullException(nameof(measurement));             if ((object)message == null)                throw new ArgumentNullException(nameof(message));             // Copy in "key" for Kafka message, for a time-series entity, the "timestamp" and "ID"            message.Key = new byte[KeySize];            Buffer.BlockCopy(BitConverter.GetBytes(measurement.Timestamp), 0, message.Key, 0, 8);            Buffer.BlockCopy(BitConverter.GetBytes(measurement.Key.ID), 0, message.Key, 8, 4);             // Copy in "value" for Kafka message, for a time-series entity, the "version", "quality" and "value"            message.Value = new byte[ValueSize];            message.Value[0] = metadataVersion;            message.Value[1] = (byte)((int)measurement.StateFlags.DeriveQualityFlags() | (DecodeVersion << 2));            Buffer.BlockCopy(BitConverter.GetBytes(measurement.AdjustedValue), 0, message.Value, 2, 8);        }         /// <summary>        /// Deserializes a Kafka message into a time-series <see cref="Measurement"/>.        /// </summary>        /// <param name="message">Time-series serialized Kafka message.</param>        /// <param name="id">Numeric ID of measurement deserialized from Kafka message.</param>        /// <param name="metadataVersion">Message metadata serialization version.</param>        /// <returns><see cref="Measurement"/> value deserialized from a time-series serialized Kafka message.</returns>        [MethodImpl(MethodImplOptions.AggressiveInlining)]        public static Measurement KafkaDeserialize(this Message message, out uint id, out byte metadataVersion)        {            if ((object)message == null)                throw new ArgumentNullException(nameof(message));             if (message.Key.Length < KeySize)                throw new ArgumentOutOfRangeException(nameof(message.Key), $"Buffer size too small - expected {KeySize} bytes for Kafka time-series message key");             if (message.Value.Length < ValueSize)                throw new ArgumentOutOfRangeException(nameof(message.Value), $"Buffer size too small - expected {ValueSize} bytes for Kafka time-series message value");             byte messageDecodeVersion = (byte)((message.Value[1] & (byte)DataQualityFlags.DecodeVersionMask) >> 2);             if (messageDecodeVersion > DecodeVersion)                throw new InvalidOperationException($"This Kafka time-series deserialization engine cannot decode version {messageDecodeVersion} serialized messages");             id = BitConverter.ToUInt32(message.Key, 8);            metadataVersion = message.Value[0];             return new Measurement            {                Timestamp = BitConverter.ToInt64(message.Key, 0),                Value = BitConverter.ToDouble(message.Value, 2),                StateFlags = DeriveStateFlags((DataQualityFlags)message.Value[1] & ~DataQualityFlags.DecodeVersionMask)                            };        }         /// <summary>        /// Derive Kafka time-seres data quality flags from measurement state flags         /// </summary>        /// <param name="stateFlags"></param>        /// <returns></returns>        [MethodImpl(MethodImplOptions.AggressiveInlining)]        public static DataQualityFlags DeriveQualityFlags(this MeasurementStateFlags stateFlags)        {            DataQualityFlags qualityFlags = DataQualityFlags.Normal;             if (stateFlags.HasFlag(MeasurementStateFlags.BadData))                qualityFlags |= DataQualityFlags.ValueQualityIsBad;             if (stateFlags.HasFlag(MeasurementStateFlags.BadTime))                qualityFlags |= DataQualityFlags.TimeQualityIsBad;             return qualityFlags;        }         /// <summary>        /// Derive measurement state flags from Kafka time-seres data quality flags        /// </summary>        /// <param name="qualityFlags"></param>        /// <returns></returns>        [MethodImpl(MethodImplOptions.AggressiveInlining)]        public static MeasurementStateFlags DeriveStateFlags(this DataQualityFlags qualityFlags)        {            MeasurementStateFlags stateFlags = MeasurementStateFlags.Normal;             if (qualityFlags.HasFlag(DataQualityFlags.ValueQualityIsBad))                stateFlags |= MeasurementStateFlags.BadData;             if (qualityFlags.HasFlag(DataQualityFlags.TimeQualityIsBad))                stateFlags |= MeasurementStateFlags.BadTime;             return stateFlags;        }    }}