History

GSF.TimeSeries: Added a handler for a new ServerCommand so that the DataPublisher may receive measurements from a DataSubscriber.

StephenCWills committed 9 years ago

Commit 4d2b00a1 parent bf057666 files · +197 −7Browse files

Unified · Split

6 files
  1. Source/Libraries/GSF.TimeSeries/GSF.TimeSeries.csproj +1 −0
  2. Source/Libraries/GSF.TimeSeries/Transport/CommandMeasurement.cs +93 −0
  3. Source/Libraries/GSF.TimeSeries/Transport/DataPublisher.cs +52 −0
  4. Source/Libraries/TimeSeriesPlatformLibrary/Transport/Constants.h +15 −7
  5. Source/Libraries/TimeSeriesPlatformLibrary/Transport/DataSubscriber.cpp +32 −0
  6. Source/Libraries/TimeSeriesPlatformLibrary/Transport/DataSubscriber.h +4 −0
Source/Libraries/GSF.TimeSeries/GSF.TimeSeries.csproj+1−0
@@ -148,6 +148,7 @@ · 147 unmodified lines
148 <Compile Include="Statistics\StatisticValueStateCache.cs" />148 <Compile Include="Statistics\StatisticValueStateCache.cs" />
149 <Compile Include="TimeSeriesStartupOperations.cs" />149 <Compile Include="TimeSeriesStartupOperations.cs" />
150 <Compile Include="BufferBlockMeasurement.cs" />150 <Compile Include="BufferBlockMeasurement.cs" />
151 <Compile Include="Transport\CommandMeasurement.cs" />
151 <Compile Include="Transport\SubscriberRightsLookup.cs" />152 <Compile Include="Transport\SubscriberRightsLookup.cs" />
152 <Compile Include="Transport\ClientConnection.cs" />153 <Compile Include="Transport\ClientConnection.cs" />
153 <Compile Include="Transport\Common.cs" />154 <Compile Include="Transport\Common.cs" />
Source/Libraries/GSF.TimeSeries/Transport/CommandMeasurement.cs+93−0
@@ -0,0 +1,93 @@
1//******************************************************************************************************
2// CommandMeasurement.cs - Gbtc
3//
4// Copyright © 2016, Grid Protection Alliance. All Rights Reserved.
5//
6// Licensed to the Grid Protection Alliance (GPA) under one or more contributor license agreements. See
7// the NOTICE file distributed with this work for additional information regarding copyright ownership.
8// The GPA licenses this file to you under the MIT License (MIT), the "License"; you may
9// not use this file except in compliance with the License. You may obtain a copy of the License at:
10//
11// http://opensource.org/licenses/MIT
12//
13// Unless agreed to in writing, the subject software distributed under the License is distributed on an
14// "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Refer to the
15// License for the specific language governing permissions and limitations.
16//
17// Code Modification History:
18// ----------------------------------------------------------------------------------------------------
19// 11/18/2016 - Stephen C. Wills
20// Generated original version of source code.
21//
22//******************************************************************************************************
23
24using System;
25using System.ComponentModel;
26using System.Configuration;
27using System.Globalization;
28
29namespace GSF.TimeSeries.Transport
30{
31 /// <summary>
32 /// Model for measurements published by the subscriber.
33 /// </summary>
34 public class CommandMeasurement
35 {
36 /// <summary>
37 /// Gets or sets the signal ID of the measurement.
38 /// </summary>
39 [Setting]
40 public Guid SignalID { get; set; }
41
42 /// <summary>
43 /// Gets or sets the time that the measurement was taken.
44 /// </summary>
45 [Setting]
46 [TypeConverter(typeof(DateTimeConverter))]
47 public DateTime Timestamp { get; set; }
48
49 /// <summary>
50 /// Gets or sets the value of the measurement.
51 /// </summary>
52 [Setting]
53 public double Value { get; set; }
54
55 /// <summary>
56 /// Gets or sets the flags that define the quality of the measurement.
57 /// </summary>
58 [Setting]
59 public MeasurementStateFlags StateFlags { get; set; }
60
61 #region [ DateTime Converter ]
62
63 private class DateTimeConverter : TypeConverter
64 {
65 TypeConverter defaultConverter = TypeDescriptor.GetConverter(typeof(DateTime));
66
67 public override bool CanConvertFrom(ITypeDescriptorContext context, Type sourceType)
68 {
69 return defaultConverter.CanConvertFrom(context, sourceType);
70 }
71
72 public override bool CanConvertTo(ITypeDescriptorContext context, Type destinationType)
73 {
74 return defaultConverter.CanConvertTo(context, destinationType);
75 }
76
77 public override object ConvertFrom(ITypeDescriptorContext context, CultureInfo culture, object value)
78 {
79 return defaultConverter.ConvertFrom(context, culture, value);
80 }
81
82 public override object ConvertTo(ITypeDescriptorContext context, CultureInfo culture, object value, Type destinationType)
83 {
84 if (destinationType == typeof(string))
85 return ((DateTime)value).ToString("yyyy-MM-dd HH:mm:ss.fffffff");
86
87 return base.ConvertTo(context, culture, value, destinationType);
88 }
89 }
90
91 #endregion
92 }
93}
Source/Libraries/GSF.TimeSeries/Transport/DataPublisher.cs+52−0
@@ -37,6 +37,7 @@ using System; · 36 unmodified lines
37using System.Collections.Concurrent;37using System.Collections.Concurrent;
38using System.Collections.Generic;38using System.Collections.Generic;
39using System.ComponentModel;39using System.ComponentModel;
40using System.Configuration;
40using System.Data;41using System.Data;
41using System.Globalization;42using System.Globalization;
42using System.IO;43using System.IO;
@@ -192,6 +193,15 @@ namespace GSF.TimeSeries.Transport · 149 unmodified lines
192 /// </remarks>193 /// </remarks>
193 ConfirmBufferBlock = 0x08,194 ConfirmBufferBlock = 0x08,
194 195
196 /// <summary>
197 /// Provides measurements to the publisher over the command channel.
198 /// </summary>
199 /// <remarks>
200 /// Allows for unsolicited publication of measurement data to the server
201 /// so that consumers of data can also provide data to other consumers.
202 /// </remarks>
203 PublishCommandMeasurements = 0x09,
204
195 /// <summary>205 /// <summary>
196 /// Code for handling user-defined commands.206 /// Code for handling user-defined commands.
197 /// </summary>207 /// </summary>
@@ -3671,6 +3681,43 @@ namespace GSF.TimeSeries.Transport · 3473 unmodified lines
3671 }3681 }
3672 }3682 }
3673 3683
3684 private void HandlePublishCommandMeasurements(ClientConnection connection, byte[] buffer, int startIndex, int length)
3685 {
3686 try
3687 {
3688 List<IMeasurement> measurements = new List<IMeasurement>();
3689
3690 int index = startIndex;
3691 int payloadByteLength = BigEndian.ToInt32(buffer, index);
3692 index += sizeof(int);
3693
3694 string dataString = connection.Encoding.GetString(buffer, index, payloadByteLength);
3695 ConnectionStringParser<SettingAttribute> connectionStringParser = new ConnectionStringParser<SettingAttribute>();
3696
3697 foreach (string measurementString in dataString.Split(new[] { ";;" }, StringSplitOptions.RemoveEmptyEntries))
3698 {
3699 CommandMeasurement measurement = new CommandMeasurement();
3700 connectionStringParser.ParseConnectionString(measurementString, measurement);
3701
3702 measurements.Add(new Measurement()
3703 {
3704 Metadata = MeasurementKey.LookUpBySignalID(measurement.SignalID).Metadata,
3705 Timestamp = measurement.Timestamp,
3706 Value = measurement.Value,
3707 StateFlags = measurement.StateFlags
3708 });
3709 }
3710
3711 OnNewMeasurements(measurements);
3712 }
3713 catch (Exception ex)
3714 {
3715 string errorMessage = $"Data packet command failed due to exception: {ex.Message}";
3716 OnProcessException(MessageLevel.Error, new Exception(errorMessage, ex));
3717 SendClientResponse(connection.ClientID, ServerResponse.Failed, ServerCommand.PublishCommandMeasurements, errorMessage);
3718 }
3719 }
3720
3674 /// <summary>3721 /// <summary>
3675 /// Handles custom commands defined by the user of the publisher API.3722 /// Handles custom commands defined by the user of the publisher API.
3676 /// </summary>3723 /// </summary>
@@ -3940,6 +3987,11 @@ namespace GSF.TimeSeries.Transport · 263 unmodified lines
3940 HandleConfirmBufferBlock(connection, buffer, index, length);3987 HandleConfirmBufferBlock(connection, buffer, index, length);
3941 break;3988 break;
3942 3989
3990 case ServerCommand.PublishCommandMeasurements:
3991 // Handle publication of command measurements
3992 HandlePublishCommandMeasurements(connection, buffer, index, length);
3993 break;
3994
3943 case ServerCommand.UserCommand00:3995 case ServerCommand.UserCommand00:
3944 case ServerCommand.UserCommand01:3996 case ServerCommand.UserCommand01:
3945 case ServerCommand.UserCommand02:3997 case ServerCommand.UserCommand02:
Source/Libraries/TimeSeriesPlatformLibrary/Transport/Constants.h+15−7
@@ -40,13 +40,16 @@ namespace Transport · 39 unmodified lines
40 40
41 struct ServerCommand41 struct ServerCommand
42 {42 {
43 static const uint8_t Authenticate = 0x00;43 static const uint8_t Authenticate = 0x00;
44 static const uint8_t MetadataRefresh = 0x01;44 static const uint8_t MetadataRefresh = 0x01;
45 static const uint8_t Subscribe = 0x02;45 static const uint8_t Subscribe = 0x02;
46 static const uint8_t Unsubscribe = 0x03;46 static const uint8_t Unsubscribe = 0x03;
47 static const uint8_t RotateCipherKeys = 0x04;47 static const uint8_t RotateCipherKeys = 0x04;
48 static const uint8_t UpdateProcessingInterval = 0x05;48 static const uint8_t UpdateProcessingInterval = 0x05;
49 static const uint8_t DefineOperationalModes = 0x06;49 static const uint8_t DefineOperationalModes = 0x06;
50 static const uint8_t ConfirmNotification = 0x07;
51 static const uint8_t ConfirmBufferBlock = 0x08;
52 static const uint8_t PublishCommandMeasurements = 0x09;
50 };53 };
51 54
52 struct ServerResponse55 struct ServerResponse
@@ -59,6 +62,9 @@ namespace Transport · 6 unmodified lines
59 static const uint8_t UpdateCipherKeys = 0x85;62 static const uint8_t UpdateCipherKeys = 0x85;
60 static const uint8_t DataStartTime = 0x86;63 static const uint8_t DataStartTime = 0x86;
61 static const uint8_t ProcessingComplete = 0x87;64 static const uint8_t ProcessingComplete = 0x87;
65 static const uint8_t BufferBlock = 0x88;
66 static const uint8_t Notify = 0x89;
67 static const uint8_t ConfigurationChanged = 0x8A;
62 static const uint8_t NoOP = 0xFF;68 static const uint8_t NoOP = 0xFF;
63 };69 };
64 70
@@ -68,6 +74,8 @@ namespace Transport · 3 unmodified lines
68 static const uint32_t CompressionModeMask = 0x000000E0;74 static const uint32_t CompressionModeMask = 0x000000E0;
69 static const uint32_t EncodingMask = 0x00000300;75 static const uint32_t EncodingMask = 0x00000300;
70 static const uint32_t UseCommonSerializationFormat = 0x01000000;76 static const uint32_t UseCommonSerializationFormat = 0x01000000;
77 static const uint32_t ReceiveExternalMetadata = 0x02000000;
78 static const uint32_t ReceiveInternalMetadata = 0x04000000;
71 static const uint32_t CompressSignalIndexCache = 0x40000000;79 static const uint32_t CompressSignalIndexCache = 0x40000000;
72 static const uint32_t CompressMetadata = 0x80000000;80 static const uint32_t CompressMetadata = 0x80000000;
73 static const uint32_t NoFlags = 0x00000000;81 static const uint32_t NoFlags = 0x00000000;
Source/Libraries/TimeSeriesPlatformLibrary/Transport/DataSubscriber.cpp+32−0
@@ -852,6 +852,38 @@ void gsfts::Transport::DataSubscriber::SendServerCommand(uint8_t commandCode) · 851 unmodified lines
852 SendServerCommand(commandCode, 0, 0, 0);852 SendServerCommand(commandCode, 0, 0, 0);
853}853}
854 854
855// Sends a command along with the given message to the server.
856void gsfts::Transport::DataSubscriber::SendServerCommand(uint8_t commandCode, std::string message)
857{
858 const std::size_t CharSize = sizeof(char);
859
860 uint32_t bufferSize;
861 std::vector<uint8_t> buffer;
862
863 uint8_t* messagePtr;
864 uint32_t messageSize;
865 uint32_t bigEndianMessageSize;
866 uint8_t* bigEndianMessageSizePtr;
867
868 messagePtr = (uint8_t*)&message[0];
869 messageSize = (uint32_t)(message.size() * CharSize);
870 bigEndianMessageSize = m_endianConverter.ConvertBigEndian(messageSize);
871 bigEndianMessageSizePtr = (uint8_t*)&bigEndianMessageSize;
872
873 bufferSize = 4 + messageSize;
874 buffer.reserve(bufferSize);
875
876 buffer[0] = bigEndianMessageSizePtr[0];
877 buffer[1] = bigEndianMessageSizePtr[1];
878 buffer[2] = bigEndianMessageSizePtr[2];
879 buffer[3] = bigEndianMessageSizePtr[3];
880
881 for (i = 0; i < messageSize; ++i)
882 buffer[4 + i] = messagePtr[i];
883
884 SendServerCommand(commandCode, &buffer[0], 0, bufferSize);
885}
886
855// Sends a command along with the given data to the server.887// Sends a command along with the given data to the server.
856void gsfts::Transport::DataSubscriber::SendServerCommand(uint8_t commandCode, uint8_t* data, std::size_t offset, std::size_t length)888void gsfts::Transport::DataSubscriber::SendServerCommand(uint8_t commandCode, uint8_t* data, std::size_t offset, std::size_t length)
857{889{
Source/Libraries/TimeSeriesPlatformLibrary/Transport/DataSubscriber.h+4−0
@@ -300,7 +300,11 @@ namespace Transport · 299 unmodified lines
300 // ServerCommand::RotateCipherKeys300 // ServerCommand::RotateCipherKeys
301 // ServerCommand::UpdateProcessingInterval301 // ServerCommand::UpdateProcessingInterval
302 // ServerCommand::DefineOperationalModes302 // ServerCommand::DefineOperationalModes
303 // ServerCommand::ConfirmNotification
304 // ServerCommand::ConfirmBufferBlock
305 // ServerCommand::PublishCommandMeasurements
303 void SendServerCommand(uint8_t commandCode);306 void SendServerCommand(uint8_t commandCode);
307 void SendServerCommand(uint8_t commandCode, std::string message);
304 void SendServerCommand(uint8_t commandCode, uint8_t* data, std::size_t offset, std::size_t length);308 void SendServerCommand(uint8_t commandCode, uint8_t* data, std::size_t offset, std::size_t length);
305 309
306 // Convenience method to send the currently defined and/or supported310 // Convenience method to send the currently defined and/or supported

Keyboard shortcuts

General

Show keyboard shortcuts
?
Focus the current search
/

Issues and pull requests

Open filters
f

Issues, pull requests and Inbox

Select next item
j↓
Select previous item
k↑
Open selected item
Enter
Dismiss filters or clear selection
Esc

Repository code

Go to file
t