2.4.4/_vuo_runtime_communicator_8hh_source.html

#pragma once


#include "VuoCompositionState.h"

#include "VuoTelemetry.hh"


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);

}