EpicsConnection.cpp

Go to the documentation of this file.

#include <rms/provider/EpicsConnection.h>
#include <unistd.h>

namespace gov {

namespace fnal {

namespace cd {

namespace rms {

namespace provider {

/**
 * The character that is used a delimiter is PV names.
 */
const std::string EpicsConnection::PV_DELIMITER("/");

/**
 * Create a new EpicsConnection given the application name and partition.
 *
 * @param applicationName The name of the application generating/receiving
 * messages.
 * @param applicationPartition The partition of the application that is
 * generating/receiving messages.  
 */
EpicsConnection::EpicsConnection(const std::string applicationName,
                                 const int applicationPartition) {
    char hostname[20];

    ca_context_create(ca_enable_preemptive_callback);
    _caCurrentContext = ca_current_context();

    _messageSource.setProperty(base::RmsDestination::TARGET_PROPERTY_NAME, applicationName);
    _messageSource.setPartitionProperty(applicationPartition);

    if (gethostname(hostname, 20) == 0) {
        _uuidGenerator = new util::UUIDGenerator(hostname);
    }
    else {
        _uuidGenerator = new util::UUIDGenerator("C++EpicsConnection");
    }
}

/**
 * Destructor, call close() and have it clean up
 * everything.
 */
EpicsConnection::~EpicsConnection() {
    close();
}

/**
 * Determine if the given destination can be reached with
 * the EPICS provider.
 *
 * @param dest The destination to check to
 * see if messages can be sent to it.
 *
 * @return True if the destination can be reached, false otherwise.
 */
bool EpicsConnection::supportsDestination(const base::RmsDestination& dest) {
    std::string destinationPV;
    std::string monitoringOutboxPV;
    EpicsMessenger *pvHandle;
    
    destinationPV = getDestinationPVFromDest(dest);
    monitoringOutboxPV = getMonitoringPVFromDest(dest);

    pvHandle = getPVHandle(destinationPV);

    if (!pvHandle) {
        return false;
    }

    pvHandle = getPVHandle(monitoringOutboxPV);

    if (!pvHandle) {
        return false;
    }

    return true;
}

/**
 * Send a string to a particular destination.
 *
 * @param dest The destination that the message is going to
 * @param messageString The string that is being sent
 */
void EpicsConnection::sendString(const base::RmsDestination& dest, 
                                 const std::string& messageString) {
    std::string destinationPV;
    EpicsMessenger *pvHandle;
    
    destinationPV = getDestinationPVFromDest(dest);
    pvHandle = getPVHandle(destinationPV);

    if (pvHandle) {
        boost::mutex::scoped_lock lk(_epicsLock);
        pvHandle->sendMessage(messageString);
    }
    else {
        std::cout << "Unable to create PV handler for:" << std::endl;
        std::cout << destinationPV << std::endl;
        std::cout << "in EpicsConnection.sendMessage()" << std::endl;
    }

    return;
}

/**
 * Send an RmsMessage to a particular destination.
 *
 * @param dest The destination that the message is going to
 * @param message The RmsMessage that is being sent
 */
void EpicsConnection::sendMessage(const base::RmsDestination& dest, 
                                  const base::RmsMessage& message) {
    std::string destinationPV;
    std::string monitoringOutboxPV;
    EpicsMessenger *pvHandle;
   
    destinationPV = getDestinationPVFromDest(dest);
    monitoringOutboxPV = getMonitoringPVFromDest(message.getReplyDestination());
    std::string serializedMessage = message.serialize();

    pvHandle = getPVHandle(destinationPV);

    if (pvHandle) {
        boost::mutex::scoped_lock lk(_epicsLock);
        pvHandle->sendMessage(serializedMessage);
    }
    else {
        std::cout << "Unable to create PV handler for:" << std::endl;
        std::cout << destinationPV << std::endl;
        std::cout << "in EpicsConnection.sendMessage()" << std::endl;
        return;
    }

    pvHandle = getPVHandle(monitoringOutboxPV);

    if (pvHandle) {
        boost::mutex::scoped_lock lk(_epicsLock);
        pvHandle->sendMessage(serializedMessage);
    }
    else {
        std::cout << "Unable to create PV handler for:" << std::endl;
        std::cout << monitoringOutboxPV << std::endl;
        std::cout << "in EpicsConnection.sendMessage()" << std::endl;
        return;
    }

    return;
}

/**
 * Add a listener to a particular PV represented by a destination.
 *
 * @param dest The destination that we will be listening to.
 * @param listener The listener that will be called when messages
 * are posted to the PV.
 */
void EpicsConnection::addListener(const base::RmsDestination& dest, 
                                  ProviderListener *listener) {
    std::string destinationPV;
    EpicsMessenger *pvHandle;

    /**
     * If the target is empty, return the PV for the monitoring
     * outbox.
     */
    if (dest.getProperty(base::RmsDestination::TARGET_PROPERTY_NAME) != 
        std::string("")) {
        destinationPV = getDestinationPVFromDest(dest);
    } 
    else {
        destinationPV = getMonitoringPVFromDest(dest);
    }

    pvHandle = getPVHandle(destinationPV);
    
    if (pvHandle) {
        boost::mutex::scoped_lock lk(_epicsLock);
        pvHandle->addListener(listener);
    } 
    else {
        std::cout << "Unable to create PV handle for:" << std::endl;
        std::cout << destinationPV << std::endl;
        std::cout << "in EpicsConnection.addListener()" << std::endl;
    }
    
    return;
}

/**
 * Remove a listener from the EpicsConnection so that it no
 * longer receives messages.
 *
 * @param dest The destination to remove.
 * @param listener The listener to remove.
 */
void EpicsConnection::removeListener(const base::RmsDestination& dest, ProviderListener *listener) {
    std::string destinationPV;
    EpicsMessenger *pvHandle;

    /**
     * If the target is empty, return the PV for the monitoring
     * outbox.
     */
    if (dest.getProperty(base::RmsDestination::TARGET_PROPERTY_NAME) != 
        std::string("")) {
        destinationPV = getDestinationPVFromDest(dest);
    } 
    else {
        destinationPV = getMonitoringPVFromDest(dest);
    }
    
    pvHandle = getPVHandle(destinationPV);
    
    if (pvHandle) {
        boost::mutex::scoped_lock lk(_epicsLock);
        pvHandle->removeListener(listener);
    } 
    else {
        std::cout << "Unable to create PV handle for:" << std::endl;
        std::cout << destinationPV << std::endl;
        std::cout << "in EpicsConnection.removeListener()" << std::endl;
    }
    
    return;
}

/**
 * Try to connect to a PV to verify that it exists and is runing.
 *
 * @param pingDestination The destination that we are going to ping.
 *
 * @return True is the PV exists and is running, false otherwise.
 */
bool EpicsConnection::ping(const base::RmsDestination& pingDestination) {
    chid channelID;
    int result;

    std::string destinationPV = getDestinationPVFromDest(pingDestination);

    boost::mutex::scoped_lock lk(_epicsLock);
    ca_attach_context(_caCurrentContext);

    ca_create_channel(destinationPV.c_str(), NULL, NULL, 10, &channelID);
    result = ca_pend_io(0.5);
    ca_clear_channel(channelID);

    if (result != ECA_NORMAL) {
        return false;
    }

    return true;
}

/**
 * Free the memory allocated for the UUID generator and
 * delete any instances of the EpicsMessenger class that
 * were created.  Finally, call ca_context_destroy().
 */
void EpicsConnection::close() {
    std::map <std::string, EpicsMessenger *>::iterator cacheIterator;
    
    if (_uuidGenerator) {
        delete _uuidGenerator;
        _uuidGenerator = NULL;
    }

    boost::mutex::scoped_lock lk(_epicsLock);

    cacheIterator = _cachedPVs.begin();

    while (cacheIterator != _cachedPVs.end()) {
        delete cacheIterator->second;
        cacheIterator++;
    }
    
    _cachedPVs.clear();

    if (_caCurrentContext) {
        ca_attach_context(_caCurrentContext);
        ca_context_destroy();
        _caCurrentContext = NULL;
    }
}

/**
 * Translate a destination object into an Epics PV.
 *
 * @param dest The destination object that will be translated
 * into a PV.
 *
 * @return The name of the PV for the location that the destination
 * object represents.
 */
std::string 
EpicsConnection::getDestinationPVFromDest(const base::RmsDestination& dest) const {
    std::string destinationPV;
    
    std::string destinationTarget = 
        dest.getProperty(base::RmsDestination::TARGET_PROPERTY_NAME);
    std::string destinationChannel = 
        dest.getProperty(base::RmsDestination::CHANNEL_PROPERTY_NAME);
    std::string destinationPartition = 
        dest.getProperty(base::RmsDestination::PARTITION_PROPERTY_NAME);
    std::string destinationEnvironment = 
        dest.getProperty(base::RmsDestination::ENVIRONMENT_PROPERTY_NAME);

    /**
     * The heartbeat channel is a special case.  For each target, their will 
     * be one heartbeat channel.  All heartbeat messages will be pushed out 
     * to the outbox.
     */
    if(destinationChannel == base::RmsDestination::HEARTBEAT_CHANNEL) {
        if (destinationEnvironment != "") {
            destinationPV.append(destinationEnvironment + "_");
        }

        destinationPV.append(destinationTarget);
        destinationPV.append(EpicsConnection::PV_DELIMITER);
        destinationPV.append(destinationChannel);
        destinationPV.append(EpicsConnection::PV_DELIMITER);
        destinationPV.append("outbox");
        return destinationPV;
    }

    if (destinationEnvironment != "") {
        destinationPV.append(destinationEnvironment + "_");
    }
        
    destinationPV.append(destinationTarget);
    destinationPV.append(EpicsConnection::PV_DELIMITER);
    destinationPV.append(destinationPartition);
    destinationPV.append(EpicsConnection::PV_DELIMITER);
    destinationPV.append(destinationChannel);
    destinationPV.append(EpicsConnection::PV_DELIMITER);
    destinationPV.append("inbox");

    return destinationPV;
}

/**
 * Find the name of the monitoring outbox that a message should be
 * posted to when sending to a particular destination.
 *
 * @param dest The destination that the message is going to
 *
 * @return This will return the name of the PV for the
 * monitoring outbox that the message should be posted to.
 */
std::string 
EpicsConnection::getMonitoringPVFromDest(const base::RmsDestination& dest) const {
    std::string targetName = 
        dest.getProperty(base::RmsDestination::TARGET_PROPERTY_NAME);
    std::string destinationPartition = 
        dest.getProperty(base::RmsDestination::PARTITION_PROPERTY_NAME);
   std::string destinationChannel = 
        dest.getProperty(base::RmsDestination::CHANNEL_PROPERTY_NAME);

    /**
     * There is no monitoring outbox for the heartbeat channel
     */
    if (destinationChannel == base::RmsDestination::HEARTBEAT_CHANNEL) {
        return std::string("");
    }

    if (targetName == std::string("")) {
        targetName = dest.getProperty(base::RmsDestination::SOURCE_PROPERTY_NAME);
    }

    targetName.append(EpicsConnection::PV_DELIMITER);
    targetName.append(destinationPartition);
    targetName.append(EpicsConnection::PV_DELIMITER);
    targetName.append(destinationChannel);
    targetName.append(EpicsConnection::PV_DELIMITER);
    targetName.append("monitoringOutbox");
    
    std::string destinationEnvironment = 
        dest.getProperty(base::RmsDestination::ENVIRONMENT_PROPERTY_NAME);

    if (destinationEnvironment != "") {
        targetName.insert(0, destinationEnvironment + "_");
    }

    return targetName;
}

/**
 * Obtain a handle to a particular PV.  If the handle has already
 * been created, find it in the _cachedPVs table and return it.
 * Otherwise, create a new one and insert it into the _cachedPVs
 * table.
 *
 * @param pvName The name of the PV to get a handle for.
 *
 * @return An EpicsMessenger for a particular PV.
 */
EpicsMessenger *EpicsConnection::getPVHandle(std::string pvName) {
    EpicsMessenger *pvHandle;
    std::map <std::string, EpicsMessenger *>::iterator cacheIterator;

    boost::mutex::scoped_lock lk(_epicsLock);

    cacheIterator = _cachedPVs.find(pvName);

    if (cacheIterator == _cachedPVs.end()) {
        pvHandle = 
            new EpicsMessenger(pvName, _caCurrentContext, 
                               1024, _uuidGenerator);
        _cachedPVs.insert(std::pair<std::string, EpicsMessenger *>
                          (pvName, pvHandle));
    }
    else {
        return cacheIterator->second;
    }

    return pvHandle;
}
  
};  // end of namespace provider

};  // end of namespace rms

};  // end of namespace cd

};  // end of namespace fnal

};  // end of namespace gov