From b6a79ec5ef673be62abc87dbde9dab54396e00fb Mon Sep 17 00:00:00 2001
From: Victor Lindquist <VLLindqvist@users.noreply.github.com>
Date: Wed, 15 Jun 2022 18:41:57 -0600
Subject: [PATCH] Refactored NetworkEngine and PointDataMessageHandler into
Data-Oriented Design to simplify development
---
modules/softwareintegration/CMakeLists.txt | 8 +-
...amessagehandler.cpp => messagehandler.cpp} | 578 ++++++++++--------
modules/softwareintegration/messagehandler.h | 50 ++
.../softwareintegration/network/network.cpp | 132 ++++
.../network/{networkengine.h => network.h} | 61 +-
.../network/networkengine.cpp | 245 --------
.../network/softwareconnection.cpp | 105 ++--
.../network/softwareconnection.h | 72 +--
.../pointdatamessagehandler.h | 106 ----
.../softwareintegrationmodule.cpp | 25 +-
.../softwareintegrationmodule.h | 6 +-
modules/softwareintegration/utils.cpp | 15 +
modules/softwareintegration/utils.h | 8 -
13 files changed, 672 insertions(+), 739 deletions(-)
rename modules/softwareintegration/{pointdatamessagehandler.cpp => messagehandler.cpp} (84%)
create mode 100644 modules/softwareintegration/messagehandler.h
create mode 100644 modules/softwareintegration/network/network.cpp
rename modules/softwareintegration/network/{networkengine.h => network.h} (69%)
delete mode 100644 modules/softwareintegration/network/networkengine.cpp
delete mode 100644 modules/softwareintegration/pointdatamessagehandler.h
diff --git a/modules/softwareintegration/CMakeLists.txt b/modules/softwareintegration/CMakeLists.txt
index d881cdff0..f6febd9ad 100644
--- a/modules/softwareintegration/CMakeLists.txt
+++ b/modules/softwareintegration/CMakeLists.txt
@@ -26,9 +26,9 @@ include(${OPENSPACE_CMAKE_EXT_DIR}/module_definition.cmake)
set(HEADER_FILES
${CMAKE_CURRENT_SOURCE_DIR}/softwareintegrationmodule.h
- ${CMAKE_CURRENT_SOURCE_DIR}/pointdatamessagehandler.h
+ ${CMAKE_CURRENT_SOURCE_DIR}/messagehandler.h
+ ${CMAKE_CURRENT_SOURCE_DIR}/network/network.h
${CMAKE_CURRENT_SOURCE_DIR}/network/softwareconnection.h
- ${CMAKE_CURRENT_SOURCE_DIR}/network/networkengine.h
${CMAKE_CURRENT_SOURCE_DIR}/rendering/renderablepointscloud.h
${CMAKE_CURRENT_SOURCE_DIR}/utils.h
${CMAKE_CURRENT_SOURCE_DIR}/syncablefloatdatastorage.h
@@ -41,9 +41,9 @@ source_group("Header Files" FILES ${HEADER_FILES})
set(SOURCE_FILES
${CMAKE_CURRENT_SOURCE_DIR}/softwareintegrationmodule.cpp
${CMAKE_CURRENT_SOURCE_DIR}/softwareintegrationmodule_lua.inl
- ${CMAKE_CURRENT_SOURCE_DIR}/pointdatamessagehandler.cpp
+ ${CMAKE_CURRENT_SOURCE_DIR}/messagehandler.cpp
+ ${CMAKE_CURRENT_SOURCE_DIR}/network/network.cpp
${CMAKE_CURRENT_SOURCE_DIR}/network/softwareconnection.cpp
- ${CMAKE_CURRENT_SOURCE_DIR}/network/networkengine.cpp
${CMAKE_CURRENT_SOURCE_DIR}/rendering/renderablepointscloud.cpp
${CMAKE_CURRENT_SOURCE_DIR}/utils.cpp
${CMAKE_CURRENT_SOURCE_DIR}/syncablefloatdatastorage.cpp
diff --git a/modules/softwareintegration/pointdatamessagehandler.cpp b/modules/softwareintegration/messagehandler.cpp
similarity index 84%
rename from modules/softwareintegration/pointdatamessagehandler.cpp
rename to modules/softwareintegration/messagehandler.cpp
index ec0ae675c..25f6cf64c 100644
--- a/modules/softwareintegration/pointdatamessagehandler.cpp
+++ b/modules/softwareintegration/messagehandler.cpp
@@ -22,7 +22,7 @@
* OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. *
****************************************************************************************/
-#include <modules/softwareintegration/pointdatamessagehandler.h>
+#include <modules/softwareintegration/messagehandler.h>
#include <modules/softwareintegration/softwareintegrationmodule.h>
#include <modules/softwareintegration/utils.h>
@@ -41,14 +41,219 @@
#include <iomanip>
namespace {
- constexpr const char* _loggerCat = "PDatMessHand";
+
+constexpr const char* _loggerCat = "PDatMessHand";
+
} // namespace
-namespace openspace {
+namespace openspace::softwareintegration::network {
+
+// Anonymous namespace
+namespace {
+
+CallbackMap callbacks{};
+std::mutex callbacksMutex{};
+size_t callbacksRetries{0};
+
+const Renderable* getRenderable(const std::string& identifier) {
+ return renderable(identifier);
+}
+
+void checkRenderable(
+ const std::vector<char>& message, size_t& messageOffset,
+ std::shared_ptr<SoftwareConnection> connection, std::string& identifier
+) {
+ std::string guiName;
+
+ try {
+ // The following order of creating variables is the exact order they are received
+ // in the message. If the order is not the same, the global variable
+ // 'message offset' will be wrong
+ identifier = simp::readString(message, messageOffset);
+ guiName = simp::readString(message, messageOffset);
+ }
+ catch (const simp::SimpError& err) {
+ LERROR(fmt::format("Error when reading identifier and guiName from message: {}", err.message));
+ return;
+ }
+
+ connection->addSceneGraphNode(identifier);
+
+ auto r = renderable(identifier);
+ bool hasCallbacks = false;
+ {
+ std::lock_guard guard(callbacksMutex);
+ hasCallbacks = callbacks.count(identifier) > 0;
+ }
+ if (!r && !hasCallbacks) {
+ LDEBUG(fmt::format("No renderable with identifier '{}' was found. Creating it.", identifier));
+
+ // Create a renderable, since it didn't exist
+ using namespace std::string_literals;
+ ghoul::Dictionary renderablePointsCloud;
+ renderablePointsCloud.setValue("Type", "RenderablePointsCloud"s);
+ renderablePointsCloud.setValue("Identifier", identifier);
+ renderablePointsCloud.setValue("Name", guiName);
+
+ ghoul::Dictionary gui;
+ gui.setValue("Name", guiName);
+ gui.setValue("Path", "/Software Integration"s);
+
+ ghoul::Dictionary node;
+ node.setValue("Identifier", identifier);
+ node.setValue("Renderable", renderablePointsCloud);
+ node.setValue("GUI", gui);
+
+ global::scriptEngine->queueScript(
+ "openspace.addSceneGraphNode(" + ghoul::formatLua(node) + ")"
+ "openspace.setPropertyValueSingle('Modules.CefWebGui.Reload', nil)", // Reload WebGUI so that SoftwareIntegration GUI appears
+ scripting::ScriptEngine::RemoteScripting::Yes
+ );
+ }
+}
-using namespace softwareintegration;
+void addCallback(
+ const std::string& identifier,
+ const Callback& newCallback
+) {
+ std::lock_guard guard(callbacksMutex);
+ auto it = callbacks.find(identifier);
+ if (it == callbacks.end()) {
+ CallbackList newCallbackList{ newCallback };
+ callbacks.emplace(identifier, newCallbackList);
+ }
+ else {
+ it->second.push_back(newCallback);
+ }
+}
-void PointDataMessageHandler::handlePointDataMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection) {
+void onFixedColorChange(
+ properties::Property* property,
+ const std::string& identifier,
+ std::shared_ptr<SoftwareConnection> connection
+) {
+ if (!connection->isConnected()) {
+ connection->removePropertySubscription(property->identifier(), identifier);
+ return;
+ }
+
+ // auto propertySubscription = connection->getPropertySubscription(identifier, property->identifier());
+ // if (!propertySubscription) return;
+ // if (!propertySubscription->shouldSendMessage) {
+ // propertySubscription->shouldSendMessage = true;
+ // return;
+ // }
+
+ glm::vec4 color = std::any_cast<glm::vec4>(property->get());
+
+ const std::string message = simp::formatColorMessage(identifier, color);
+ connection->sendMessage(message);
+}
+
+void onOpacityChange(
+ properties::Property* property,
+ const std::string& identifier,
+ std::shared_ptr<SoftwareConnection> connection
+) {
+ if (!connection->isConnected()) {
+ connection->removePropertySubscription(property->identifier(), identifier);
+ return;
+ }
+
+ // auto propertySubscription = connection->getPropertySubscription(identifier, property->identifier());
+ // if (!propertySubscription) return;
+ // if (!propertySubscription->shouldSendMessage) {
+ // propertySubscription->shouldSendMessage = true;
+ // return;
+ // }
+
+ float value = std::any_cast<float>(property->get());
+ std::string hex_value = simp::floatToHex(value);
+
+ const std::string message = simp::formatUpdateMessage(simp::MessageType::Opacity, identifier, hex_value);
+ connection->sendMessage(message);
+}
+
+void onFixedPointSizeChange(
+ properties::Property* property,
+ const std::string& identifier,
+ std::shared_ptr<SoftwareConnection> connection
+) {
+ if (!connection->isConnected()) {
+ connection->removePropertySubscription(property->identifier(), identifier);
+ return;
+ }
+
+ // auto propertySubscription = connection->getPropertySubscription(identifier, property->identifier());
+ // if (!propertySubscription) return;
+ // if (!propertySubscription->shouldSendMessage) {
+ // propertySubscription->shouldSendMessage = true;
+ // return;
+ // }
+
+ float value = std::any_cast<float>(property->get());
+ std::string hex_value = simp::floatToHex(value);
+
+ const std::string message = simp::formatUpdateMessage(simp::MessageType::FixedSize, identifier, hex_value);
+ connection->sendMessage(message);
+}
+
+void onVisibilityChange(
+ properties::Property* property,
+ const std::string& identifier,
+ std::shared_ptr<SoftwareConnection> connection
+) {
+ if (!connection->isConnected()) {
+ connection->removePropertySubscription(property->identifier(), identifier);
+ return;
+ }
+
+ // auto propertySubscription = connection->getPropertySubscription(identifier, property->identifier());
+ // if (!propertySubscription) return;
+ // if (!propertySubscription->shouldSendMessage) {
+ // propertySubscription->shouldSendMessage = true;
+ // return;
+ // }
+
+ bool isVisible = std::any_cast<bool>(property->get());
+ std::string_view visibilityFlag = isVisible ? "T" : "F";
+
+ const std::string message = simp::formatUpdateMessage(simp::MessageType::Visibility, identifier, visibilityFlag);
+ connection->sendMessage(message);
+}
+
+// TODO: Move to SIMP / use distanceconversion
+void convertToMeterPerSecond(simp::LengthUnit currLengthUnit, std::vector<float>& data) {
+ // distanceconversion::convertDistance
+ float multiplier = 1.0;
+ switch (currLengthUnit) {
+ case simp::LengthUnit::km:
+ multiplier = 1000.0;
+ break;
+ case simp::LengthUnit::AU:
+ multiplier = static_cast<float>(distanceconstants::AstronomicalUnit);
+ break;
+ case simp::LengthUnit::lyr:
+ multiplier = static_cast<float>(distanceconstants::LightYear);
+ break;
+ case simp::LengthUnit::pc:
+ multiplier = static_cast<float>(distanceconstants::Parsec);
+ break;
+ case simp::LengthUnit::kpc:
+ multiplier = static_cast<float>(distanceconstants::Parsec * 1.0e3);
+ break;
+ case simp::LengthUnit::Mpc:
+ multiplier = static_cast<float>(distanceconstants::Parsec * 1.0e6);
+ break;
+ default:
+ break;
+ }
+
+ std::transform(std::begin(data), std::end(data), std::begin(data),
+ [multiplier] (float f) {return f * multiplier;});
+}
+
+void handlePointDataMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection) {
size_t messageOffset = 0;
std::string identifier;
@@ -86,8 +291,8 @@ void PointDataMessageHandler::handlePointDataMessage(const std::vector<char>& me
addCallback(identifier, { reanchorCallback, { storage::Key::DataPoints }, "reanchorCallback" });
}
-void PointDataMessageHandler::handleVelocityDataMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection) {
- LWARNING(fmt::format("PointDataMessageHandler::handleVelocityDataMessage()"));
+void handleVelocityDataMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection) {
+ LWARNING(fmt::format("handleVelocityDataMessage()"));
size_t messageOffset = 0;
std::string identifier;
@@ -139,10 +344,11 @@ void PointDataMessageHandler::handleVelocityDataMessage(const std::vector<char>&
auto module = global::moduleEngine->module<SoftwareIntegrationModule>();
module->storeData(identifier, storage::Key::VelocityData, std::move(velocities));
- auto velNaNModeCallback = [this, identifier, velNaNMode, velNaNColor, connection] {
+ auto velNaNModeCallback = [identifier, velNaNMode, velNaNColor, connection] {
if (velNaNMode == simp::NaNRenderMode::Color) {
// Get renderable
auto r = getRenderable(identifier);
+ if (!r) return;
// Get velNaNColor of renderable
properties::Property* velNaNColorProperty = r->property("VelNaNColor");
@@ -182,7 +388,7 @@ void PointDataMessageHandler::handleVelocityDataMessage(const std::vector<char>&
addCallback(identifier, { enableMotionCallback, { storage::Key::VelocityData }, "Enable motion mode, wait for VelocityData" });
}
-void PointDataMessageHandler::handleFixedColorMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection) {
+void handleFixedColorMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection) {
size_t messageOffset = 0;
std::string identifier;
@@ -207,7 +413,7 @@ void PointDataMessageHandler::handleFixedColorMessage(const std::vector<char>& m
// Create weak_ptr, safer than shared_ptr for lambdas
std::weak_ptr<SoftwareConnection> connWeakPtr{ connection };
- auto setFixedColorCallback = [this, identifier, color, colorProperty, connWeakPtr] {
+ auto setFixedColorCallback = [identifier, color, colorProperty, connWeakPtr] {
if (!colorProperty || connWeakPtr.expired()) return;
// auto conn = connWeakPtr.lock()
@@ -239,14 +445,14 @@ void PointDataMessageHandler::handleFixedColorMessage(const std::vector<char>& m
addCallback(identifier, { setFixedColorCallback, {}, "handleFixedColorMessage" });
// Create and set onChange for color
- auto updateColor = [this, colorProperty, identifier, connWeakPtr] {
+ auto updateColor = [colorProperty, identifier, connWeakPtr] {
if (!colorProperty || connWeakPtr.expired()) return;
onFixedColorChange(colorProperty, identifier, connWeakPtr.lock());
};
connection->addPropertySubscription(colorProperty->identifier(), identifier, updateColor);
}
-void PointDataMessageHandler::handleColormapMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection) {
+void handleColormapMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection) {
size_t messageOffset = 0;
std::string identifier;
@@ -285,9 +491,10 @@ void PointDataMessageHandler::handleColormapMessage(const std::vector<char>& mes
auto module = global::moduleEngine->module<SoftwareIntegrationModule>();
module->storeData(identifier, storage::Key::Colormap, std::move(colorMap));
- auto colormapLimitsCallback = [this, identifier, min, max, connection] {
+ auto colormapLimitsCallback = [identifier, min, max, connection] {
// Get renderable
auto r = getRenderable(identifier);
+ if (!r) return;
properties::Property* colormapMinProperty = r->property("ColormapMin");
// auto minPropertySub = connection->getPropertySubscription(identifier, colormapMinProperty->identifier());
@@ -324,10 +531,11 @@ void PointDataMessageHandler::handleColormapMessage(const std::vector<char>& mes
};
addCallback(identifier, { colormapLimitsCallback, {}, "colormapLimitsCallback" });
- auto cmapNaNModeCallback = [this, identifier, cmapNaNMode, cmapNaNColor, connection] {
+ auto cmapNaNModeCallback = [identifier, cmapNaNMode, cmapNaNColor, connection] {
if (cmapNaNMode == simp::NaNRenderMode::Color) {
// Get renderable
auto r = getRenderable(identifier);
+ if (!r) return;
// Get cmapNaNColor of renderable
properties::Property* cmapNaNColorProperty = r->property("CmapNaNColor");
@@ -355,7 +563,7 @@ void PointDataMessageHandler::handleColormapMessage(const std::vector<char>& mes
};
addCallback(identifier, { cmapNaNModeCallback, {}, "cmapNaNModeCallback" });
- auto enableColormapCallback = [this, identifier] {
+ auto enableColormapCallback = [identifier] {
global::scriptEngine->queueScript(
fmt::format(
"openspace.setPropertyValueSingle('Scene.{}.Renderable.ColormapEnabled', {});",
@@ -369,7 +577,7 @@ void PointDataMessageHandler::handleColormapMessage(const std::vector<char>& mes
addCallback(identifier, { enableColormapCallback, std::move(dataToWaitFor), "enableColormapCallback" });
}
-void PointDataMessageHandler::handleAttributeDataMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection) {
+void handleAttributeDataMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection) {
size_t messageOffset = 0;
std::string identifier;
@@ -409,7 +617,7 @@ void PointDataMessageHandler::handleAttributeDataMessage(const std::vector<char>
std::string callbackDescription = "handleAttributeDataMessage, key=" + storage::getStorageKeyString(key);
switch (key) {
case storage::Key::ColormapAttrData : {
- auto callback = [this, identifier] {
+ auto callback = [identifier] {
global::scriptEngine->queueScript(
fmt::format(
"openspace.setPropertyValueSingle('Scene.{}.Renderable.ColormapEnabled', {});",
@@ -422,7 +630,7 @@ void PointDataMessageHandler::handleAttributeDataMessage(const std::vector<char>
break;
}
case storage::Key::LinearSizeAttrData: {
- auto callback = [this, identifier] {
+ auto callback = [identifier] {
global::scriptEngine->queueScript(
fmt::format(
"openspace.setPropertyValueSingle('Scene.{}.Renderable.LinearSizeEnabled', {});",
@@ -439,7 +647,7 @@ void PointDataMessageHandler::handleAttributeDataMessage(const std::vector<char>
}
}
-void PointDataMessageHandler::handleOpacityMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection) {
+void handleOpacityMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection) {
size_t messageOffset = 0;
std::string identifier;
@@ -464,7 +672,7 @@ void PointDataMessageHandler::handleOpacityMessage(const std::vector<char>& mess
// Create weak_ptr, safer than shared_ptr for lambdas
std::weak_ptr<SoftwareConnection> connWeakPtr{ connection };
- auto callback = [this, identifier, opacity, opacityProperty, connWeakPtr] {
+ auto callback = [identifier, opacity, opacityProperty, connWeakPtr] {
if (!opacityProperty || connWeakPtr.expired()) return;
// auto conn = connWeakPtr.lock()
@@ -490,14 +698,14 @@ void PointDataMessageHandler::handleOpacityMessage(const std::vector<char>& mess
addCallback(identifier, { callback, {}, "handleOpacityMessage" });
// Create and set onChange for opacity
- auto updateOpacity = [this, opacityProperty, identifier, connWeakPtr] {
+ auto updateOpacity = [opacityProperty, identifier, connWeakPtr] {
if (!opacityProperty || connWeakPtr.expired()) return;
onOpacityChange(opacityProperty, identifier, connWeakPtr.lock());
};
connection->addPropertySubscription(opacityProperty->identifier(), identifier, updateOpacity);
}
-void PointDataMessageHandler::handleFixedPointSizeMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection) {
+void handleFixedPointSizeMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection) {
size_t messageOffset = 0;
std::string identifier;
@@ -522,7 +730,7 @@ void PointDataMessageHandler::handleFixedPointSizeMessage(const std::vector<char
// Create weak_ptr, safer than shared_ptr for lambdas
std::weak_ptr<SoftwareConnection> connWeakPtr{ connection };
- auto callback = [this, identifier, size, sizeProperty, connWeakPtr] {
+ auto callback = [identifier, size, sizeProperty, connWeakPtr] {
if (!sizeProperty || connWeakPtr.expired()) return;
// auto conn = connWeakPtr.lock()
@@ -556,14 +764,14 @@ void PointDataMessageHandler::handleFixedPointSizeMessage(const std::vector<char
};
addCallback(identifier, { callback, {}, "handleFixedPointSizeMessage" });
- auto updateSize = [this, sizeProperty, identifier, connWeakPtr] {
+ auto updateSize = [sizeProperty, identifier, connWeakPtr] {
if (!sizeProperty || connWeakPtr.expired()) return;
onFixedPointSizeChange(sizeProperty, identifier, connWeakPtr.lock());
};
connection->addPropertySubscription(sizeProperty->identifier(), identifier, updateSize);
}
-void PointDataMessageHandler::handleLinearPointSizeMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection) {
+void handleLinearPointSizeMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection) {
size_t messageOffset = 0;
std::string identifier;
@@ -582,9 +790,10 @@ void PointDataMessageHandler::handleLinearPointSizeMessage(const std::vector<cha
return;
}
- auto linearSizeCallback = [this, identifier, size, min, max] {
+ auto linearSizeCallback = [identifier, size, min, max] {
// Get renderable
auto r = getRenderable(identifier);
+ if (!r) return;
// Get size from renderable
properties::Property* sizeProperty = r->property("Size");
@@ -628,7 +837,7 @@ void PointDataMessageHandler::handleLinearPointSizeMessage(const std::vector<cha
};
addCallback(identifier, { linearSizeCallback, {}, "linearSizeCallback" });
- auto enableLinearSizeCallback = [this, identifier] {
+ auto enableLinearSizeCallback = [identifier] {
global::scriptEngine->queueScript(
fmt::format(
"openspace.setPropertyValueSingle('Scene.{}.Renderable.LinearSizeEnabled', {});",
@@ -647,7 +856,7 @@ void PointDataMessageHandler::handleLinearPointSizeMessage(const std::vector<cha
);
}
-void PointDataMessageHandler::handleVisibilityMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection) {
+void handleVisibilityMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection) {
size_t messageOffset = 0;
std::string identifier;
@@ -674,7 +883,7 @@ void PointDataMessageHandler::handleVisibilityMessage(const std::vector<char>& m
const bool visibility = visibilityMessage == "T";
- auto callback = [this, identifier, visibility, visibilityProperty, connWeakPtr] {
+ auto callback = [identifier, visibility, visibilityProperty, connWeakPtr] {
if (!visibilityProperty || connWeakPtr.expired()) return;
// auto conn = connWeakPtr.lock()
@@ -702,14 +911,14 @@ void PointDataMessageHandler::handleVisibilityMessage(const std::vector<char>& m
addCallback(identifier, { callback, {}, "handleVisibilityMessage" });
// Create and set onChange for visibility
- auto toggleVisibility = [this, visibilityProperty, identifier, connWeakPtr] {
+ auto toggleVisibility = [visibilityProperty, identifier, connWeakPtr] {
if (!visibilityProperty || connWeakPtr.expired()) return;
onVisibilityChange(visibilityProperty, identifier, connWeakPtr.lock());
};
connection->addPropertySubscription(visibilityProperty->identifier(), identifier, toggleVisibility);
}
-void PointDataMessageHandler::handleRemoveSGNMessage(const std::vector<char>& message,std::shared_ptr<SoftwareConnection> connection) {
+void handleRemoveSGNMessage(const std::vector<char>& message,std::shared_ptr<SoftwareConnection> connection) {
size_t messageOffset = 0;
std::string identifier;
@@ -742,12 +951,95 @@ void PointDataMessageHandler::handleRemoveSGNMessage(const std::vector<char>& me
LDEBUG(fmt::format("Scene graph node '{}' removed.", identifier));
}
-void PointDataMessageHandler::postSync() {
- std::lock_guard guard(_onceNodeExistsCallbacksMutex);
+} // namespace
+
+void handleMessage(IncomingMessage& incomingMessage) {
+ if(incomingMessage.connection.expired()) {
+ LDEBUG(fmt::format("Trying to handle message from disconnected peer. Aborting."));
+ return;
+ }
+
+ auto connectionPtr = incomingMessage.connection.lock();
+
+ const simp::MessageType messageType = incomingMessage.type;
+ std::vector<char>& message = incomingMessage.content;
+
+ switch (messageType) {
+ case simp::MessageType::Connection: {
+ LDEBUG(fmt::format("Message recieved... Connection: {}", connectionPtr->id()));
+ size_t offset = 0;
+ const std::string software = simp::readString(message, offset);
+
+ // Send back message to software to complete handshake
+ connectionPtr->sendMessage(simp::formatConnectionMessage(software));
+ LINFO(fmt::format("OpenSpace has connected with {} through socket", software));
+ break;
+ }
+ case simp::MessageType::PointData: {
+ LDEBUG("Message recieved.. Point data");
+ handlePointDataMessage(message, connectionPtr);
+ break;
+ }
+ case simp::MessageType::VelocityData: {
+ LDEBUG("Message recieved... Velocity data");
+ handleVelocityDataMessage(message, connectionPtr);
+ break;
+ }
+ case simp::MessageType::RemoveSceneGraphNode: {
+ LDEBUG(fmt::format("Message recieved.. Remove SGN"));
+ handleRemoveSGNMessage(message, connectionPtr);
+ break;
+ }
+ case simp::MessageType::Color: {
+ LDEBUG(fmt::format("Message recieved.. New color"));
+ handleFixedColorMessage(message, connectionPtr);
+ break;
+ }
+ case simp::MessageType::Colormap: {
+ LDEBUG(fmt::format("Message recieved.. New colormap"));
+ handleColormapMessage(message, connectionPtr);
+ break;
+ }
+ case simp::MessageType::AttributeData: {
+ LDEBUG(fmt::format("Message recieved.. New attribute data"));
+ handleAttributeDataMessage(message, connectionPtr);
+ break;
+ }
+ case simp::MessageType::Opacity: {
+ LDEBUG(fmt::format("Message recieved.. New Opacity"));
+ handleOpacityMessage(message, connectionPtr);
+ break;
+ }
+ case simp::MessageType::FixedSize: {
+ LDEBUG(fmt::format("Message recieved.. New size"));
+ handleFixedPointSizeMessage(message, connectionPtr);
+ break;
+ }
+ case simp::MessageType::LinearSize: {
+ LDEBUG(fmt::format("Message recieved.. New linear size"));
+ handleLinearPointSizeMessage(message, connectionPtr);
+ break;
+ }
+ case simp::MessageType::Visibility: {
+ LDEBUG(fmt::format("Message recieved.. New visibility"));
+ handleVisibilityMessage(message, connectionPtr);
+ break;
+ }
+ default: {
+ LERROR(fmt::format(
+ "Unsupported message type: {}", incomingMessage.rawMessageType
+ ));
+ break;
+ }
+ }
+}
+
+void postSyncCallbacks() {
+ std::lock_guard guard(callbacksMutex);
// Check if the scene graph node has been created.
// If so, call the corresponding callback functions to set up any subscriptions
- auto callbackMapIt = _onceNodeExistsCallbacks.begin();
- while (callbackMapIt != _onceNodeExistsCallbacks.end()) {
+ auto callbackMapIt = callbacks.begin();
+ while (callbackMapIt != callbacks.end()) {
auto& [identifier, callbackList] = *callbackMapIt;
try {
@@ -779,225 +1071,19 @@ void PointDataMessageHandler::postSync() {
}
if (callbackList.empty()) {
- callbackMapIt = _onceNodeExistsCallbacks.erase(callbackMapIt);
- _onceNodeExistsCallbacksRetries = 0;
+ callbackMapIt = callbacks.erase(callbackMapIt);
+ callbacksRetries = 0;
} else {
callbackMapIt++;
}
}
catch(std::exception &err) {
- ++_onceNodeExistsCallbacksRetries;
- ghoul_assert(_onceNodeExistsCallbacksRetries < 10, "Too many callback retries");
+ ++callbacksRetries;
+ ghoul_assert(callbacksRetries < 10, "Too many callback retries");
LDEBUG(fmt::format("Error when trying to run callback: {}", err.what()));
break;
}
}
}
-const Renderable* PointDataMessageHandler::getRenderable(const std::string& identifier) {
- return renderable(identifier);
-}
-
-void PointDataMessageHandler::checkRenderable(
- const std::vector<char>& message, size_t& messageOffset,
- std::shared_ptr<SoftwareConnection> connection, std::string& identifier
-) {
- std::string guiName;
-
- try {
- // The following order of creating variables is the exact order they are received
- // in the message. If the order is not the same, the global variable
- // 'message offset' will be wrong
- identifier = simp::readString(message, messageOffset);
- guiName = simp::readString(message, messageOffset);
- }
- catch (const simp::SimpError& err) {
- LERROR(fmt::format("Error when reading identifier and guiName from message: {}", err.message));
- return;
- }
-
- connection->addSceneGraphNode(identifier);
-
- const Renderable* r = renderable(identifier);
- bool hasCallbacks = false;
- {
- std::lock_guard guard(_onceNodeExistsCallbacksMutex);
- hasCallbacks = _onceNodeExistsCallbacks.count(identifier) > 0;
- }
- if (!r && !hasCallbacks) {
- LDEBUG(fmt::format("No renderable with identifier '{}' was found. Creating it.", identifier));
-
- // Create a renderable, since it didn't exist
- using namespace std::string_literals;
- ghoul::Dictionary renderablePointsCloud;
- renderablePointsCloud.setValue("Type", "RenderablePointsCloud"s);
- renderablePointsCloud.setValue("Identifier", identifier);
- renderablePointsCloud.setValue("Name", guiName);
-
- ghoul::Dictionary gui;
- gui.setValue("Name", guiName);
- gui.setValue("Path", "/Software Integration"s);
-
- ghoul::Dictionary node;
- node.setValue("Identifier", identifier);
- node.setValue("Renderable", renderablePointsCloud);
- node.setValue("GUI", gui);
-
- global::scriptEngine->queueScript(
- "openspace.addSceneGraphNode(" + ghoul::formatLua(node) + ")"
- "openspace.setPropertyValueSingle('Modules.CefWebGui.Reload', nil)", // Reload WebGUI so that SoftwareIntegration GUI appears
- scripting::ScriptEngine::RemoteScripting::Yes
- );
- }
-}
-
-void PointDataMessageHandler::addCallback(
- const std::string& identifier,
- const Callback& newCallback
-) {
- std::lock_guard guard(_onceNodeExistsCallbacksMutex);
- auto it = _onceNodeExistsCallbacks.find(identifier);
- if (it == _onceNodeExistsCallbacks.end()) {
- CallbackList newCallbackList{ newCallback };
- _onceNodeExistsCallbacks.emplace(identifier, newCallbackList);
- }
- else {
- it->second.push_back(newCallback);
- }
-}
-
-void PointDataMessageHandler::onFixedColorChange(
- properties::Property* property,
- const std::string& identifier,
- std::shared_ptr<SoftwareConnection> connection
-) {
- if (!connection->isConnected()) {
- SoftwareConnection::PointDataMessageHandlerFriends::removePropertySubscription(
- connection, property->identifier(), identifier
- );
- return;
- }
-
- // auto propertySubscription = connection->getPropertySubscription(identifier, property->identifier());
- // if (!propertySubscription) return;
- // if (!propertySubscription->shouldSendMessage) {
- // propertySubscription->shouldSendMessage = true;
- // return;
- // }
-
- glm::vec4 color = std::any_cast<glm::vec4>(property->get());
-
- const std::string message = simp::formatColorMessage(identifier, color);
- connection->sendMessage(message);
-}
-
-void PointDataMessageHandler::onOpacityChange(
- properties::Property* property,
- const std::string& identifier,
- std::shared_ptr<SoftwareConnection> connection
-) {
- if (!connection->isConnected()) {
- SoftwareConnection::PointDataMessageHandlerFriends::removePropertySubscription(
- connection, property->identifier(), identifier
- );
- return;
- }
-
- // auto propertySubscription = connection->getPropertySubscription(identifier, property->identifier());
- // if (!propertySubscription) return;
- // if (!propertySubscription->shouldSendMessage) {
- // propertySubscription->shouldSendMessage = true;
- // return;
- // }
-
- float value = std::any_cast<float>(property->get());
- std::string hex_value = simp::floatToHex(value);
-
- const std::string message = simp::formatUpdateMessage(simp::MessageType::Opacity, identifier, hex_value);
- connection->sendMessage(message);
-}
-
-void PointDataMessageHandler::onFixedPointSizeChange(
- properties::Property* property,
- const std::string& identifier,
- std::shared_ptr<SoftwareConnection> connection
-) {
- if (!connection->isConnected()) {
- SoftwareConnection::PointDataMessageHandlerFriends::removePropertySubscription(
- connection, property->identifier(), identifier
- );
- return;
- }
-
- // auto propertySubscription = connection->getPropertySubscription(identifier, property->identifier());
- // if (!propertySubscription) return;
- // if (!propertySubscription->shouldSendMessage) {
- // propertySubscription->shouldSendMessage = true;
- // return;
- // }
-
- float value = std::any_cast<float>(property->get());
- std::string hex_value = simp::floatToHex(value);
-
- const std::string message = simp::formatUpdateMessage(simp::MessageType::FixedSize, identifier, hex_value);
- connection->sendMessage(message);
-}
-
-void PointDataMessageHandler::onVisibilityChange(
- properties::Property* property,
- const std::string& identifier,
- std::shared_ptr<SoftwareConnection> connection
-) {
- if (!connection->isConnected()) {
- SoftwareConnection::PointDataMessageHandlerFriends::removePropertySubscription(
- connection, property->identifier(), identifier
- );
- return;
- }
-
- // auto propertySubscription = connection->getPropertySubscription(identifier, property->identifier());
- // if (!propertySubscription) return;
- // if (!propertySubscription->shouldSendMessage) {
- // propertySubscription->shouldSendMessage = true;
- // return;
- // }
-
- bool isVisible = std::any_cast<bool>(property->get());
- std::string_view visibilityFlag = isVisible ? "T" : "F";
-
- const std::string message = simp::formatUpdateMessage(simp::MessageType::Visibility, identifier, visibilityFlag);
- connection->sendMessage(message);
-}
-
-// TODO: Move to SIMP / use distanceconversion
-void PointDataMessageHandler::convertToMeterPerSecond(simp::LengthUnit currLengthUnit, std::vector<float>& data) {
- // distanceconversion::convertDistance
- float multiplier = 1.0;
- switch (currLengthUnit) {
- case simp::LengthUnit::km:
- multiplier = 1000.0;
- break;
- case simp::LengthUnit::AU:
- multiplier = static_cast<float>(distanceconstants::AstronomicalUnit);
- break;
- case simp::LengthUnit::lyr:
- multiplier = static_cast<float>(distanceconstants::LightYear);
- break;
- case simp::LengthUnit::pc:
- multiplier = static_cast<float>(distanceconstants::Parsec);
- break;
- case simp::LengthUnit::kpc:
- multiplier = static_cast<float>(distanceconstants::Parsec * 1.0e3);
- break;
- case simp::LengthUnit::Mpc:
- multiplier = static_cast<float>(distanceconstants::Parsec * 1.0e6);
- break;
- default:
- break;
- }
-
- std::transform(std::begin(data), std::end(data), std::begin(data),
- [multiplier] (float f) {return f * multiplier;});
-}
-
-} // namespace openspace
+} // namespace openspace::softwareintegration::messagehandler
diff --git a/modules/softwareintegration/messagehandler.h b/modules/softwareintegration/messagehandler.h
new file mode 100644
index 000000000..47daed4bb
--- /dev/null
+++ b/modules/softwareintegration/messagehandler.h
@@ -0,0 +1,50 @@
+/*****************************************************************************************
+ * *
+ * OpenSpace *
+ * *
+ * Copyright (c) 2014-2022 *
+ * *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of this *
+ * software and associated documentation files (the "Software"), to deal in the Software *
+ * without restriction, including without limitation the rights to use, copy, modify, *
+ * merge, publish, distribute, sublicense, and/or sell copies of the Software, and to *
+ * permit persons to whom the Software is furnished to do so, subject to the following *
+ * conditions: *
+ * *
+ * The above copyright notice and this permission notice shall be included in all copies *
+ * or substantial portions of the Software. *
+ * *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, *
+ * INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A *
+ * PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT *
+ * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF *
+ * CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE *
+ * OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. *
+ ****************************************************************************************/
+
+#ifndef __OPENSPACE_MODULE_SOFTWAREINTEGRATION___MESSAGEHANDLER___H__
+#define __OPENSPACE_MODULE_SOFTWAREINTEGRATION___MESSAGEHANDLER___H__
+
+#include <unordered_map>
+
+#include <openspace/properties/propertyowner.h>
+
+#include <modules/softwareintegration/network/softwareconnection.h>
+
+namespace openspace::softwareintegration::network {
+
+struct Callback {
+ std::function<void()> function;
+ std::vector<softwareintegration::storage::Key> waitForData = {};
+ std::string description = "???"; // To help debugging. Maybe remove?
+};
+using CallbackList = std::vector<Callback>;
+using CallbackMap = std::unordered_map<std::string, CallbackList>;
+
+void postSyncCallbacks();
+
+void handleMessage(IncomingMessage& incomingMessage);
+
+} // namespace openspace::softwareintegration::messagehandler
+
+#endif // __OPENSPACE_MODULE_SOFTWAREINTEGRATION___MESSAGEHANDLER___H__
diff --git a/modules/softwareintegration/network/network.cpp b/modules/softwareintegration/network/network.cpp
new file mode 100644
index 000000000..84645866f
--- /dev/null
+++ b/modules/softwareintegration/network/network.cpp
@@ -0,0 +1,132 @@
+/*****************************************************************************************
+ * *
+ * OpenSpace *
+ * *
+ * Copyright (c) 2014-2022 *
+ * *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of this *
+ * software and associated documentation files (the "Software"), to deal in the Software *
+ * without restriction, including without limitation the rights to use, copy, modify, *
+ * merge, publish, distribute, sublicense, and/or sell copies of the Software, and to *
+ * permit persons to whom the Software is furnished to do so, subject to the following *
+ * conditions: *
+ * *
+ * The above copyright notice and this permission notice shall be included in all copies *
+ * or substantial portions of the Software. *
+ * *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, *
+ * INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A *
+ * PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT *
+ * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF *
+ * CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE *
+ * OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. *
+ ****************************************************************************************/
+
+#include <modules/softwareintegration/network/network.h>
+
+#include <modules/softwareintegration/messagehandler.h>
+#include <modules/softwareintegration/utils.h>
+#include <openspace/engine/globals.h>
+#include <openspace/engine/globalscallbacks.h>
+#include <openspace/scene/scene.h>
+#include <openspace/scripting/scriptengine.h>
+#include <ghoul/logging/logmanager.h>
+
+
+namespace {
+
+ constexpr const char* _loggerCat = "NetworkEngine";
+
+} // namespace
+
+namespace openspace::softwareintegration::network {
+
+namespace {
+
+void eventLoop(std::weak_ptr<NetworkState> networkStateWeakPtr) {
+ while (!networkStateWeakPtr.expired()) {
+ auto networkState = networkStateWeakPtr.lock();
+ if (networkState->shouldStopThreads) break;
+ // The call to "pop" below will block execution
+ // on this thread until interrupt is called
+ try {
+ auto pm = networkState->incomingMessages.pop();
+ handleMessage(pm);
+ }
+ catch (const ghoul::RuntimeError&) {
+ break;
+ }
+ }
+}
+
+void serverLoop(std::weak_ptr<NetworkState> networkStateWeakPtr) {
+ while (!networkStateWeakPtr.expired()) {
+ auto networkState = networkStateWeakPtr.lock();
+ if (networkState->shouldStopThreads) break;
+ std::unique_ptr<ghoul::io::TcpSocket> socket = networkState->server.awaitPendingTcpSocket();
+
+ if (!socket) return;
+
+ socket->startStreams();
+
+ auto p = std::make_shared<SoftwareConnection>(std::move(socket));
+ std::lock_guard guard(networkState->softwareConnectionsMutex);
+ auto [it, peerInserted] = networkState->softwareConnections.emplace(p->id(), std::move(p));
+
+ if (peerInserted) {
+ auto connectionWeak = std::weak_ptr<SoftwareConnection>{ it->second };
+ auto thread = std::thread{
+ [connectionWeak, networkStateWeakPtr] {
+ connection::eventLoop(connectionWeak, networkStateWeakPtr);
+ }
+ };
+ it->second->setThread(thread);
+ }
+ }
+}
+
+} // namespace
+
+std::shared_ptr<NetworkState> serve(const int port) {
+ auto networkState = std::make_shared<NetworkState>();
+
+ // 4700, is the defualt port where the tcp socket will be opened to the ext. software
+ networkState->server.listen(port);
+
+ std::weak_ptr<NetworkState> networkStateWeakPtr = networkState;
+ networkState->serverThread = std::thread{ [networkStateWeakPtr] {
+ serverLoop(networkStateWeakPtr);
+ } };
+
+ networkState->eventLoopThread = std::thread{ [networkStateWeakPtr] {
+ eventLoop(networkStateWeakPtr);
+ } };
+
+ return networkState;
+};
+
+void stopServer(std::shared_ptr<NetworkState> networkState) {
+ networkState->shouldStopThreads = true;
+
+ networkState->incomingMessages.interrupt();
+
+ networkState->server.close();
+
+ {
+ std::lock_guard guardSoftwareConnections(networkState->softwareConnectionsMutex);
+ networkState->softwareConnections.clear();
+ }
+
+ if (networkState->serverThread.joinable()) {
+ networkState->serverThread.join();
+ }
+ if (networkState->eventLoopThread.joinable()) {
+ networkState->eventLoopThread.join();
+ }
+}
+
+SoftwareConnectionLostError::SoftwareConnectionLostError(const std::string& msg)
+ : ghoul::RuntimeError(fmt::format("{}{}", "Software connection lost", msg), "SoftwareConnection")
+{}
+
+} // namespace openspace::softwareintegration::network
diff --git a/modules/softwareintegration/network/networkengine.h b/modules/softwareintegration/network/network.h
similarity index 69%
rename from modules/softwareintegration/network/networkengine.h
rename to modules/softwareintegration/network/network.h
index 7e68b14dd..13546c061 100644
--- a/modules/softwareintegration/network/networkengine.h
+++ b/modules/softwareintegration/network/network.h
@@ -26,54 +26,45 @@
#define __OPENSPACE_MODULE_SOFTWAREINTEGRATION___NETWORKENGINE___H__
#include <modules/softwareintegration/network/softwareconnection.h>
-#include <modules/softwareintegration/pointdatamessagehandler.h>
-#include <modules/softwareintegration/interruptibleconcurrentqueue.h>
#include <modules/softwareintegration/utils.h>
#include <ghoul/io/socket/tcpsocketserver.h>
+#include <modules/softwareintegration/interruptibleconcurrentqueue.h>
-namespace openspace {
+#include <functional>
+#include <unordered_map>
-class NetworkEngine {
+namespace openspace::softwareintegration::network {
+
+class SoftwareConnectionLostError : public ghoul::RuntimeError {
public:
- NetworkEngine(const int port = 4700);
- ~NetworkEngine();
+ explicit SoftwareConnectionLostError(const std::string& msg);
+};
- struct IncomingMessage {
- size_t connection_id;
- SoftwareConnection::Message message{ softwareintegration::simp::MessageType::Unknown };
- };
+struct IncomingMessage {
+ std::weak_ptr<SoftwareConnection> connection;
+ softwareintegration::simp::MessageType type{ softwareintegration::simp::MessageType::Unknown };
+ std::vector<char> content{};
+ std::string rawMessageType{""};
+};
- void start();
- void stop();
- void postSync();
+struct NetworkState {
+ ghoul::io::TcpSocketServer server;
-private:
- void handleNewSoftwareConnections();
- void handleIncomingMessage(IncomingMessage incomingMessage);
- void peerEventLoop(size_t connection_id);
- void eventLoop();
+ std::thread serverThread;
+ std::thread eventLoopThread;
- // The destuction of the object a shared_ptr is pointing to, occurs when the pointer no longer has any owners
- std::shared_ptr<SoftwareConnection> getSoftwareConnection(size_t id);
+ std::unordered_map<size_t, std::shared_ptr<SoftwareConnection>> softwareConnections{};
+ std::mutex softwareConnectionsMutex{};
- std::unordered_map<size_t, std::shared_ptr<SoftwareConnection>> _softwareConnections;
- std::mutex _softwareConnectionsMutex;
-
- ghoul::io::TcpSocketServer _socketServer;
- std::thread _serverThread;
- std::atomic_bool _shouldStopServerThread = false;
- std::thread _eventLoopThread;
- std::atomic_bool _shouldStopEventThread = false;
-
+ std::atomic_bool shouldStopThreads{ false };
- const int _port;
+ InterruptibleConcurrentQueue<IncomingMessage> incomingMessages{};
+};
- // Message handlers
- PointDataMessageHandler _pointDataMessageHandler;
+std::shared_ptr<NetworkState> serve(const int port = 4700);
- InterruptibleConcurrentQueue<IncomingMessage> _incomingMessages;
-};
+void stopServer(std::shared_ptr<NetworkState> networkState);
-} // namespace openspace
+} // namespace openspace::softwareintegration::network
#endif // __OPENSPACE_MODULE_SOFTWAREINTEGRATION___NETWORKENGINE___H__
diff --git a/modules/softwareintegration/network/networkengine.cpp b/modules/softwareintegration/network/networkengine.cpp
deleted file mode 100644
index 9cade135c..000000000
--- a/modules/softwareintegration/network/networkengine.cpp
+++ /dev/null
@@ -1,245 +0,0 @@
-/*****************************************************************************************
- * *
- * OpenSpace *
- * *
- * Copyright (c) 2014-2022 *
- * *
- * Permission is hereby granted, free of charge, to any person obtaining a copy of this *
- * software and associated documentation files (the "Software"), to deal in the Software *
- * without restriction, including without limitation the rights to use, copy, modify, *
- * merge, publish, distribute, sublicense, and/or sell copies of the Software, and to *
- * permit persons to whom the Software is furnished to do so, subject to the following *
- * conditions: *
- * *
- * The above copyright notice and this permission notice shall be included in all copies *
- * or substantial portions of the Software. *
- * *
- * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, *
- * INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A *
- * PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT *
- * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF *
- * CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE *
- * OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. *
- ****************************************************************************************/
-
-#include <modules/softwareintegration/network/networkengine.h>
-
-#include <openspace/engine/globals.h>
-#include <openspace/engine/globalscallbacks.h>
-#include <openspace/scene/scene.h>
-#include <openspace/scripting/scriptengine.h>
-#include <ghoul/logging/logmanager.h>
-
-
-namespace {
- constexpr const char* _loggerCat = "NetworkEngine";
-} // namespace
-
-namespace openspace {
-
-using namespace softwareintegration;
-
-NetworkEngine::NetworkEngine(const int port)
- : _port{port}
-{}
-
-NetworkEngine::~NetworkEngine() {
- stop();
-}
-
-void NetworkEngine::start() {
- _socketServer.listen(_port);
-
- _serverThread = std::thread([this]() { handleNewSoftwareConnections(); });
- _eventLoopThread = std::thread([this]() { eventLoop(); });
-}
-
-void NetworkEngine::stop() {
- _shouldStopServerThread = true;
-
- {
- std::lock_guard guardSoftwareConnections(_softwareConnectionsMutex);
- for (auto& [id, connectionPtr] : _softwareConnections) {
- SoftwareConnection::NetworkEngineFriends::stopThread(connectionPtr);
- }
- }
-
- _incomingMessages.interrupt();
-
- _shouldStopEventThread = true;
- _socketServer.close();
- _softwareConnections.clear();
-
- if (_serverThread.joinable()) {
- _serverThread.join();
- }
- if (_eventLoopThread.joinable()) {
- _eventLoopThread.join();
- }
-}
-
-void NetworkEngine::postSync() {
- _pointDataMessageHandler.postSync();
-}
-
-void NetworkEngine::handleNewSoftwareConnections() {
- while (!_shouldStopServerThread) {
- std::unique_ptr<ghoul::io::TcpSocket> socket = _socketServer.awaitPendingTcpSocket();
-
- if (!socket) return;
-
- socket->startStreams();
-
- auto p = std::make_shared<SoftwareConnection>(std::move(socket));
- std::lock_guard guard(_softwareConnectionsMutex);
- auto [it, peerInserted] = _softwareConnections.emplace(p->id(), p);
-
- if (peerInserted) {
- auto& connectionPtr = it->second;
- auto thread = std::thread{
- [this, &connectionPtr] {
- peerEventLoop(connectionPtr->id());
- }
- };
- connectionPtr->setThread(thread);
- }
- }
-}
-
-void NetworkEngine::peerEventLoop(size_t connection_id) {
- using namespace std::literals::chrono_literals;
- auto connectionPtr = getSoftwareConnection(connection_id);
-
- while (!connectionPtr->shouldStopThread()) {
- try {
- SoftwareConnection::Message m = connectionPtr->receiveMessageFromSoftware();
- _incomingMessages.push({ connection_id, m });
- }
- catch (const SoftwareConnection::SoftwareConnectionLostError& err) {
- if (connectionPtr->shouldStopThread()) break;
-
- if (connectionPtr && (!connectionPtr->shouldStopThread() || !connectionPtr->isConnectedOrConnecting())) {
- LDEBUG(fmt::format("Connection lost to {}: {}", connection_id, err.message));
- _incomingMessages.push({
- connection_id,
- SoftwareConnection::Message{ simp::MessageType::InternalDisconnection }
- });
- }
- break;
- }
- }
-}
-
-void NetworkEngine::eventLoop() {
- while (!_shouldStopEventThread) {
- // The call to "pop" below will block execution
- // on this thread until interrupt is called
- try {
- auto pm = _incomingMessages.pop();
- handleIncomingMessage(pm);
- }
- catch (const ghoul::RuntimeError&) {
- break;
- }
- }
-}
-
-std::shared_ptr<SoftwareConnection> NetworkEngine::getSoftwareConnection(size_t id) {
- std::lock_guard guard(_softwareConnectionsMutex);
- auto it = _softwareConnections.find(id);
- if (it == _softwareConnections.end()) return nullptr;
- return it->second;
-}
-
-void NetworkEngine::handleIncomingMessage(IncomingMessage incomingMessage) {
- auto connectionPtr = getSoftwareConnection(incomingMessage.connection_id);
-
- if(!connectionPtr) {
- LDEBUG(fmt::format("Trying to handle message from disconnected peer. Aborting."));
- return;
- }
-
- const simp::MessageType messageType = incomingMessage.message.type;
- std::vector<char>& message = incomingMessage.message.content;
-
- switch (messageType) {
- case simp::MessageType::Connection: {
- LDEBUG(fmt::format("Message recieved... Connection: {}", incomingMessage.connection_id));
- size_t offset = 0;
- const std::string software = simp::readString(message, offset);
-
- // Send back message to software to complete handshake
- connectionPtr->sendMessage(simp::formatConnectionMessage(software));
- LINFO(fmt::format("OpenSpace has connected with {} through socket", software));
- break;
- }
- case simp::MessageType::PointData: {
- LDEBUG("Message recieved... Point data");
- _pointDataMessageHandler.handlePointDataMessage(message, connectionPtr);
- break;
- }
- case simp::MessageType::VelocityData: {
- LDEBUG("Message recieved... Velocity data");
- _pointDataMessageHandler.handleVelocityDataMessage(message, connectionPtr);
- break;
- }
- case simp::MessageType::RemoveSceneGraphNode: {
- LDEBUG(fmt::format("Message recieved... Remove SGN"));
- _pointDataMessageHandler.handleRemoveSGNMessage(message, connectionPtr);
- break;
- }
- case simp::MessageType::Color: {
- LDEBUG(fmt::format("Message recieved... Color"));
- _pointDataMessageHandler.handleFixedColorMessage(message, connectionPtr);
- break;
- }
- case simp::MessageType::Colormap: {
- LDEBUG(fmt::format("Message recieved... Colormap"));
- _pointDataMessageHandler.handleColormapMessage(message, connectionPtr);
- break;
- }
- case simp::MessageType::AttributeData: {
- LDEBUG(fmt::format("Message recieved... Attribute data"));
- _pointDataMessageHandler.handleAttributeDataMessage(message, connectionPtr);
- break;
- }
- case simp::MessageType::Opacity: {
- LDEBUG(fmt::format("Message recieved... Opacity"));
- _pointDataMessageHandler.handleOpacityMessage(message, connectionPtr);
- break;
- }
- case simp::MessageType::FixedSize: {
- LDEBUG(fmt::format("Message recieved... Size"));
- _pointDataMessageHandler.handleFixedPointSizeMessage(message, connectionPtr);
- break;
- }
- case simp::MessageType::LinearSize: {
- LDEBUG(fmt::format("Message recieved... Linear size"));
- _pointDataMessageHandler.handleLinearPointSizeMessage(message, connectionPtr);
- break;
- }
- case simp::MessageType::Visibility: {
- LDEBUG(fmt::format("Message recieved... Visibility"));
- _pointDataMessageHandler.handleVisibilityMessage(message, connectionPtr);
- break;
- }
- case simp::MessageType::InternalDisconnection: {
- LDEBUG(fmt::format("Message recieved... Disconnection from software connection: {}", incomingMessage.connection_id));
- std::lock_guard guard(_softwareConnectionsMutex);
- SoftwareConnection::NetworkEngineFriends::stopThread(connectionPtr);
-
- if (_softwareConnections.count(incomingMessage.connection_id)) {
- _softwareConnections.erase(incomingMessage.connection_id);
- }
- break;
- }
- default: {
- LERROR(fmt::format(
- "Unsupported message type: {}", incomingMessage.message.rawMessageType
- ));
- break;
- }
- }
-}
-
-} // namespace openspace
diff --git a/modules/softwareintegration/network/softwareconnection.cpp b/modules/softwareintegration/network/softwareconnection.cpp
index d3d7d9f87..6a06f7623 100644
--- a/modules/softwareintegration/network/softwareconnection.cpp
+++ b/modules/softwareintegration/network/softwareconnection.cpp
@@ -2,7 +2,7 @@
* *
* OpenSpace *
* *
- * Copyright (c) 2014-2021 *
+ * Copyright (c) 2014-2022 *
* *
* Permission is hereby granted, free of charge, to any person obtaining a copy of this *
* software and associated documentation files (the "Software"), to deal in the Software *
@@ -24,6 +24,7 @@
#include <modules/softwareintegration/network/softwareconnection.h>
+#include <modules/softwareintegration/network/network.h>
#include <ghoul/logging/logmanager.h>
#include <openspace/engine/globals.h>
#include <openspace/engine/syncengine.h>
@@ -39,10 +40,6 @@ namespace openspace {
std::atomic_size_t SoftwareConnection::_nextConnectionId = 1;
-SoftwareConnection::SoftwareConnectionLostError::SoftwareConnectionLostError(const std::string& msg)
- : ghoul::RuntimeError(fmt::format("{}{}", "Software connection lost", msg), "SoftwareConnection")
-{}
-
SoftwareConnection::SoftwareConnection(std::unique_ptr<ghoul::io::TcpSocket> socket)
: _id{ _nextConnectionId++ }, _socket{ std::move(socket) }, _sceneGraphNodes{},
_thread{}, _shouldStopThread{ false }
@@ -52,7 +49,7 @@ SoftwareConnection::SoftwareConnection(std::unique_ptr<ghoul::io::TcpSocket> soc
SoftwareConnection::SoftwareConnection(SoftwareConnection&& sc)
: _id{ std::move(sc._id) }, _socket{ std::move(sc._socket) },
- _isConnected{ sc._isConnected }, _sceneGraphNodes{ std::move(sc._sceneGraphNodes) },
+ _sceneGraphNodes{ std::move(sc._sceneGraphNodes) },
_thread{}, _shouldStopThread{ false }
{}
@@ -61,16 +58,13 @@ SoftwareConnection::~SoftwareConnection() {
// destructor is called when disconnecting external
// since NetworkEngine and MessageHandler has
// shared_ptrs to SoftwareConnection, which can cause
- // bugs if not handled properly.
+ // bugs if not handled properly.
// Tips: use weak_ptr instead of shared_ptr in callbacks.
LDEBUG(fmt::format("Removing software connection {}", _id));
- if (!_isConnected) return;
- _isConnected = false;
-
- if (_socket) {
- _socket->disconnect();
- }
+ _shouldStopThread = true;
+ _thread.detach();
+ disconnect();
}
void SoftwareConnection::addPropertySubscription(
@@ -199,8 +193,7 @@ void SoftwareConnection::removePropertySubscriptions(const std::string& identifi
_subscribedProperties.erase(propertySubscriptions);
}
-void SoftwareConnection::PointDataMessageHandlerFriends::removePropertySubscription(
- std::shared_ptr<SoftwareConnection> connectionPtr,
+void SoftwareConnection::removePropertySubscription(
const std::string& propertyName,
const std::string& identifier
) {
@@ -224,8 +217,8 @@ void SoftwareConnection::PointDataMessageHandlerFriends::removePropertySubscript
auto property = r->property(propertyName);
- auto propertySubscriptions = connectionPtr->_subscribedProperties.find(identifier);
- if (propertySubscriptions != connectionPtr->_subscribedProperties.end()) {
+ auto propertySubscriptions = _subscribedProperties.find(identifier);
+ if (propertySubscriptions != _subscribedProperties.end()) {
// At least one property have been subscribed to on this SGN
auto propertySubscription = propertySubscriptions->second.find(propertyName);
if (propertySubscription != propertySubscriptions->second.end()) {
@@ -246,11 +239,11 @@ void SoftwareConnection::disconnect() {
}
bool SoftwareConnection::isConnected() const {
- return _isConnected && _socket && _socket->isConnected();
+ return _socket && _socket->isConnected();
}
bool SoftwareConnection::isConnectedOrConnecting() const {
- return _isConnected && _socket && (_socket->isConnected() || _socket->isConnecting());
+ return _socket && (_socket->isConnected() || _socket->isConnecting());
}
bool SoftwareConnection::sendMessage(const std::string& message) {
@@ -285,7 +278,51 @@ void SoftwareConnection::removeSceneGraphNode(const std::string& identifier) {
}
}
-SoftwareConnection::Message SoftwareConnection::receiveMessageFromSoftware() {
+size_t SoftwareConnection::id() {
+ return _id;
+}
+
+void SoftwareConnection::setThread(std::thread& t) {
+ _thread = std::move(t);
+}
+
+ghoul::io::TcpSocket* SoftwareConnection::socket() {
+ return _socket.get();
+}
+
+namespace softwareintegration::network::connection {
+
+void eventLoop(
+ std::weak_ptr<SoftwareConnection> connectionWeakPtr,
+ std::weak_ptr<NetworkState> networkStateWeakPtr
+) {
+ while (!connectionWeakPtr.expired()) {
+ auto connectionPtr = connectionWeakPtr.lock();
+ if (connectionPtr->_shouldStopThread) break;
+
+ try {
+ IncomingMessage m = receiveMessageFromSoftware(connectionPtr);
+ if (networkStateWeakPtr.expired()) break;
+ networkStateWeakPtr.lock()->incomingMessages.push(m);
+ }
+ catch (const SoftwareConnectionLostError& err) {
+ if (!networkStateWeakPtr.expired()
+ & (!connectionPtr->_shouldStopThread || !connectionPtr->isConnectedOrConnecting())
+ ) {
+ LDEBUG(fmt::format("Connection lost to {}: {}", connectionPtr->id(), err.message));
+ auto networkState = networkStateWeakPtr.lock();
+ if (networkState->softwareConnections.count(connectionPtr->id())) {
+ networkState->softwareConnections.erase(connectionPtr->id());
+ }
+ }
+ break;
+ }
+ }
+}
+
+IncomingMessage receiveMessageFromSoftware(
+ std::shared_ptr<SoftwareConnection> connectionPtr
+) {
// Header consists of version (3 char), message type (4 char) & subject size (15 char)
size_t headerSize = 22 * sizeof(char);
@@ -294,7 +331,7 @@ SoftwareConnection::Message SoftwareConnection::receiveMessageFromSoftware() {
std::vector<char> subjectBuffer;
// Receive the header data
- if (!_socket->get(headerBuffer.data(), headerSize)) {
+ if (!connectionPtr->socket()->get(headerBuffer.data(), headerSize)) {
throw SoftwareConnectionLostError("Failed to read header from socket. Disconnecting.");
}
@@ -329,34 +366,16 @@ SoftwareConnection::Message SoftwareConnection::receiveMessageFromSoftware() {
auto typeEnum = softwareintegration::simp::getMessageType(type);
// Receive the message data
- if (typeEnum != softwareintegration::simp::MessageType::InternalDisconnection && typeEnum != softwareintegration::simp::MessageType::Unknown) {
+ if (typeEnum != softwareintegration::simp::MessageType::Unknown) {
subjectBuffer.resize(subjectSize);
- if (!_socket->get(subjectBuffer.data(), subjectSize)) {
+ if (!connectionPtr->socket()->get(subjectBuffer.data(), subjectSize)) {
throw SoftwareConnectionLostError("Failed to read message from socket. Disconnecting.");
}
}
- return Message{ typeEnum, subjectBuffer, type };
+ return { connectionPtr, typeEnum, subjectBuffer, type };
}
-bool SoftwareConnection::shouldStopThread() {
- return _shouldStopThread;
-}
-
-size_t SoftwareConnection::id() {
- return _id;
-}
-
-void SoftwareConnection::setThread(std::thread& t) {
- _thread = std::move(t);
-}
-
-void SoftwareConnection::NetworkEngineFriends::stopThread(std::shared_ptr<SoftwareConnection> connectionPtr) {
- connectionPtr->_shouldStopThread = true;
- connectionPtr->disconnect();
- if (connectionPtr->_thread.joinable()) {
- connectionPtr->_thread.join();
- }
-}
+} // namespace softwareintegration::network::connection
} // namespace openspace
diff --git a/modules/softwareintegration/network/softwareconnection.h b/modules/softwareintegration/network/softwareconnection.h
index 4de782f99..dee240acd 100644
--- a/modules/softwareintegration/network/softwareconnection.h
+++ b/modules/softwareintegration/network/softwareconnection.h
@@ -2,7 +2,7 @@
* *
* OpenSpace *
* *
- * Copyright (c) 2014-2021 *
+ * Copyright (c) 2014-2022 *
* *
* Permission is hereby granted, free of charge, to any person obtaining a copy of this *
* software and associated documentation files (the "Software"), to deal in the Software *
@@ -25,8 +25,8 @@
#ifndef __OPENSPACE_MODULE_SOFTWAREINTEGRATION___SOFTWARECONNECTION___H__
#define __OPENSPACE_MODULE_SOFTWAREINTEGRATION___SOFTWARECONNECTION___H__
-#include <openspace/network/messagestructures.h>
#include <modules/softwareintegration/utils.h>
+#include <openspace/network/messagestructures.h>
#include <ghoul/io/socket/tcpsocket.h>
#include <openspace/properties/property.h>
@@ -36,8 +36,30 @@
namespace openspace {
class Renderable;
+class SoftwareConnection;
+
+namespace softwareintegration::network {
+
+struct NetworkState;
+struct IncomingMessage;
+
+namespace connection {
+ void eventLoop(
+ std::weak_ptr<SoftwareConnection> connectionWeakPtr,
+ std::weak_ptr<softwareintegration::network::NetworkState> networkStateWeakPtr
+ );
+
+ IncomingMessage receiveMessageFromSoftware(
+ std::shared_ptr<SoftwareConnection> connectionPtr
+ );
+} // namespace connection
+
+} // namespace softwareintegration::network
+
+using namespace softwareintegration::network;
class SoftwareConnection {
+
public:
using OnChangeHandle = properties::Property::OnChangeHandle;
struct PropertySubscription {
@@ -49,14 +71,6 @@ public:
using Identifier = std::string;
using SubscribedProperties = std::unordered_map<Identifier, PropertySubscriptions>;
- struct Message {
- softwareintegration::simp::MessageType type;
- std::vector<char> content{};
- std::string rawMessageType{""};
- };
-
- class SoftwareConnectionLostError;
-
explicit SoftwareConnection(std::unique_ptr<ghoul::io::TcpSocket> socket);
SoftwareConnection(SoftwareConnection&& p);
~SoftwareConnection();
@@ -76,40 +90,33 @@ public:
// const std::string& identifier
// );
- SoftwareConnection::Message receiveMessageFromSoftware();
-
void addSceneGraphNode(const std::string& identifier);
void removeSceneGraphNode(const std::string& identifier);
size_t id();
- size_t nConnections();
void setThread(std::thread& t);
- bool shouldStopThread();
- class NetworkEngineFriends {
- private:
- static void stopThread(std::shared_ptr<SoftwareConnection> connectionPtr);
- friend class NetworkEngine;
- };
+ friend void connection::eventLoop(
+ std::weak_ptr<SoftwareConnection> connectionWeakPtr,
+ std::weak_ptr<NetworkState> networkStateWeakPtr
+ );
- class PointDataMessageHandlerFriends {
- private:
- static void removePropertySubscription(
- std::shared_ptr<SoftwareConnection> connectionPtr,
- const std::string& propertyName,
- const std::string& identifier
- );
- friend class PointDataMessageHandler;
- };
+ friend IncomingMessage connection::receiveMessageFromSoftware(
+ std::shared_ptr<SoftwareConnection> connectionPtr
+ );
+
+ void removePropertySubscription(const std::string& propertyName, const std::string& identifier);
-private:
void removePropertySubscriptions(const std::string& identifier);
+ ghoul::io::TcpSocket* socket();
+
+private:
+
SubscribedProperties _subscribedProperties;
std::unordered_set<std::string> _sceneGraphNodes;
- bool _isConnected = true;
std::unique_ptr<ghoul::io::TcpSocket> _socket;
size_t _id;
@@ -119,11 +126,6 @@ private:
static std::atomic_size_t _nextConnectionId;
};
-class SoftwareConnection::SoftwareConnectionLostError : public ghoul::RuntimeError {
-public:
- explicit SoftwareConnectionLostError(const std::string& msg);
-};
-
} // namespace openspace
#endif // __OPENSPACE_MODULE_SOFTWAREINTEGRATION___SOFTWAREINTEGRATIONMODULE___H__
diff --git a/modules/softwareintegration/pointdatamessagehandler.h b/modules/softwareintegration/pointdatamessagehandler.h
deleted file mode 100644
index 5fe4cd399..000000000
--- a/modules/softwareintegration/pointdatamessagehandler.h
+++ /dev/null
@@ -1,106 +0,0 @@
-/*****************************************************************************************
- * *
- * OpenSpace *
- * *
- * Copyright (c) 2014-2021 *
- * *
- * Permission is hereby granted, free of charge, to any person obtaining a copy of this *
- * software and associated documentation files (the "Software"), to deal in the Software *
- * without restriction, including without limitation the rights to use, copy, modify, *
- * merge, publish, distribute, sublicense, and/or sell copies of the Software, and to *
- * permit persons to whom the Software is furnished to do so, subject to the following *
- * conditions: *
- * *
- * The above copyright notice and this permission notice shall be included in all copies *
- * or substantial portions of the Software. *
- * *
- * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, *
- * INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A *
- * PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT *
- * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF *
- * CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE *
- * OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. *
- ****************************************************************************************/
-
-#ifndef __OPENSPACE_MODULE_SOFTWAREINTEGRATION___POINTDATAMESSAGEHANDLER___H__
-#define __OPENSPACE_MODULE_SOFTWAREINTEGRATION___POINTDATAMESSAGEHANDLER___H__
-
-#include <unordered_map>
-
-#include <openspace/properties/propertyowner.h>
-
-#include <modules/softwareintegration/network/softwareconnection.h>
-
-namespace openspace {
-
-class Renderable;
-
-class PointDataMessageHandler {
- struct Callback {
- std::function<void()> function;
- std::vector<softwareintegration::storage::Key> waitForData = {};
- std::string description = "???"; // To help debugging. Maybe remove?
- };
- using CallbackList = std::vector<Callback>;
- using CallbackMap = std::unordered_map<std::string, CallbackList>;
-
-public:
- void handlePointDataMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection);
- void handleVelocityDataMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection);
- void handleFixedColorMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection);
- void handleColormapMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection);
- void handleAttributeDataMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection);
- void handleOpacityMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection);
- void handleFixedPointSizeMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection);
- void handleLinearPointSizeMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection);
- void handleVisibilityMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection);
- void handleRemoveSGNMessage(const std::vector<char>& message, std::shared_ptr<SoftwareConnection> connection);
-
- void postSync();
-
-private:
- const Renderable* getRenderable(const std::string& identifier);
- void checkRenderable(
- const std::vector<char>& message, size_t& messageOffset,
- std::shared_ptr<SoftwareConnection> connection, std::string& identifier
- );
-
- void addCallback(
- const std::string& identifier,
- const Callback& newCallback
- );
-
- CallbackMap _onceNodeExistsCallbacks;
- std::mutex _onceNodeExistsCallbacksMutex;
- size_t _onceNodeExistsCallbacksRetries{0};
-
- void onFixedColorChange(
- properties::Property* property,
- const std::string& identifier,
- std::shared_ptr<SoftwareConnection> connection
- );
- void onOpacityChange(
- properties::Property* property,
- const std::string& identifier,
- std::shared_ptr<SoftwareConnection> connection
- );
- void onFixedPointSizeChange(
- properties::Property* property,
- const std::string& identifier,
- std::shared_ptr<SoftwareConnection> connection
- );
- void onVisibilityChange(
- properties::Property* property,
- const std::string& identifier,
- std::shared_ptr<SoftwareConnection> connection
- );
-
- void convertToMeterPerSecond(
- softwareintegration::simp::LengthUnit currLengthUnit,
- std::vector<float>& data
- );
-};
-
-} // namespace openspace
-
-#endif // __OPENSPACE_MODULE_SOFTWAREINTEGRATION___POINTDATAMESSAGEHANDLER___H__
diff --git a/modules/softwareintegration/softwareintegrationmodule.cpp b/modules/softwareintegration/softwareintegrationmodule.cpp
index 91ed0ec77..bfe2c0b2b 100644
--- a/modules/softwareintegration/softwareintegrationmodule.cpp
+++ b/modules/softwareintegration/softwareintegrationmodule.cpp
@@ -24,13 +24,13 @@
#include <modules/softwareintegration/softwareintegrationmodule.h>
-#include <ghoul/filesystem/filesystem.h>
+#include <modules/softwareintegration/rendering/renderablepointscloud.h>
+#include <modules/softwareintegration/messagehandler.h>
#include <ghoul/logging/logmanager.h>
#include <openspace/engine/globals.h>
#include <openspace/engine/syncengine.h>
#include <openspace/engine/moduleengine.h>
#include <openspace/scripting/scriptengine.h>
-#include <modules/softwareintegration/rendering/renderablepointscloud.h>
#include <openspace/documentation/documentation.h>
#include <openspace/engine/globalscallbacks.h>
#include <openspace/engine/windowdelegate.h>
@@ -48,14 +48,9 @@ constexpr const char* _loggerCat = "SoftwareIntegrationModule";
namespace openspace {
-SoftwareIntegrationModule::SoftwareIntegrationModule() : OpenSpaceModule(Name) {
- if (global::windowDelegate->isMaster()) {
- // The Master node will handle all communication with the external software
- // and forward it to the Client nodes
- // 4700, is the defualt port where the tcp socket will be opened to the ext. software
- _networkEngine = std::make_unique<NetworkEngine>();
- }
-}
+SoftwareIntegrationModule::SoftwareIntegrationModule()
+ : OpenSpaceModule(Name)
+{}
SoftwareIntegrationModule::~SoftwareIntegrationModule() {
internalDeinitialize();
@@ -89,17 +84,21 @@ void SoftwareIntegrationModule::internalInitialize(const ghoul::Dictionary&) {
fRenderable->registerClass<RenderablePointsCloud>("RenderablePointsCloud");
if (global::windowDelegate->isMaster()) {
- _networkEngine->start();
+ // The Master node will handle all communication with the external software
+ // and forward it to the Client nodes
+ _networkState = softwareintegration::network::serve();
global::callback::postSyncPreDraw->emplace_back([this]() {
- if (!_networkEngine) return;
- _networkEngine->postSync();
+ softwareintegration::network::postSyncCallbacks();
});
}
}
void SoftwareIntegrationModule::internalDeinitialize() {
global::syncEngine->removeSyncables(getSyncables());
+ if (_networkState) {
+ softwareintegration::network::stopServer(_networkState);
+ }
}
std::vector<documentation::Documentation> SoftwareIntegrationModule::documentations() const {
diff --git a/modules/softwareintegration/softwareintegrationmodule.h b/modules/softwareintegration/softwareintegrationmodule.h
index 411ed2249..e889f6068 100644
--- a/modules/softwareintegration/softwareintegrationmodule.h
+++ b/modules/softwareintegration/softwareintegrationmodule.h
@@ -26,10 +26,9 @@
#define __OPENSPACE_MODULE_SOFTWAREINTEGRATION___SOFTWAREINTEGRATIONMODULE___H__
#include <openspace/util/openspacemodule.h>
-
-#include <modules/softwareintegration/network/networkengine.h>
#include <modules/softwareintegration/syncablefloatdatastorage.h>
#include <openspace/documentation/documentation.h>
+#include <modules/softwareintegration/network/network.h>
namespace openspace {
@@ -64,8 +63,7 @@ private:
// Centralized storage for datasets
SyncableFloatDataStorage _syncableFloatDataStorage;
- // Network engine
- std::unique_ptr<NetworkEngine> _networkEngine;
+ std::shared_ptr<softwareintegration::network::NetworkState> _networkState;
};
} // namespace openspace
diff --git a/modules/softwareintegration/utils.cpp b/modules/softwareintegration/utils.cpp
index 2d76da955..90308bf9c 100644
--- a/modules/softwareintegration/utils.cpp
+++ b/modules/softwareintegration/utils.cpp
@@ -38,6 +38,20 @@ namespace openspace {
namespace softwareintegration {
namespace storage {
+
+ // Anonymous namespace
+ namespace {
+
+ const std::unordered_map<std::string, Key> _keyStringFromKey{
+ { "DataPoints", Key::DataPoints },
+ { "VelocityData", Key::VelocityData },
+ { "Colormap", Key::Colormap },
+ { "ColormapAttributeData", Key::ColormapAttrData },
+ { "LinearSizeAttributeData", Key::LinearSizeAttrData },
+ };
+
+ } // namespace
+
bool hasStorageKey(const std::string& key) {
return _keyStringFromKey.count(key) > 0;
}
@@ -72,6 +86,7 @@ namespace {
const std::unordered_map<std::string, MessageType> _messageTypeFromSIMPType{
{ "CONN", MessageType::Connection },
{ "PDAT", MessageType::PointData },
+ { "VDAT", MessageType::VelocityData },
{ "RSGN", MessageType::RemoveSceneGraphNode },
{ "FCOL", MessageType::Color },
{ "LCOL", MessageType::Colormap },
diff --git a/modules/softwareintegration/utils.h b/modules/softwareintegration/utils.h
index cd6804c34..c6caa77bd 100644
--- a/modules/softwareintegration/utils.h
+++ b/modules/softwareintegration/utils.h
@@ -40,13 +40,6 @@ enum class Key : uint8_t {
Unknown
};
-const std::unordered_map<std::string, Key> _keyStringFromKey{
- { "DataPoints", Key::DataPoints },
- { "Colormap", Key::Colormap },
- { "ColormapAttributeData", Key::ColormapAttrData },
- { "LinearSizeAttributeData", Key::LinearSizeAttrData },
-};
-
Key getStorageKey(const std::string& key);
std::string getStorageKeyString(const Key key);
@@ -73,7 +66,6 @@ enum class MessageType : uint32_t {
FixedSize,
LinearSize,
Visibility,
- InternalDisconnection,
Unknown
};
--
GitLab