History

TimeSeriesPlatformLibrary: Added extra null reference handling to DataSubscriber for cases where all callbacks may not be registered and race conditions on shutdown related to class termination

J. Ritchie Carroll committed 8 years ago

Commit be76cd31 parent 7a0085d1 file · +102 −49Browse files

Unified · Split

Source/Libraries/TimeSeriesPlatformLibrary/Transport/DataSubscriber.cpp+102−49
@@ -430,14 +430,17 @@ void DataSubscriber::HandleSucceeded(uint8_t commandCode, uint8_t* data, uint32_ · 429 unmodified lines
430 case ServerCommand::RotateCipherKeys:430 case ServerCommand::RotateCipherKeys:
431 // Each of these responses come with a message that will431 // Each of these responses come with a message that will
432 // be delivered to the user via the status message callback.432 // be delivered to the user via the status message callback.
433 messageStart = reinterpret_cast<char*>(data + offset);433 if (data != nullptr)
434 messageEnd = messageStart + messageLength;434 {
435 messageStream << "Received success code in response to server command 0x" << hex << static_cast<int>(commandCode) << ": ";435 messageStart = reinterpret_cast<char*>(data + offset);
436 messageEnd = messageStart + messageLength;
437 messageStream << "Received success code in response to server command 0x" << hex << static_cast<int>(commandCode) << ": ";
436 438
437 for (messageIter = messageStart; messageIter < messageEnd; ++messageIter)439 for (messageIter = messageStart; messageIter < messageEnd; ++messageIter)
438 messageStream << *messageIter;440 messageStream << *messageIter;
439 441
440 DispatchStatusMessage(messageStream.str());442 DispatchStatusMessage(messageStream.str());
443 }
441 break;444 break;
442 445
443 default:446 default:
@@ -453,6 +456,9 @@ void DataSubscriber::HandleSucceeded(uint8_t commandCode, uint8_t* data, uint32_ · 9 unmodified lines
453// Handles failure messages from the server.456// Handles failure messages from the server.
454void DataSubscriber::HandleFailed(uint8_t commandCode, uint8_t* data, uint32_t offset, uint32_t length)457void DataSubscriber::HandleFailed(uint8_t commandCode, uint8_t* data, uint32_t offset, uint32_t length)
455{458{
459 if (data == nullptr)
460 return;
461
456 const uint32_t messageLength = length / sizeof(char);462 const uint32_t messageLength = length / sizeof(char);
457 stringstream messageStream;463 stringstream messageStream;
458 464
@@ -497,6 +503,9 @@ void DataSubscriber::HandleProcessingComplete(uint8_t* data, uint32_t offset, ui · 38 unmodified lines
497// Cache signal IDs sent by the server into the signal index cache.503// Cache signal IDs sent by the server into the signal index cache.
498void DataSubscriber::HandleUpdateSignalIndexCache(uint8_t* data, uint32_t offset, uint32_t length)504void DataSubscriber::HandleUpdateSignalIndexCache(uint8_t* data, uint32_t offset, uint32_t length)
499{505{
506 if (data == nullptr)
507 return;
508
500 const bool swapBytes = m_endianConverter.NativeOrder() == EndianConverter::LittleEndian;509 const bool swapBytes = m_endianConverter.NativeOrder() == EndianConverter::LittleEndian;
501 510
502 uint32_t* referenceCountPtr;511 uint32_t* referenceCountPtr;
@@ -586,6 +595,9 @@ void DataSubscriber::HandleUpdateSignalIndexCache(uint8_t* data, uint32_t offset · 83 unmodified lines
586// Updates base time offsets.595// Updates base time offsets.
587void DataSubscriber::HandleUpdateBaseTimes(uint8_t* data, uint32_t offset, uint32_t length)596void DataSubscriber::HandleUpdateBaseTimes(uint8_t* data, uint32_t offset, uint32_t length)
588{597{
598 if (data == nullptr)
599 return;
600
589 int32_t* timeIndexPtr = reinterpret_cast<int32_t*>(data + offset);601 int32_t* timeIndexPtr = reinterpret_cast<int32_t*>(data + offset);
590 int64_t* timeOffsetsPtr = reinterpret_cast<int64_t*>(timeIndexPtr + 1);602 int64_t* timeOffsetsPtr = reinterpret_cast<int64_t*>(timeIndexPtr + 1);
591 603
@@ -648,43 +660,65 @@ void DataSubscriber::DispatchErrorMessage(const string& message) · 56 unmodified lines
648// Dispatcher function for status messages. Decodes the message and provides it to the user via the status message callback.660// Dispatcher function for status messages. Decodes the message and provides it to the user via the status message callback.
649void DataSubscriber::StatusMessageDispatcher(DataSubscriber* source, const vector<uint8_t>& buffer)661void DataSubscriber::StatusMessageDispatcher(DataSubscriber* source, const vector<uint8_t>& buffer)
650{662{
663 if (source == nullptr)
664 return;
665
651 const MessageCallback statusMessageCallback = source->m_statusMessageCallback;666 const MessageCallback statusMessageCallback = source->m_statusMessageCallback;
652 stringstream messageStream;667
668 if (statusMessageCallback != nullptr)
669 {
670 stringstream messageStream;
653 671
654 for (uint32_t i = 0; i < buffer.size(); ++i)672 for (uint32_t i = 0; i < buffer.size(); ++i)
655 messageStream << buffer[i];673 messageStream << buffer[i];
656 674
657 if (statusMessageCallback != nullptr)675 if (statusMessageCallback != nullptr)
658 statusMessageCallback(source, messageStream.str());676 statusMessageCallback(source, messageStream.str());
677 }
659}678}
660 679
661// Dispatcher function for error messages. Decodes the message and provides it to the user via the error message callback.680// Dispatcher function for error messages. Decodes the message and provides it to the user via the error message callback.
662void DataSubscriber::ErrorMessageDispatcher(DataSubscriber* source, const vector<uint8_t>& buffer)681void DataSubscriber::ErrorMessageDispatcher(DataSubscriber* source, const vector<uint8_t>& buffer)
663{682{
664 const MessageCallback errorMessageCallback = source->m_errorMessageCallback;683 if (source == nullptr)
665 stringstream messageStream;684 return;
666 685
667 for (uint32_t i = 0; i < buffer.size(); ++i)686 const MessageCallback errorMessageCallback = source->m_errorMessageCallback;
668 messageStream << buffer[i];
669 687
670 if (errorMessageCallback != nullptr)688 if (errorMessageCallback != nullptr)
671 errorMessageCallback(source, messageStream.str());689 {
690 stringstream messageStream;
691
692 for (uint32_t i = 0; i < buffer.size(); ++i)
693 messageStream << buffer[i];
694
695 if (errorMessageCallback != nullptr)
696 errorMessageCallback(source, messageStream.str());
697 }
672}698}
673 699
674// Dispatcher function for data start time. Decodes the start time and provides it to the user via the data start time callback.700// Dispatcher function for data start time. Decodes the start time and provides it to the user via the data start time callback.
675void DataSubscriber::DataStartTimeDispatcher(DataSubscriber* source, const vector<uint8_t>& buffer)701void DataSubscriber::DataStartTimeDispatcher(DataSubscriber* source, const vector<uint8_t>& buffer)
676{702{
703 if (source == nullptr)
704 return;
705
677 const DataStartTimeCallback dataStartTimeCallback = source->m_dataStartTimeCallback;706 const DataStartTimeCallback dataStartTimeCallback = source->m_dataStartTimeCallback;
678 EndianConverter endianConverter = source->m_endianConverter;
679 const int64_t dataStartTime = endianConverter.ConvertBigEndian(*reinterpret_cast<const int64_t*>(&buffer[0]));
680 707
681 if (dataStartTimeCallback != nullptr)708 if (dataStartTimeCallback != nullptr)
709 {
710 EndianConverter endianConverter = source->m_endianConverter;
711 const int64_t dataStartTime = endianConverter.ConvertBigEndian(*reinterpret_cast<const int64_t*>(&buffer[0]));
682 dataStartTimeCallback(source, dataStartTime);712 dataStartTimeCallback(source, dataStartTime);
713 }
683}714}
684 715
685// Dispatcher function for metadata. Provides encoded metadata to the user via the metadata callback.716// Dispatcher function for metadata. Provides encoded metadata to the user via the metadata callback.
686void DataSubscriber::MetadataDispatcher(DataSubscriber* source, const vector<uint8_t>& buffer)717void DataSubscriber::MetadataDispatcher(DataSubscriber* source, const vector<uint8_t>& buffer)
687{718{
719 if (source == nullptr)
720 return;
721
688 const MetadataCallback metadataCallback = source->m_metadataCallback;722 const MetadataCallback metadataCallback = source->m_metadataCallback;
689 723
690 if (metadataCallback != nullptr)724 if (metadataCallback != nullptr)
@@ -694,46 +728,53 @@ void DataSubscriber::MetadataDispatcher(DataSubscriber* source, const vector<uin · 3 unmodified lines
694// Dispatcher function for new measurements. Decodes the measurements and provides them to the user via the new measurements callback.728// Dispatcher function for new measurements. Decodes the measurements and provides them to the user via the new measurements callback.
695void DataSubscriber::NewMeasurementsDispatcher(DataSubscriber* source, const vector<uint8_t>& buffer)729void DataSubscriber::NewMeasurementsDispatcher(DataSubscriber* source, const vector<uint8_t>& buffer)
696{730{
697 if (source->m_newMeasurementsCallback == nullptr)731 if (source == nullptr)
698 return;732 return;
699 733
700 SubscriptionInfo& info = source->m_subscriptionInfo;734 const NewMeasurementsCallback newMeasurementsCallback = source->m_newMeasurementsCallback;
701 uint8_t dataPacketFlags;
702 int64_t frameLevelTimestamp = -1;
703 uint32_t offset = 0;
704 735
705 bool includeTime = info.IncludeTime;736 if (newMeasurementsCallback != nullptr)
737 {
738 SubscriptionInfo& info = source->m_subscriptionInfo;
739 uint8_t dataPacketFlags;
740 int64_t frameLevelTimestamp = -1;
741 uint32_t offset = 0;
706 742
707 // Read data packet flags743 bool includeTime = info.IncludeTime;
708 dataPacketFlags = buffer[offset];
709 offset++;
710 744
711 // Read frame-level timestamp, if available745 // Read data packet flags
712 if (dataPacketFlags & DataPacketFlags::Synchronized)746 dataPacketFlags = buffer[offset];
713 {747 offset++;
714 frameLevelTimestamp = source->m_endianConverter.ConvertBigEndian(*reinterpret_cast<const int64_t*>(&buffer[offset]));
715 offset += 8;
716 includeTime = false;
717 }
718 748
719 // Read measurement count and gather statistics749 // Read frame-level timestamp, if available
720 const uint32_t count = source->m_endianConverter.ConvertBigEndian(*reinterpret_cast<const uint32_t*>(&buffer[offset]));750 if (dataPacketFlags & DataPacketFlags::Synchronized)
721 source->m_totalMeasurementsReceived += count;751 {
722 offset += 4;752 frameLevelTimestamp = source->m_endianConverter.ConvertBigEndian(*reinterpret_cast<const int64_t*>(&buffer[offset]));
753 offset += 8;
754 includeTime = false;
755 }
723 756
724 const NewMeasurementsCallback newMeasurementsCallback = source->m_newMeasurementsCallback;757 // Read measurement count and gather statistics
725 vector<MeasurementPtr> measurements;758 const uint32_t count = source->m_endianConverter.ConvertBigEndian(*reinterpret_cast<const uint32_t*>(&buffer[offset]));
759 source->m_totalMeasurementsReceived += count;
760 offset += 4;
726 761
727 if (dataPacketFlags & DataPacketFlags::Compressed)762 vector<MeasurementPtr> measurements;
728 ParseTSSCMeasurements(source, buffer, offset, measurements);
729 else
730 ParseCompactMeasurements(source, buffer, offset, includeTime, info.UseMillisecondResolution, frameLevelTimestamp, measurements);
731 763
732 newMeasurementsCallback(source, measurements);764 if (dataPacketFlags & DataPacketFlags::Compressed)
765 ParseTSSCMeasurements(source, buffer, offset, measurements);
766 else
767 ParseCompactMeasurements(source, buffer, offset, includeTime, info.UseMillisecondResolution, frameLevelTimestamp, measurements);
768
769 newMeasurementsCallback(source, measurements);
770 }
733}771}
734 772
735void DataSubscriber::ParseTSSCMeasurements(DataSubscriber* source, const vector<uint8_t>& buffer, uint32_t offset, vector<MeasurementPtr>& measurements)773void DataSubscriber::ParseTSSCMeasurements(DataSubscriber* source, const vector<uint8_t>& buffer, uint32_t offset, vector<MeasurementPtr>& measurements)
736{774{
775 if (source == nullptr)
776 return;
777
737 MeasurementPtr measurement;778 MeasurementPtr measurement;
738 string errorMessage;779 string errorMessage;
739 780
@@ -837,6 +878,11 @@ void DataSubscriber::ParseTSSCMeasurements(DataSubscriber* source, const vector< · 97 unmodified lines
837 878
838void DataSubscriber::ParseCompactMeasurements(DataSubscriber* source, const vector<uint8_t>& buffer, uint32_t offset, bool includeTime, bool useMillisecondResolution, int64_t frameLevelTimestamp, vector<MeasurementPtr>& measurements)879void DataSubscriber::ParseCompactMeasurements(DataSubscriber* source, const vector<uint8_t>& buffer, uint32_t offset, bool includeTime, bool useMillisecondResolution, int64_t frameLevelTimestamp, vector<MeasurementPtr>& measurements)
839{880{
881 if (source == nullptr)
882 return;
883
884 const MessageCallback errorMessageCallback = source->m_errorMessageCallback;
885
840 MeasurementPtr measurement;886 MeasurementPtr measurement;
841 887
842 // Create measurement parser888 // Create measurement parser
@@ -848,9 +894,7 @@ void DataSubscriber::ParseCompactMeasurements(DataSubscriber* source, const vect · 5 unmodified lines
848 {894 {
849 if (!measurementParser.TryParseMeasurement(buffer, offset, length))895 if (!measurementParser.TryParseMeasurement(buffer, offset, length))
850 {896 {
851 const MessageCallback errorMessageCallback = source->m_errorMessageCallback;897 if (errorMessageCallback != nullptr)
852
853 if (errorMessageCallback)
854 errorMessageCallback(source, "Error parsing measurement");898 errorMessageCallback(source, "Error parsing measurement");
855 899
856 break;900 break;
@@ -868,6 +912,9 @@ void DataSubscriber::ParseCompactMeasurements(DataSubscriber* source, const vect · 11 unmodified lines
868// Dispatcher for processing complete message that is sent by the server at the end of a temporal session.912// Dispatcher for processing complete message that is sent by the server at the end of a temporal session.
869void DataSubscriber::ProcessingCompleteDispatcher(DataSubscriber* source, const vector<uint8_t>& buffer)913void DataSubscriber::ProcessingCompleteDispatcher(DataSubscriber* source, const vector<uint8_t>& buffer)
870{914{
915 if (source == nullptr)
916 return;
917
871 const MessageCallback processingCompleteCallback = source->m_processingCompleteCallback;918 const MessageCallback processingCompleteCallback = source->m_processingCompleteCallback;
872 919
873 if (processingCompleteCallback != nullptr)920 if (processingCompleteCallback != nullptr)
@@ -884,7 +931,13 @@ void DataSubscriber::ProcessingCompleteDispatcher(DataSubscriber* source, const · 10 unmodified lines
884// Dispatcher for processing complete message that is sent by the server at the end of a temporal session.931// Dispatcher for processing complete message that is sent by the server at the end of a temporal session.
885void DataSubscriber::ConfigurationChangedDispatcher(DataSubscriber* source, const vector<uint8_t>& buffer)932void DataSubscriber::ConfigurationChangedDispatcher(DataSubscriber* source, const vector<uint8_t>& buffer)
886{933{
887 source->m_configurationChangedCallback(source);934 if (source == nullptr)
935 return;
936
937 const ConfigurationChangedCallback configurationChangedCallback = source->m_configurationChangedCallback;
938
939 if (configurationChangedCallback != nullptr)
940 source->m_configurationChangedCallback(source);
888}941}
889 942
890// Dispatcher for connection terminated. This is called from its own separate thread943// Dispatcher for connection terminated. This is called from its own separate thread

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