[occ] Emit END_OF_DATA event from OccServer

Author: teo

occ/occlib/OccServer.cxx

@@ -51,6 +51,42 @@ OccServer::~OccServer()
     m_checkerThread.join();
 }
 
+grpc::Status OccServer::EventStream(grpc::ServerContext* context,
+                                    const occ_pb::EventStreamRequest* request,
+                                    grpc::ServerWriter<occ_pb::EventStreamReply>* writer)
+{
+    std::cout << "[request StateStream] handler BEGIN" << std::endl;
+
+    boost::uuids::basic_random_generator<boost::mt19937> gen;
+    std::string id = boost::uuids::to_string(gen());
+
+    boost::lockfree::queue<pb::DeviceEvent*> eventQueue;
+    m_eventQueues[id] = &eventQueue;
+    DEFER({
+        m_eventQueues.erase(id);
+    });
+
+    bool isStreamOpen = true;
+    while (!m_destroying && isStreamOpen) {
+        pb::DeviceEvent *newEvent;
+        bool ok = eventQueue.pop(newEvent);
+        if (!ok) {  // queue empty, sleep and retry
+            std::this_thread::sleep_for(2ms);
+            continue;
+        }
+
+        pb::EventStreamReply response;
+        if (newEvent) {
+            response.mutable_event()->CopyFrom(*newEvent);
+            isStreamOpen = writer->Write(response);
+            delete newEvent;
+        }
+    }
+
+    std::cout << "[request StateStream] handler END" << std::endl;
+    return ::grpc::Status::OK;
+}
+
 grpc::Status OccServer::StateStream(grpc::ServerContext* context,
                                     const pb::StateStreamRequest* request,
                                     grpc::ServerWriter<pb::StateStreamReply>* writer)
@@ -132,9 +168,9 @@ grpc::Status OccServer::Transition(grpc::ServerContext* context,
     }
 
     std::string srcStateStr = request->srcstate();
-    std::string event = request->event();
+    std::string event = request->transitionevent();
     auto arguments = request->arguments();
-    const std::string finalState = EXPECTED_FINAL_STATE.at(request->event());
+    const std::string finalState = EXPECTED_FINAL_STATE.at(request->transitionevent());
 
     t_State currentState = m_rco->getState();
     std::string currentStateStr = getStringFromState(currentState);
@@ -161,7 +197,7 @@ grpc::Status OccServer::Transition(grpc::ServerContext* context,
     std::string newStateStr = getStringFromState(newState);
 
     response->set_state(newStateStr);
-    response->set_event(request->event());
+    response->set_transitionevent(request->transitionevent());
     response->set_ok(newStateStr == finalState);
     if (newState == error) {                   // ERROR state
         response->set_trigger(pb::DEVICE_ERROR);
@@ -329,6 +365,16 @@ void OccServer::publishState(t_State s)
     }
 }
 
+void OccServer::pushEvent(pb::DeviceEvent* event)
+{
+    for (auto item : m_eventQueues) {
+        item.second->push(event);
+    }
+    printf("Object: %s - pushing event = %s\n",
+           m_rco->getName().c_str(),
+           pb::DeviceEventType_Name(event->type()).c_str());
+}
+
 bool OccServer::checkMachineDone()
 {
     std::lock_guard<std::mutex> lock(m_mu);
@@ -350,7 +396,12 @@ void OccServer::runChecker()
         // execute periodic actions, as defined for t_State::running
         if (currentState == t_State::running) {
             int err = m_rco->iterateRunning();
-            if (err) {
+            if (err == 1) { // signal EndOfData event
+                auto eodEvent = new pb::DeviceEvent;
+                eodEvent->set_type(pb::END_OF_DATA);
+                pushEvent(eodEvent);
+            }
+            else if (err) {
                 updateState(t_State::error);
             }
         }
@@ -367,4 +418,4 @@ void OccServer::runChecker()
             std::this_thread::sleep_for(1ms);
         }
     }
-}
+}

occ/occlib/OccServer.h

@@ -70,6 +70,10 @@ class OccServer final : public pb::Occ::Service
      */
     virtual ~OccServer();
 
+    grpc::Status EventStream(grpc::ServerContext* context,
+                             const pb::EventStreamRequest* request,
+                             grpc::ServerWriter<pb::EventStreamReply>* writer) override;
+
     grpc::Status StateStream(grpc::ServerContext* context,
                              const pb::StateStreamRequest* request,
                              grpc::ServerWriter<pb::StateStreamReply>* writer) override;
@@ -89,6 +93,7 @@ class OccServer final : public pb::Occ::Service
     void updateState(t_State s);
 
     void publishState(t_State s);
+    void pushEvent(pb::DeviceEvent* event);
 
     void runChecker();
 
@@ -100,6 +105,7 @@ class OccServer final : public pb::Occ::Service
     bool m_machineDone;
 
     std::unordered_map<std::string, boost::lockfree::queue<t_State>* > m_stateQueues;
+    std::unordered_map<std::string, boost::lockfree::queue<pb::DeviceEvent*>* > m_eventQueues;
 };
 
 

occ/plugin/OccPluginServer.cxx

@@ -177,7 +177,7 @@ OccPluginServer::Transition(grpc::ServerContext* context,
     }
 
     std::string srcState = request->srcstate();
-    std::string event = request->event();
+    std::string event = request->transitionevent();
     auto arguments = request->arguments();
 
     std::string currentState = fair::mq::PluginServices::ToStr(m_pluginServices->GetCurrentDeviceState());
@@ -311,8 +311,8 @@ OccPluginServer::Transition(grpc::ServerContext* context,
         return grpc::Status(grpc::INTERNAL, "cannot request transition, OCC plugin has no device control");
     }
     catch (std::out_of_range& e) {
-        OLOG(ERROR) << "[request Transition] invalid transition name: " << request->event();
-        return grpc::Status(grpc::INVALID_ARGUMENT, "argument " + request->event() + " is not a valid transition name");
+        OLOG(ERROR) << "[request Transition] invalid transition name: " << request->transitionevent();
+        return grpc::Status(grpc::INVALID_ARGUMENT, "argument " + request->transitionevent() + " is not a valid transition name");
     }
 
     {
@@ -362,7 +362,7 @@ OccPluginServer::Transition(grpc::ServerContext* context,
     }
 
     response->set_state(newStates.back());
-    response->set_event(request->event());
+    response->set_transitionevent(request->transitionevent());
     response->set_ok(newStates.back() == finalState);
     if (newStates.back() == "ERROR") {              // ERROR state
         response->set_trigger(pb::DEVICE_ERROR);

occ/protos/occ.proto

@@ -33,6 +33,7 @@ option go_package = "pb";
 service Occ {
     // We have to have a notification stream because the FairMQDevice might transition
     // on its own for whatever reason.
+    rpc EventStream (EventStreamRequest) returns (stream EventStreamReply) {}
     rpc StateStream (StateStreamRequest) returns (stream StateStreamReply) {}
     rpc GetState (GetStateRequest) returns (GetStateReply) {}
     rpc Transition (TransitionRequest) returns (TransitionReply) {}
@@ -49,12 +50,26 @@ enum StateType {
     STATE_INTERMEDIATE = 1;
 }
 
+enum DeviceEventType {
+    NULL_DEVICE_EVENT = 0;
+    END_OF_DATA = 1;
+}
+
 message StateStreamRequest {}
+
 message StateStreamReply {
     StateType type = 1;
     string state = 2;
 }
 
+message EventStreamRequest {}
+message DeviceEvent {
+    DeviceEventType type = 1;
+}
+message EventStreamReply {
+    DeviceEvent event = 1;
+}
+
 message GetStateRequest {}
 message GetStateReply {
     string state = 1;
@@ -66,12 +81,12 @@ message ConfigEntry {
 }
 message TransitionRequest {
     string srcState = 1;
-    string event = 2;
+    string transitionEvent = 2;
     repeated ConfigEntry arguments = 3;
 }
 message TransitionReply {
     StateChangeTrigger trigger = 1;
     string state = 2;
-    string event = 3;
+    string transitionEvent = 3;
     bool ok = 4;
 }