History

TimeSeriesPlatformLibrary: Added temporal support options to DataPublisher interfaces.

J. Ritchie Carroll committed 7 years ago

Commit 5c5b4061 parent 04505f228 files · +603 −300Browse files

Showing the changes to Source/Libraries/TimeSeriesPlatformLibrary/Transport/SubscriberConnection.cpp (1 of 28 files). Show all

Unified · Split

Source/Libraries/TimeSeriesPlatformLibrary/Transport/SubscriberConnection.cpp+114−58
@@ -48,6 +48,9 @@ SubscriberConnection::SubscriberConnection(DataPublisherPtr parent, IOContext& c · 47 unmodified lines
4848 m_subscriberID(NewGuid()),
4949 m_operationalModes(OperationalModes::NoFlags),
5050 m_encoding(OperationalEncoding::UTF8),
51+ m_startTimeConstraint(DateTime::MaxValue),
52+ m_stopTimeConstraint(DateTime::MaxValue),
53+ m_processingInterval(-1),
5154 m_usePayloadCompression(false),
5255 m_useCompactMeasurementFormat(true),
5356 m_includeTime(true),
@@ -80,6 +83,11 @@ const DataPublisherPtr& SubscriberConnection::GetParent() const · 26 unmodified lines
8083 return m_parent;
8184 }
8285
86+SubscriberConnectionPtr SubscriberConnection::GetReference()
87+{
88+ return shared_from_this();
89+}
90+
8391 TcpSocket& SubscriberConnection::CommandChannelSocket()
8492 {
8593 return m_commandChannelSocket;
@@ -126,6 +134,43 @@ uint32_t SubscriberConnection::GetEncoding() const · 40 unmodified lines
126134 return m_encoding;
127135 }
128136
137+bool SubscriberConnection::GetIsTemporalSubscription() const
138+{
139+ return m_startTimeConstraint < DateTime::MaxValue;
140+}
141+
142+const GSF::datetime_t& SubscriberConnection::GetStartTimeConstraint() const
143+{
144+ return m_startTimeConstraint;
145+}
146+
147+void SubscriberConnection::SetStartTimeConstraint(const GSF::datetime_t& value)
148+{
149+ m_startTimeConstraint = value;
150+}
151+
152+const GSF::datetime_t& SubscriberConnection::GetStopTimeConstraint() const
153+{
154+ return m_stopTimeConstraint;
155+}
156+
157+void SubscriberConnection::SetStopTimeConstraint(const GSF::datetime_t& value)
158+{
159+ m_stopTimeConstraint = value;
160+}
161+
162+int32_t SubscriberConnection::GetProcessingInterval() const
163+{
164+ return m_processingInterval;
165+}
166+
167+void SubscriberConnection::SetProcessingInterval(int32_t value)
168+{
169+ m_processingInterval = value;
170+ m_parent->DispatchTemporalProcessingIntervalChangeRequested(shared_from_this());
171+ m_parent->DispatchStatusMessage(m_connectionID + " was assigned a new processing interval of " + ToString(value) + ".");
172+}
173+
129174 bool SubscriberConnection::GetUsePayloadCompression() const
130175 {
131176 return m_usePayloadCompression;
@@ -313,50 +358,7 @@ void SubscriberConnection::Stop() · 181 unmodified lines
313358 m_pingTimer.Stop();
314359 m_commandChannelSocket.shutdown(socket_base::shutdown_both);
315360 m_commandChannelSocket.cancel();
316− m_parent->RemoveConnection(shared_from_this());
317−}
318−
319−void SubscriberConnection::PublishMeasurements(const vector<Measurement>& measurements)
320−{
321− if (measurements.empty() || !m_isSubscribed)
322− return;
323−
324− if (!m_startTimeSent)
325− m_startTimeSent = SendDataStartTime(measurements[0].Timestamp);
326−
327− // TODO: Consider queuing measurements for processing
328−
329− CompactMeasurement serializer(m_signalIndexCache, m_baseTimeOffsets, m_includeTime, m_useCompactMeasurementFormat);
330− vector<uint8_t> packet, buffer;
331− int32_t count = 0;
332−
333− packet.reserve(MaxPacketSize);
334− buffer.reserve(16);
335−
336− for (size_t i = 0; i < measurements.size(); i++)
337− {
338− const Measurement& measurement = measurements[i];
339− const uint16_t runtimeID = m_signalIndexCache->GetSignalIndex(measurement.SignalID);
340−
341− if (runtimeID == UInt16::MaxValue)
342− continue;
343−
344− const uint32_t length = serializer.SerializeMeasurement(measurement, buffer, runtimeID);
345−
346− if (packet.size() + length > MaxPacketSize)
347− {
348− PublishDataPacket(packet, count);
349− packet.clear();
350− count = 0;
351− }
352−
353− WriteBytes(packet, buffer);
354− buffer.clear();
355− count++;
356− }
357−
358− if (count > 0)
359− PublishDataPacket(packet, count);
361+ m_parent->ConnectionTerminated(shared_from_this());
360362 }
361363
362364 void SubscriberConnection::PublishMeasurements(const vector<MeasurementPtr>& measurements)
@@ -402,6 +404,11 @@ void SubscriberConnection::PublishMeasurements(const vector<MeasurementPtr>& mea · 39 unmodified lines
402404 PublishDataPacket(packet, count);
403405 }
404406
407+void SubscriberConnection::CompleteTemporalSubscription()
408+{
409+ SendResponse(ServerResponse::ProcessingComplete, ServerCommand::Subscribe, ToString(m_parent->GetNodeID()));
410+}
411+
405412 void SubscriberConnection::HandleSubscribe(uint8_t* data, uint32_t length)
406413 {
407414 try
@@ -441,6 +448,24 @@ void SubscriberConnection::HandleSubscribe(uint8_t* data, uint32_t length) · 33 unmodified lines
441448 if (TryGetValue(settings, "requestNaNValueFilter", setting))
442449 SetIsNaNFiltered(ParseBoolean(setting));
443450
451+ if (TryGetValue(settings, "startTimeConstraint", setting))
452+ TryParseTimestamp(setting.c_str(), m_startTimeConstraint, DateTime::MaxValue);
453+
454+ if (TryGetValue(settings, "startTimeConstraint", setting))
455+ TryParseTimestamp(setting.c_str(), m_startTimeConstraint, DateTime::MaxValue);
456+
457+ if (TryGetValue(settings, "processingInterval", setting) && !setting.empty())
458+ TryParseInt32(setting, m_processingInterval, -1);
459+
460+ if (GetIsTemporalSubscription())
461+ {
462+ if (!m_parent->GetSupportsTemporalSubscriptions())
463+ throw PublisherException("Publisher does not temporal subscriptions");
464+
465+ if (m_startTimeConstraint > m_stopTimeConstraint)
466+ throw PublisherException("Specified stop time of requested temporal subscription precedes start time");
467+ }
468+
444469 SetUsePayloadCompression(usePayloadCompression);
445470 SetUseCompactMeasurementFormat(useCompactMeasurementFormat);
446471
@@ -518,36 +543,48 @@ void SubscriberConnection::HandleSubscribe(uint8_t* data, uint32_t length) · 71 unmodified lines
518543 SetIsSubscribed(true);
519544 SendResponse(ServerResponse::Succeeded, ServerCommand::Subscribe, message);
520545 m_parent->DispatchStatusMessage(message);
546+
547+ if (GetIsTemporalSubscription())
548+ m_parent->DispatchTemporalSubscriptionRequested(shared_from_this());
521549 }
522550 else
523551 {
524− const string message = byteLength > 0 ?
552+ HandleSubscribeFailure(byteLength > 0 ?
525553 "Not enough buffer was provided to parse client data subscription." :
526− "Cannot initialize client data subscription without a connection string.";
527−
528− SendResponse(ServerResponse::Failed, ServerCommand::Subscribe, message);
529− m_parent->DispatchErrorMessage(message);
554+ "Cannot initialize client data subscription without a connection string.");
530555 }
531556 }
532557 }
533558 else
534559 {
535− const string message = "Not enough buffer was provided to parse client data subscription.";
536− SendResponse(ServerResponse::Failed, ServerCommand::Subscribe, message);
537− m_parent->DispatchErrorMessage(message);
560+ HandleSubscribeFailure("Not enough buffer was provided to parse client data subscription.");
538561 }
539562 }
540− catch (const std::exception& ex)
563+ catch (const PublisherException& ex)
541564 {
542− const string message = "Failed to process client data subscription due to exception: " + string(ex.what());
543− SendResponse(ServerResponse::Failed, ServerCommand::Subscribe, message);
544− m_parent->DispatchErrorMessage(message);
565+ HandleSubscribeFailure("Failed to process client data subscription due to exception: " + string(ex.what()));
566+ }
567+ catch (...)
568+ {
569+ HandleSubscribeFailure("Failed to process client data subscription due to exception: " + boost::current_exception_diagnostic_information(true));
545570 }
546571 }
547572
573+void SubscriberConnection::HandleSubscribeFailure(const std::string& message)
574+{
575+ SendResponse(ServerResponse::Failed, ServerCommand::Subscribe, message);
576+ m_parent->DispatchErrorMessage(message);
577+
578+ if (GetIsTemporalSubscription())
579+ CompleteTemporalSubscription();
580+}
581+
548582 void SubscriberConnection::HandleUnsubscribe()
549583 {
550584 SetIsSubscribed(false);
585+
586+ if (GetIsTemporalSubscription())
587+ CompleteTemporalSubscription();
551588 }
552589
553590 void SubscriberConnection::HandleMetadataRefresh(uint8_t* data, uint32_t length)
@@ -559,7 +596,7 @@ void SubscriberConnection::HandleMetadataRefresh(uint8_t* data, uint32_t length) · 5 unmodified lines
559596 m_parent->DispatchStatusMessage("Received meta-data refresh request from " + GetConnectionID() + ", preparing response...");
560597
561598 StringMap<ExpressionTreePtr> filterExpressions;
562− const DateTime startTime = UtcNow(); //-V821
599+ const datetime_t startTime = UtcNow(); //-V821
563600
564601 try
565602 {
@@ -622,8 +659,22 @@ void SubscriberConnection::HandleRotateCipherKeys() · 56 unmodified lines
622659 {
623660 }
624661
625−void SubscriberConnection::HandleUpdateProcessingInterval(uint8_t* data, uint32_t length)
662+void SubscriberConnection::HandleUpdateProcessingInterval(const uint8_t* data, uint32_t length)
626663 {
664+ // Make sure there is enough buffer for new processing interval value
665+ if (length >= 4)
666+ {
667+ // Next 4 bytes are an integer representing the new processing interval
668+ const int32_t processingInterval = EndianConverter::Default.ConvertBigEndian(*reinterpret_cast<const int32_t*>(data));
669+ SetProcessingInterval(processingInterval);
670+ SendResponse(ServerResponse::Succeeded, ServerCommand::UpdateProcessingInterval, "New processing interval of " + ToString(processingInterval) + " assigned.");
671+ }
672+ else
673+ {
674+ const string message = "Not enough buffer was provided to update client processing interval.";
675+ SendResponse(ServerResponse::Failed, ServerCommand::UpdateProcessingInterval, message);
676+ m_parent->DispatchErrorMessage(message);
677+ }
627678 }
628679
629680 void SubscriberConnection::HandleDefineOperationalModes(uint8_t* data, uint32_t length)
@@ -1056,6 +1107,11 @@ void SubscriberConnection::WriteHandler(const ErrorCode& error, uint32_t bytesTr · 426 unmodified lines
10561107 }
10571108 }
10581109
1110+bool SubscriberConnection::SendResponse(uint8_t responseCode, uint8_t commandCode)
1111+{
1112+ return SendResponse(responseCode, commandCode, vector<uint8_t>(0));
1113+}
1114+
10591115 bool SubscriberConnection::SendResponse(uint8_t responseCode, uint8_t commandCode, const string& message)
10601116 {
10611117 return SendResponse(responseCode, commandCode, EncodeString(message));

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