EpicsMessenger.cpp

Go to the documentation of this file.

#include <rms/provider/EpicsMessenger.h>

namespace gov {

namespace fnal {

namespace cd {

namespace rms {

namespace provider {

/**
 * Callback function for handling notifications from EPICS.  Calls the 
 * monitorChanged() method on the messenger object that is interited
 * in the PV that changed.
 *
 * @param eventHandlerArgs Parameters from EPICS
 */
static void 
EpicsMessengerCallback(struct event_handler_args eventHandlerArgs) {
    EpicsMessenger *messengerObject;
    char *messengerData;

    messengerObject = (EpicsMessenger *)eventHandlerArgs.usr;
    messengerData = (char *)eventHandlerArgs.dbr;
    
    messengerObject->monitorChanged(messengerData);
}

/**
 * Constructs an instance of this class for the specified
 * EPICS process variable (which is expected to a waveform record).
 *
 * @param pvName The name of the PV to connect to
 * @param caContext The EPICS context
 * @param maxSize The maximum size of messages that should be posted to the PV
 * @param uuidGenerator A UUID Generator to add UUIDs to messages
 */
EpicsMessenger::EpicsMessenger(std::string pvName, 
                               struct ca_client_context *caContext, 
                               int maxSize, 
                               util::UUIDGenerator *uuidGenerator) {
    int result;
    
    _pvName = pvName;
    _caContext = caContext;
    _maxMessageSize = maxSize;
    _uuidGenerator = uuidGenerator;

    ca_attach_context(_caContext);
    ca_create_channel(_pvName.c_str(), NULL, NULL, 10, &_pvChannel);
    result = ca_pend_io(_PEND_OP_TIMEOUT);
    ca_detach_context();
    
    if (result != ECA_NORMAL) {
        std::cout << "Failed to connect to " << pvName;
        std::cout << " in " << _PEND_OP_TIMEOUT;
        std::cout << " seconds..." << std::endl;
    }
    
    _firstListener = true;
}

/**
 * Destructor, just call close() which frees any allocated
 * memory and clears out open channels.
 */
EpicsMessenger::~EpicsMessenger() {
    close();
}

/**
 * Get the name of the PV that this class connects to.
 *
 * @return The name of the PV that this class connects to.
 */
std::string EpicsMessenger::getName() const {
    return _pvName;
}

/**
 * Post a message to the PV.  This method will split up
 * the message so that it can be reassembled by each
 * process that receives it.
 *
 * @param messageText The message to send
 */
void EpicsMessenger::sendMessage(const std::string& messageText) {
    std::vector <MessageFragment *> fragments;
    MessageSplitter splitter;
    char *fragment;
    const char *uuid;
    int size;

    uuid = _uuidGenerator->getUUID();
    
    fragments = splitter.splitMessage(messageText, _maxMessageSize, 
                                      uuid);

    delete[] uuid;

    ca_attach_context(_caContext);
    for (unsigned int i = 0; i < fragments.size(); i++) {
        fragments[i]->getBytes(&fragment, &size);
        ca_array_put(DBR_CHAR, size, _pvChannel, fragment);
        ca_pend_io(_PEND_OP_TIMEOUT);
        delete [] fragment;
        delete fragments[i];
    }
    ca_detach_context();
}

/**
 * Adds the specified listener to this messenger.
 *
 * @param listener The ProviderListener that will be
 * called when updates to this PV occur.
 */
void EpicsMessenger::addListener(ProviderListener *listener) {
    if (_firstListener) {
        ca_attach_context(_caContext);
        ca_create_subscription(DBR_CHAR, 0, _pvChannel, DBE_VALUE, 
                               EpicsMessengerCallback, (void *)this, &_pvSubscription);
        ca_pend_io(_PEND_OP_TIMEOUT);
        ca_detach_context();
        _firstListener = false;
    }
    
    for (unsigned int i = 0; i < _listenerList.size(); i++) {
        if (listener == _listenerList[i]) {
            return;
        }
    }
  
    _listenerList.push_back(listener);
}

/**
 * Removes the specified listener from this messenger.
 *
 * @param listener The listener to remove
 */
void EpicsMessenger::removeListener(ProviderListener *listener) {
    std::vector <ProviderListener *>::iterator i;

    i = _listenerList.begin();

    while (i != _listenerList.end()) {
        if (*i == listener) {
            _listenerList.erase(i);
            return;
        }
        i++;
    }
}

/**
 * Processes changes to the PV value.  This method is called by the
 * static callback function when the PV value changes.
 *
 * @param data The new value of the PV
 */
void EpicsMessenger::monitorChanged(char *data) {
    std::vector <MessageAssembler *>::iterator i;
    MessageFragment *fragment = new MessageFragment(data);

    i = _assemblerList.begin();

    while (i != _assemblerList.end()) {
        if ((*i)->addFragment(fragment)) {
            if ((*i)->hasCompleteMessage()) {
                std::string messageText = (*i)->getMessage();
                _assemblerList.erase(i);
                notifyListeners(messageText);
                delete *i;
            }
            
            return;
        }
        i++;
    }

    // Didn't find an assmbler for the message. Make a new one..
    MessageAssembler *newAssembler = new MessageAssembler(fragment);
    
    if (newAssembler->hasCompleteMessage()) {
        std::string messageText = newAssembler->getMessage();
        notifyListeners(messageText);    
        
        delete newAssembler;

        return;
    }
    
    _assemblerList.push_back(newAssembler);
}

/**
 * Go through all the listeners that have expressed
 * interest in this PV and post the message to them.
 *
 * @param messageText A message to send to all the listeners.
 */
void EpicsMessenger::notifyListeners(std::string messageText) {
    boost::shared_ptr<std::string> message(new std::string(messageText));

    for (unsigned int i = 0; i < _listenerList.size(); i++) {
        _listenerList[i]->textReceived(message);
    }
}

/**
 * Close down all the EPICS connections and free any
 * memory that has been allocated.  This includes all
 * assemblers, the channel access subscription and
 * the channel itself.
 */
void EpicsMessenger::close() {
    std::vector <MessageAssembler *>::iterator assemblerIterator;

    ca_attach_context(_caContext);

    ca_flush_io();
    ca_pend_io(_PEND_OP_TIMEOUT);

    if (!_firstListener) {
        ca_clear_subscription(_pvSubscription);
    }

    ca_clear_channel(_pvChannel);
    ca_detach_context();

    assemblerIterator = _assemblerList.begin();

    while (assemblerIterator != _assemblerList.end()) {
        delete *assemblerIterator;
        assemblerIterator++;
    }
}
                   
};  // end of namespace provider

};  // end of namespace rms

};  // end of namespace cd

};  // end of namespace fnal

};  // end of namespace gov