VuoRuntimeCommunicator.hh

Go to the documentation of this file.

 
#pragma once
 
#include "VuoCompositionState.h"
#include "VuoTelemetry.h"
 
class VuoRuntimePersistentState;
struct NodeContext;
 
class VuoRuntimeCommunicator
{
private:
    bool _hasZmqConnection;   
 
    void *zmqContext;   
    void *zmqControl;   
    void *zmqSelfReceive;   
    void *zmqSelfSend;      
    void *zmqTelemetry;     
 
    dispatch_queue_t controlQueue;  
    dispatch_queue_t telemetryQueue;  
    dispatch_source_t controlTimer;  
    dispatch_source_t telemetryTimer;  
    dispatch_semaphore_t controlCanceled;  
    dispatch_semaphore_t telemetryCanceled;  
 
    int runnerPipe;  
 
    set<string> compositionsSendingAllTelemetry;  
    set<string> compositionsSendingEventTelemetry;  
    map<string, set<string> > portsSendingDataTelemetry;  
 
    VuoRuntimePersistentState *persistentState;  
 
    void sendControlReply(enum VuoControlReply reply, zmq_msg_t *messages, unsigned int messageCount);
    void sendTelemetry(enum VuoTelemetry type, zmq_msg_t *messages, unsigned int messageCount);
 
    static char * mergeEnumDetails(string type, const char *details);
 
    void subscribeToPortDataTelemetry(const char *compositionIdentifier, const char *portIdentifer);
    void unsubscribeFromPortDataTelemetry(const char *compositionIdentifier, const char *portIdentifer);
    bool isSubscribedToPortDataTelemetry(const char *compositionIdentifier, const char *portIdentifer);
 
    void subscribeToEventTelemetry(const char *compositionIdentifier);
    void unsubscribeFromEventTelemetry(const char *compositionIdentifier);
    bool isSubscribedToEventTelemetry(const char *compositionIdentifier);
 
    void subscribeToAllTelemetry(const char *compositionIdentifier);
    void unsubscribeFromAllTelemetry(const char *compositionIdentifier);
    bool isSubscribedToAllTelemetry(const char *compositionIdentifier);
 
    void sendHeartbeat(bool blocking = false);
 
 
    typedef void (*vuoInstanceInitType)(void);
    vuoInstanceInitType vuoInstanceInit;
    typedef void (*vuoInstanceTriggerStartType)(void);
    vuoInstanceTriggerStartType vuoInstanceTriggerStart;
    typedef void (*vuoInstanceTriggerStopType)(void);
    vuoInstanceTriggerStopType vuoInstanceTriggerStop;
    typedef void (*vuoSetInputPortValueType)(const char *portIdentifier, char *valueAsString);
    vuoSetInputPortValueType vuoSetInputPortValue;
    typedef unsigned int (*getPublishedInputPortCountType)(void);
    getPublishedInputPortCountType getPublishedInputPortCount;
    typedef unsigned int (*getPublishedOutputPortCountType)(void);
    getPublishedOutputPortCountType getPublishedOutputPortCount;
    typedef char ** (*getPublishedInputPortNamesType)(void);
    getPublishedInputPortNamesType getPublishedInputPortNames;
    typedef char ** (*getPublishedOutputPortNamesType)(void);
    getPublishedOutputPortNamesType getPublishedOutputPortNames;
    typedef char ** (*getPublishedInputPortTypesType)(void);
    getPublishedInputPortTypesType getPublishedInputPortTypes;
    typedef char ** (*getPublishedOutputPortTypesType)(void);
    getPublishedOutputPortTypesType getPublishedOutputPortTypes;
    typedef char ** (*getPublishedInputPortDetailsType)(void);
    getPublishedInputPortDetailsType getPublishedInputPortDetails;
    typedef char ** (*getPublishedOutputPortDetailsType)(void);
    getPublishedOutputPortDetailsType getPublishedOutputPortDetails;
    typedef void (*firePublishedInputPortEventType)(const char *const *names, unsigned int count);
    firePublishedInputPortEventType firePublishedInputPortEvent;
    typedef void (*setPublishedInputPortValueType)(const char *portIdentifier, const char *valueAsString);
    setPublishedInputPortValueType setPublishedInputPortValue;
    typedef char * (*getPublishedInputPortValueType)(const char *portIdentifier, int shouldUseInterprocessSerialization);
    getPublishedInputPortValueType getPublishedInputPortValue;
    typedef char * (*getPublishedOutputPortValueType)(const char *portIdentifier, int shouldUseInterprocessSerialization);
    getPublishedOutputPortValueType getPublishedOutputPortValue;
 
public:
    VuoRuntimeCommunicator(VuoRuntimePersistentState *persistentState);
    ~VuoRuntimeCommunicator(void);
    void updateCompositionSymbols(void *compositionBinaryHandle);
    void openConnection(void *_zmqContext, const char *controlURL, const char *telemetryURL, int runnerPipe);
    void closeConnection(void);
 
    bool hasZmqConnection(void);
    dispatch_queue_t getControlQueue(void);
 
    void startSendingHeartbeat(void);
    void stopSendingAndCleanUpHeartbeat(void);
    void startListeningForControl(void);
    void stopListeningForControl(void);
    void interruptListeningForControl(void);
    void cleanUpControl(void);
    void startListeningForRunnerExit(void);
 
    void sendNodeExecutionStarted(const char *compositionIdentifier, const char *nodeIdentifier);
    void sendNodeExecutionFinished(const char *compositionIdentifier, const char *nodeIdentifier);
    void sendInputPortsUpdated(const char *compositionIdentifier, const char *portIdentifier, bool receivedEvent, bool receivedData, const char *portDataSummary);
    void sendOutputPortsUpdated(const char *compositionIdentifier, const char *portIdentifier, bool sentEvent, bool sentData, const char *portDataSummary);
    void sendPublishedOutputPortsUpdated(const char *portIdentifier, bool sentData, const char *portDataSummary);
    void sendEventFinished(unsigned long eventId, NodeContext *compositionContext);
    void sendEventDropped(const char *compositionIdentifier, const char *portIdentifier);
    void sendError(const char *message);
    void sendStopRequested(void);
    void sendCompositionStoppingAndCloseControl(void);
 
    bool shouldSendPortDataTelemetry(const char *compositionIdentifier, const char *portIdentifier);
};
 
extern "C"
{
void vuoSendNodeExecutionStarted(VuoCompositionState *compositionState, const char *nodeIdentifier);
void vuoSendNodeExecutionFinished(VuoCompositionState *compositionState, const char *nodeIdentifier);
void vuoSendInputPortsUpdated(VuoCompositionState *compositionState, const char *portIdentifier, bool receivedEvent, bool receivedData, const char *portDataSummary);
void vuoSendOutputPortsUpdated(VuoCompositionState *compositionState, const char *portIdentifier, bool sentEvent, bool sentData, const char *portDataSummary);
void vuoSendPublishedOutputPortsUpdated(VuoCompositionState *compositionState, const char *portIdentifier, bool sentData, const char *portDataSummary);
void vuoSendEventFinished(VuoCompositionState *compositionState, unsigned long eventId);
void vuoSendEventDropped(VuoCompositionState *compositionState, const char *portIdentifier);
bool vuoShouldSendPortDataTelemetry(VuoCompositionState *compositionState, const char *portIdentifier);
char * vuoGetInputPortString(VuoCompositionState *compositionState, const char *portIdentifier, bool shouldUseInterprocessSerialization);
char * vuoGetOutputPortString(VuoCompositionState *compositionState, const char *portIdentifier, bool shouldUseInterprocessSerialization);
}