[occ] Fix large cfg payloads cut off because of gRPC slices

Author: teo

occ/plugin/OccFMQCommon.cxx

@@ -77,6 +77,7 @@ std::tuple<OccLite::nopb::TransitionResponse, ::grpc::Status> doTransition(fair:
     auto onDeviceStateChange = [&](fair::mq::PluginServices::DeviceState reachedState) {
         // CONFIGURE arguments must be pushed during InitializingDevice
         if (reachedState == fair::mq::PluginServices::DeviceState::InitializingDevice) {
+            OLOG(debug) << "INITIALIZING_DEVICE reached, pushing configuration parameters";
 
             // FIXME: workaround which special cases a stoi for certain properties
             // which must be pushed as int.
@@ -92,38 +93,42 @@ std::tuple<OccLite::nopb::TransitionResponse, ::grpc::Status> doTransition(fair:
             for (auto it = arguments.cbegin(); it != arguments.cend(); ++it) {
                 std::string key = it->key;
                 std::string value = it->value;
-                if (boost::starts_with(key, "chans.")) {
-                    std::vector<std::string> split;
-                    boost::split(split, key, std::bind(std::equal_to<>(), '.', std::placeholders::_1));
-                    if (std::find(intKeys.begin(), intKeys.end(), split.back()) != intKeys.end()) {
-                        auto intValue = std::stoi(value);
-                        m_pluginServices->SetProperty(key, intValue);
-                        OLOG(debug) << "SetProperty(chan int) called " << key << ":" << intValue;
-                    }
-                    else {
+                try {
+                    if (boost::starts_with(key, "chans.")) {
+                        std::vector<std::string> split;
+                        boost::split(split, key, std::bind(std::equal_to<>(), '.', std::placeholders::_1));
+                        if (std::find(intKeys.begin(), intKeys.end(), split.back()) != intKeys.end()) {
+                            auto intValue = std::stoi(value);
+                            m_pluginServices->SetProperty(key, intValue);
+                            OLOG(debug) << "SetProperty(chan int) called " << key << ":" << intValue;
+                        } else {
+                            m_pluginServices->SetProperty(key, value);
+                            OLOG(debug) << "SetProperty(chan string) called " << key << ":" << value;
+                        }
+                    } else if (boost::starts_with(key, "__ptree__:")) {
+                        // we need to ptreefy whatever payload we got under this kind of key, on a best-effort basis
+                        auto[newKey, newValue] = propMapEntryToPtree(key, value);
+                        if (newKey ==
+                            key) { // Means something went wrong and the called function already printed out the message
+                            continue;
+                        }
+
+                        m_pluginServices->SetProperty(newKey, newValue);
+                        OLOG(debug) << "SetProperty(ptree) called " << newKey << ":" << value;
+                    } else { // default case, 1 k-v ==> 1 SetProperty
                         m_pluginServices->SetProperty(key, value);
-                        OLOG(debug) << "SetProperty(chan string) called " << key << ":" << value;
-                    }
-                }
-                else if (boost::starts_with(key, "__ptree__:")) {
-                    // we need to ptreefy whatever payload we got under this kind of key, on a best-effort basis
-                    auto [newKey, newValue] = propMapEntryToPtree(key, value);
-                    if (newKey == key) { // Means something went wrong and the called function already printed out the message
-                        continue;
+                        OLOG(debug) << "SetProperty(string) called " << key << ":" << value;
                     }
-
-                    m_pluginServices->SetProperty(newKey, newValue);
-                    OLOG(debug) << "SetProperty(ptree) called " << newKey << ":" << value;
                 }
-                else { // default case, 1 k-v ==> 1 SetProperty
-                    m_pluginServices->SetProperty(key, value);
-                    OLOG(debug) << "SetProperty(string) called " << key << ":" << value;
+                catch (std::runtime_error &e) {
+                    OLOG(warning) << "SetProperty call failed for key " + key + " with reason: " << e.what();
                 }
             }
             auto chInfo = m_pluginServices->GetPropertiesAsStringStartingWith("chans.");
             for (auto it = chInfo.cbegin(); it != chInfo.cend(); ++it) {
-                OLOG(debug) << "Written chan cfg: " << it->first << ":" << it->second;
+                OLOG(debug) << "Written chan cfg: " << it->first << ": " << it->second;
             }
+            OLOG(debug) << "INITIALIZING_DEVICE configuration push DONE";
         }
 
         std::unique_lock<std::mutex> lk(cv_mu);

occ/plugin/OccLiteServer.cxx

@@ -79,6 +79,8 @@ ::grpc::Status OccLite::Service::GetState(::grpc::ServerContext* context,
     (void) context;
     (void) request;
 
+    OLOG(info) << "Incoming GetState request: " << request->JsonMessage::Serialize();
+
     auto state = fair::mq::PluginServices::ToStr(m_pluginServices->GetCurrentDeviceState());
     pid_t pid = getpid();
 
@@ -93,16 +95,19 @@ ::grpc::Status OccLite::Service::Transition(::grpc::ServerContext* context,
                                             const OccLite::nopb::TransitionRequest* request,
                                             OccLite::nopb::TransitionResponse* response)
 {
-    OLOG(info) << "Incoming Transition request:" << request->JsonMessage::Serialize();
+    OLOG(info) << "Incoming Transition request: " << request->JsonMessage::Serialize();
 
     auto transitionOutcome = doTransition(m_pluginServices, *request);
     ::grpc::Status grpcStatus = std::get<1>(transitionOutcome);
     if (!grpcStatus.ok()) {
-        return grpcStatus;
+        OLOG(error) << "Transition failed with error: " << grpcStatus.error_code() << " " << grpcStatus.error_message() << " " << grpcStatus.error_details();
+        return grpc::Status::CANCELLED;
     }
 
     auto nopbResponse = std::get<0>(transitionOutcome);
     *response = nopbResponse;
+    OLOG(info) << "Transition response: " << response->state << "ok: " << response->ok;
+
     return grpc::Status::OK;
 }
 

occ/plugin/litestructs/JsonMessage.cxx

@@ -24,6 +24,7 @@
 
 #include "JsonMessage.h"
 
+
 std::string OccLite::nopb::JsonMessage::Serialize() const
 {
     rapidjson::StringBuffer ss;
@@ -59,20 +60,29 @@ bool OccLite::nopb::JsonMessage::Deserialize(::grpc::ByteBuffer* byte_buffer)
 {
     auto slices = new std::vector<::grpc::Slice>;
     auto status = byte_buffer->Dump(slices);
-    if (slices->size() != 1) return false;
 
-    auto rawSlice = (*slices)[0].c_slice();
-    std::string str = grpc::StringFromCopiedSlice(rawSlice);
-    ::grpc::g_core_codegen_interface->grpc_slice_unref(rawSlice);
-    OLOG(info) << "Deserializing Message:\n" << str;
+    if (!status.ok()) {
+        OLOG(info) << "Cannot dump JsonMessage slices, error code " << status.error_code() << " " << status.error_message() << " " << status.error_details();
+        return false;
+    }
+
+    std::stringstream ss;
+    for (auto sl = slices->cbegin(); sl != slices->cend(); sl++) {
+        auto rawSlice = sl->c_slice();
+        std::string str = grpc::StringFromCopiedSlice(rawSlice);
+        ss << str;
+        ::grpc::g_core_codegen_interface->grpc_slice_unref(rawSlice);
+    }
+
+    OLOG(info) << "Deserialized JsonMessage: " << ss.str();
 
-    return Deserialize(str);
+    return Deserialize(ss.str());
 }
 
 ::grpc::ByteBuffer* OccLite::nopb::JsonMessage::SerializeToByteBuffer() const
 {
     std::string str = Serialize();
-    OLOG(info) << "Serialized Message:\n" << str;
+    OLOG(info) << "Serialized JsonMessage: " << str;
 
     // grpc::string = std::string
     // We build a Slice(grpc::string) and we add it to the ByteBuffer

occ/plugin/litestructs/Transition.cxx

@@ -41,11 +41,8 @@ bool OccLite::nopb::TransitionRequest::Serialize(rapidjson::Writer<rapidjson::St
 
 bool OccLite::nopb::TransitionRequest::Deserialize(const rapidjson::Value& obj)
 {
-    OLOG(info) << "Deserializing TransitionRequest";
-
     srcState = obj["srcState"].GetString();
     transitionEvent = obj["transitionEvent"].GetString();
-    OLOG(info) << "state and transitionEvent ok";
 
     if (obj.HasMember("arguments")) {
         auto array = obj["arguments"].GetArray();
@@ -59,8 +56,7 @@ bool OccLite::nopb::TransitionRequest::Deserialize(const rapidjson::Value& obj)
             arguments.push_back(*ce);
         }
     }
-    OLOG(info) << "Deserialized TransitionRequest:";
-    OLOG(info) << JsonMessage::Serialize();
+    OLOG(info) << "Deserialized TransitionRequest: " << JsonMessage::Serialize();
 
     return true;
 }