http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/cf813d4d/src/FlowController.cpp ---------------------------------------------------------------------- diff --git a/src/FlowController.cpp b/src/FlowController.cpp index b18953e..2684b04 100644 --- a/src/FlowController.cpp +++ b/src/FlowController.cpp @@ -27,17 +27,17 @@ #include <thread> #include <libxml/parser.h> #include <libxml/tree.h> +#include <yaml-cpp/yaml.h> #include "FlowController.h" #include "ProcessContext.h" -FlowController::FlowController(std::string name) -: _name(name) -{ +FlowController::FlowController(std::string name) : + _name(name) { uuid_generate(_uuid); // Setup the default values - _xmlFileName = DEFAULT_FLOW_XML_FILE_NAME; + _configurationFileName = DEFAULT_FLOW_XML_FILE_NAME; _maxEventDrivenThreads = DEFAULT_MAX_EVENT_DRIVEN_THREAD; _maxTimerDrivenThreads = DEFAULT_MAX_TIMER_DRIVEN_THREAD; _running = false; @@ -48,34 +48,29 @@ FlowController::FlowController(std::string name) // NiFi config properties _configure = Configure::getConfigure(); - _configure->get(Configure::nifi_flow_configuration_file, _xmlFileName); - _logger->log_info("FlowController NiFi XML file %s", _xmlFileName.c_str()); + _configure->get(Configure::nifi_flow_configuration_file, _configurationFileName); + _logger->log_info("FlowController NiFi XML file %s", _configurationFileName.c_str()); // Create repos for flow record and provenance _logger->log_info("FlowController %s created", _name.c_str()); } -FlowController::~FlowController() -{ +FlowController::~FlowController() { stop(true); unload(); delete _protocol; } -bool FlowController::isRunning() -{ +bool FlowController::isRunning() { return (_running); } -bool FlowController::isInitialized() -{ +bool FlowController::isInitialized() { return (_initialized); } -void FlowController::stop(bool force) -{ - if (_running) - { +void FlowController::stop(bool force) { + if (_running) { _logger->log_info("Stop Flow Controller"); this->_timerScheduler.stop(); // Wait for sometime for thread stop @@ -86,14 +81,11 @@ void FlowController::stop(bool force) } } -void FlowController::unload() -{ - if (_running) - { +void FlowController::unload() { + if (_running) { stop(true); } - if (_initialized) - { + if (_initialized) { _logger->log_info("Unload Flow Controller"); if (_root) delete _root; @@ -105,43 +97,34 @@ void FlowController::unload() return; } -void FlowController::reload(std::string xmlFile) -{ +void FlowController::reload(std::string xmlFile) { _logger->log_info("Starting to reload Flow Controller with xml %s", xmlFile.c_str()); stop(true); unload(); - std::string oldxmlFile = this->_xmlFileName; - this->_xmlFileName = xmlFile; - load(); + std::string oldxmlFile = this->_configurationFileName; + this->_configurationFileName = xmlFile; + load(ConfigFormat::XML); start(); - if (!this->_root) - { - this->_xmlFileName = oldxmlFile; + if (!this->_root) { + this->_configurationFileName = oldxmlFile; _logger->log_info("Rollback Flow Controller to xml %s", oldxmlFile.c_str()); stop(true); unload(); - load(); + load(ConfigFormat::XML); start(); } } -Processor *FlowController::createProcessor(std::string name, uuid_t uuid) -{ +Processor *FlowController::createProcessor(std::string name, uuid_t uuid) { Processor *processor = NULL; - if (name == GenerateFlowFile::ProcessorName) - { + _logger->log_trace("Received request to create a processor for name %s", name.c_str()); + if (name == GenerateFlowFile::ProcessorName) { processor = new GenerateFlowFile(name, uuid); - } - else if (name == LogAttribute::ProcessorName) - { + } else if (name == LogAttribute::ProcessorName) { processor = new LogAttribute(name, uuid); - } - else if (name == RealTimeDataCollector::ProcessorName) - { + } else if (name == RealTimeDataCollector::ProcessorName) { processor = new RealTimeDataCollector(name, uuid); - } - else - { + } else { _logger->log_error("No Processor defined for %s", name.c_str()); return NULL; } @@ -152,29 +135,24 @@ Processor *FlowController::createProcessor(std::string name, uuid_t uuid) return processor; } -ProcessGroup *FlowController::createRootProcessGroup(std::string name, uuid_t uuid) -{ +ProcessGroup *FlowController::createRootProcessGroup(std::string name, uuid_t uuid) { return new ProcessGroup(ROOT_PROCESS_GROUP, name, uuid); } -ProcessGroup *FlowController::createRemoteProcessGroup(std::string name, uuid_t uuid) -{ +ProcessGroup *FlowController::createRemoteProcessGroup(std::string name, uuid_t uuid) { return new ProcessGroup(REMOTE_PROCESS_GROUP, name, uuid); } -Connection *FlowController::createConnection(std::string name, uuid_t uuid) -{ +Connection *FlowController::createConnection(std::string name, uuid_t uuid) { return new Connection(name, uuid); } -void FlowController::parseConnection(xmlDoc *doc, xmlNode *node, ProcessGroup *parent) -{ +void FlowController::parseConnection(xmlDoc *doc, xmlNode *node, ProcessGroup *parent) { uuid_t uuid; xmlNode *currentNode; Connection *connection = NULL; - if (!parent) - { + if (!parent) { _logger->log_error("parseProcessNode: no parent group existed"); return; } @@ -182,67 +160,49 @@ void FlowController::parseConnection(xmlDoc *doc, xmlNode *node, ProcessGroup *p // generate the random UIID uuid_generate(uuid); - for (currentNode = node->xmlChildrenNode; currentNode != NULL; currentNode = currentNode->next) - { - if (currentNode->type == XML_ELEMENT_NODE) - { - if (xmlStrcmp(currentNode->name, BAD_CAST "id") == 0) - { + for (currentNode = node->xmlChildrenNode; currentNode != NULL; currentNode = currentNode->next) { + if (currentNode->type == XML_ELEMENT_NODE) { + if (xmlStrcmp(currentNode->name, BAD_CAST "id") == 0) { char *id = (char *) xmlNodeGetContent(currentNode); - if (id) - { + if (id) { _logger->log_debug("parseConnection: id => [%s]", id); uuid_parse(id, uuid); xmlFree(id); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "name") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "name") == 0) { char *name = (char *) xmlNodeGetContent(currentNode); - if (name) - { + if (name) { _logger->log_debug("parseConnection: name => [%s]", name); connection = this->createConnection(name, uuid); - if (connection == NULL) - { + if (connection == NULL) { xmlFree(name); return; } xmlFree(name); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "sourceId") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "sourceId") == 0) { char *id = (char *) xmlNodeGetContent(currentNode); - if (id) - { + if (id) { _logger->log_debug("parseConnection: sourceId => [%s]", id); uuid_parse(id, uuid); xmlFree(id); if (connection) connection->setSourceProcessorUUID(uuid); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "destinationId") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "destinationId") == 0) { char *id = (char *) xmlNodeGetContent(currentNode); - if (id) - { + if (id) { _logger->log_debug("parseConnection: destinationId => [%s]", id); uuid_parse(id, uuid); xmlFree(id); if (connection) connection->setDestinationProcessorUUID(uuid); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "maxWorkQueueSize") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "maxWorkQueueSize") == 0) { char *temp = (char *) xmlNodeGetContent(currentNode); int64_t maxWorkQueueSize = 0; - if (temp) - { - if (Property::StringToInt(temp, maxWorkQueueSize)) - { + if (temp) { + if (Property::StringToInt(temp, maxWorkQueueSize)) { _logger->log_debug("parseConnection: maxWorkQueueSize => [%d]", maxWorkQueueSize); if (connection) connection->setMaxQueueSize(maxWorkQueueSize); @@ -250,15 +210,11 @@ void FlowController::parseConnection(xmlDoc *doc, xmlNode *node, ProcessGroup *p } xmlFree(temp); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "maxWorkQueueDataSize") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "maxWorkQueueDataSize") == 0) { char *temp = (char *) xmlNodeGetContent(currentNode); int64_t maxWorkQueueDataSize = 0; - if (temp) - { - if (Property::StringToInt(temp, maxWorkQueueDataSize)) - { + if (temp) { + if (Property::StringToInt(temp, maxWorkQueueDataSize)) { _logger->log_debug("parseConnection: maxWorkQueueDataSize => [%d]", maxWorkQueueDataSize); if (connection) connection->setMaxQueueDataSize(maxWorkQueueDataSize); @@ -266,22 +222,16 @@ void FlowController::parseConnection(xmlDoc *doc, xmlNode *node, ProcessGroup *p } xmlFree(temp); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "relationship") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "relationship") == 0) { char *temp = (char *) xmlNodeGetContent(currentNode); - if (temp) - { + if (temp) { std::string relationshipName = temp; - if (!relationshipName.empty()) - { + if (!relationshipName.empty()) { Relationship relationship(relationshipName, ""); _logger->log_debug("parseConnection: relationship => [%s]", relationshipName.c_str()); if (connection) connection->setRelationship(relationship); - } - else - { + } else { Relationship empty; _logger->log_debug("parseConnection: relationship => [%s]", empty.getName().c_str()); if (connection) @@ -299,8 +249,7 @@ void FlowController::parseConnection(xmlDoc *doc, xmlNode *node, ProcessGroup *p return; } -void FlowController::parseRootProcessGroup(xmlDoc *doc, xmlNode *node) -{ +void FlowController::parseRootProcessGroup(xmlDoc *doc, xmlNode *node) { uuid_t uuid; xmlNode *currentNode; ProcessGroup *group = NULL; @@ -308,29 +257,21 @@ void FlowController::parseRootProcessGroup(xmlDoc *doc, xmlNode *node) // generate the random UIID uuid_generate(uuid); - for (currentNode = node->xmlChildrenNode; currentNode != NULL; currentNode = currentNode->next) - { - if (currentNode->type == XML_ELEMENT_NODE) - { - if (xmlStrcmp(currentNode->name, BAD_CAST "id") == 0) - { + for (currentNode = node->xmlChildrenNode; currentNode != NULL; currentNode = currentNode->next) { + if (currentNode->type == XML_ELEMENT_NODE) { + if (xmlStrcmp(currentNode->name, BAD_CAST "id") == 0) { char *id = (char *) xmlNodeGetContent(currentNode); - if (id) - { + if (id) { _logger->log_debug("parseRootProcessGroup: id => [%s]", id); uuid_parse(id, uuid); xmlFree(id); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "name") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "name") == 0) { char *name = (char *) xmlNodeGetContent(currentNode); - if (name) - { + if (name) { _logger->log_debug("parseRootProcessGroup: name => [%s]", name); group = this->createRootProcessGroup(name, uuid); - if (group == NULL) - { + if (group == NULL) { xmlFree(name); return; } @@ -339,57 +280,388 @@ void FlowController::parseRootProcessGroup(xmlDoc *doc, xmlNode *node) this->_name = name; xmlFree(name); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "processor") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "processor") == 0) { this->parseProcessorNode(doc, currentNode, group); - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "connection") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "connection") == 0) { this->parseConnection(doc, currentNode, group); - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "remoteProcessGroup") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "remoteProcessGroup") == 0) { this->parseRemoteProcessGroup(doc, currentNode, group); } } // if (currentNode->type == XML_ELEMENT_NODE) - } // for node + } // for node +} + +void FlowController::parseRootProcessGroupYaml(YAML::Node rootFlowNode) { + uuid_t uuid; + ProcessGroup *group = NULL; + + // generate the random UIID + uuid_generate(uuid); + + std::string flowName = rootFlowNode["name"].as<std::string>(); + + char uuidStr[37]; + uuid_unparse(_uuid, uuidStr); + _logger->log_debug("parseRootProcessGroup: id => [%s]", uuidStr); + _logger->log_debug("parseRootProcessGroup: name => [%s]", flowName.c_str()); + group = this->createRootProcessGroup(flowName, uuid); + this->_root = group; + this->_name = flowName; +} + +void FlowController::parseProcessorNodeYaml(YAML::Node processorsNode, ProcessGroup *parentGroup) { + int64_t schedulingPeriod = -1; + int64_t penalizationPeriod = -1; + int64_t yieldPeriod = -1; + int64_t runDurationNanos = -1; + uuid_t uuid; + Processor *processor = NULL; + + if (!parentGroup) { + _logger->log_error("parseProcessNodeYaml: no parent group exists"); + return; + } + + if (processorsNode) { + // Evaluate sequence of processors + int numProcessors = processorsNode.size(); + if (numProcessors < 1) { + throw new std::invalid_argument("There must be at least one processor configured."); + } + + std::vector<ProcessorConfig> processorConfigs; + + if (processorsNode.IsSequence()) { + for (YAML::const_iterator iter = processorsNode.begin(); iter != processorsNode.end(); ++iter) { + ProcessorConfig procCfg; + YAML::Node procNode = iter->as<YAML::Node>(); + + procCfg.name = procNode["name"].as<std::string>(); + _logger->log_debug("parseProcessorNode: name => [%s]", procCfg.name.c_str()); + procCfg.javaClass = procNode["class"].as<std::string>(); + _logger->log_debug("parseProcessorNode: class => [%s]", procCfg.javaClass.c_str()); + + char uuidStr[37]; + uuid_unparse(_uuid, uuidStr); + + // generate the random UUID + uuid_generate(uuid); + + // Determine the processor name only from the Java class + int lastOfIdx = procCfg.javaClass.find_last_of("."); + if (lastOfIdx != std::string::npos) { + lastOfIdx++; // if a value is found, increment to move beyond the . + int nameLength = procCfg.javaClass.length() - lastOfIdx; + std::string processorName = procCfg.javaClass.substr(lastOfIdx, nameLength); + processor = this->createProcessor(processorName, uuid); + } + + if (!processor) { + _logger->log_error("Could not create a processor %s with name %s", procCfg.name.c_str(), uuidStr); + throw std::invalid_argument("Could not create processor " + procCfg.name); + } + processor->setName(procCfg.name); + + procCfg.maxConcurrentTasks = procNode["max concurrent tasks"].as<std::string>(); + _logger->log_debug("parseProcessorNode: max concurrent tasks => [%s]", procCfg.maxConcurrentTasks.c_str()); + procCfg.schedulingStrategy = procNode["scheduling strategy"].as<std::string>(); + _logger->log_debug("parseProcessorNode: scheduling strategy => [%s]", + procCfg.schedulingStrategy.c_str()); + procCfg.schedulingPeriod = procNode["scheduling period"].as<std::string>(); + _logger->log_debug("parseProcessorNode: scheduling period => [%s]", procCfg.schedulingPeriod.c_str()); + procCfg.penalizationPeriod = procNode["penalization period"].as<std::string>(); + _logger->log_debug("parseProcessorNode: penalization period => [%s]", + procCfg.penalizationPeriod.c_str()); + procCfg.yieldPeriod = procNode["yield period"].as<std::string>(); + _logger->log_debug("parseProcessorNode: yield period => [%s]", procCfg.yieldPeriod.c_str()); + procCfg.yieldPeriod = procNode["run duration nanos"].as<std::string>(); + _logger->log_debug("parseProcessorNode: run duration nanos => [%s]", procCfg.runDurationNanos.c_str()); + + // handle auto-terminated relationships + YAML::Node autoTerminatedSequence = procNode["auto-terminated relationships list"]; + std::vector<std::string> rawAutoTerminatedRelationshipValues; + if (autoTerminatedSequence.IsSequence() && !autoTerminatedSequence.IsNull() + && autoTerminatedSequence.size() > 0) { + _logger->log_debug("Found non-empty auto terminated sequence... interpreting."); + for (YAML::const_iterator relIter = autoTerminatedSequence.begin(); + relIter != autoTerminatedSequence.end(); ++relIter) { + std::string autoTerminatedRel = relIter->as<std::string>(); + _logger->log_debug("Auto terminating relationship %s", autoTerminatedRel.c_str()); + rawAutoTerminatedRelationshipValues.push_back(autoTerminatedRel); + } + } else { + _logger->log_debug("no relationships are auto terminated here..."); + } + procCfg.autoTerminatedRelationships = rawAutoTerminatedRelationshipValues; + + // handle processor properties + YAML::Node propertiesNode = procNode["Properties"]; + std::vector<Property> properties; + + if (propertiesNode.IsMap() && !propertiesNode.IsNull() && propertiesNode.size() > 0) { + std::map<std::string, std::string> propertiesMap = propertiesNode.as< + std::map<std::string, std::string>>(); + for (std::map<std::string, std::string>::iterator propsIter = propertiesMap.begin(); + propsIter != propertiesMap.end(); propsIter++) { + std::string propertyName = propsIter->first; + std::string propertyValue = propsIter->second; + if (!processor->setProperty(propertyName, propertyValue)) { + _logger->log_warn( + "Received property %s with value %s but is not one of the properties for %s", + propertyName.c_str(), propertyValue.c_str(), procCfg.name.c_str()); + } + } + } + + // Take care of scheduling + TimeUnit unit; + if (Property::StringToTime(procCfg.schedulingPeriod, schedulingPeriod, unit) + && Property::ConvertTimeUnitToNS(schedulingPeriod, unit, schedulingPeriod)) { + _logger->log_debug("convert: parseProcessorNode: schedulingPeriod => [%d] ns", schedulingPeriod); + processor->setSchedulingPeriodNano(schedulingPeriod); + } + + if (Property::StringToTime(procCfg.penalizationPeriod, penalizationPeriod, unit) + && Property::ConvertTimeUnitToMS(penalizationPeriod, unit, penalizationPeriod)) { + _logger->log_debug("convert: parseProcessorNode: penalizationPeriod => [%d] ms", + penalizationPeriod); + processor->setPenalizationPeriodMsec(penalizationPeriod); + } + + if (Property::StringToTime(procCfg.yieldPeriod, yieldPeriod, unit) + && Property::ConvertTimeUnitToMS(yieldPeriod, unit, yieldPeriod)) { + _logger->log_debug("convert: parseProcessorNode: yieldPeriod => [%d] ms", yieldPeriod); + processor->setYieldPeriodMsec(yieldPeriod); + } + + // Default to running + processor->setScheduledState(RUNNING); + + if (procCfg.schedulingStrategy == "TIMER_DRIVEN") { + processor->setSchedulingStrategy(TIMER_DRIVEN); + _logger->log_debug("setting scheduling strategy as %s", procCfg.schedulingStrategy.c_str()); + } else if (procCfg.schedulingStrategy == "EVENT_DRIVEN") { + processor->setSchedulingStrategy(EVENT_DRIVEN); + _logger->log_debug("setting scheduling strategy as %s", procCfg.schedulingStrategy.c_str()); + } else { + processor->setSchedulingStrategy(CRON_DRIVEN); + _logger->log_debug("setting scheduling strategy as %s", procCfg.schedulingStrategy.c_str()); + + } + + int64_t maxConcurrentTasks; + if (Property::StringToInt(procCfg.maxConcurrentTasks, maxConcurrentTasks)) { + _logger->log_debug("parseProcessorNode: maxConcurrentTasks => [%d]", maxConcurrentTasks); + processor->setMaxConcurrentTasks(maxConcurrentTasks); + } + + if (Property::StringToInt(procCfg.runDurationNanos, runDurationNanos)) { + _logger->log_debug("parseProcessorNode: runDurationNanos => [%d]", runDurationNanos); + processor->setRunDurationNano(runDurationNanos); + } + + std::set<Relationship> autoTerminatedRelationships; + for (auto&& relString : procCfg.autoTerminatedRelationships) { + Relationship relationship(relString, ""); + _logger->log_debug("parseProcessorNode: autoTerminatedRelationship => [%s]", relString.c_str()); + autoTerminatedRelationships.insert(relationship); + } + + processor->setAutoTerminatedRelationships(autoTerminatedRelationships); + + parentGroup->addProcessor(processor); + } + } + } else { + throw new std::invalid_argument( + "Cannot instantiate a MiNiFi instance without a defined Processors configuration node."); + } +} + +void FlowController::parseRemoteProcessGroupYaml(YAML::Node *rpgNode, ProcessGroup *parentGroup) { + uuid_t uuid; + + if (!parentGroup) { + _logger->log_error("parseRemoteProcessGroupYaml: no parent group exists"); + return; + } + + if (rpgNode) { + if (rpgNode->IsSequence()) { + for (YAML::const_iterator iter = rpgNode->begin(); iter != rpgNode->end(); ++iter) { + YAML::Node rpgNode = iter->as<YAML::Node>(); + + auto name = rpgNode["name"].as<std::string>(); + _logger->log_debug("parseRemoteProcessGroupYaml: name => [%s]", name.c_str()); + + std::string url = rpgNode["url"].as<std::string>(); + _logger->log_debug("parseRemoteProcessGroupYaml: url => [%s]", url.c_str()); + + std::string timeout = rpgNode["timeout"].as<std::string>(); + _logger->log_debug("parseRemoteProcessGroupYaml: timeout => [%s]", timeout.c_str()); + + std::string yieldPeriod = rpgNode["yield period"].as<std::string>(); + _logger->log_debug("parseRemoteProcessGroupYaml: timeout => [%s]", yieldPeriod.c_str()); + + YAML::Node inputPorts = rpgNode["Input Ports"].as<YAML::Node>(); + + if (inputPorts.IsSequence()) { + for (YAML::const_iterator portIter = inputPorts.begin(); portIter != inputPorts.end(); ++portIter) { + _logger->log_debug("Got a current port, iterating..."); + + YAML::Node currPort = portIter->as<YAML::Node>(); + + this->parsePortYaml(&currPort, parentGroup, SEND); + + } // for node + char uuidStr[37]; + uuid_unparse(_uuid, uuidStr); + + // generate the random UUID + uuid_generate(uuid); + + int64_t timeoutValue = -1; + int64_t yieldPeriodValue = -1; + + ProcessGroup* group = this->createRemoteProcessGroup(name.c_str(), uuid); + group->setParent(parentGroup); + parentGroup->addProcessGroup(group); + + TimeUnit unit; + + if (Property::StringToTime(yieldPeriod, yieldPeriodValue, unit) + && Property::ConvertTimeUnitToMS(yieldPeriodValue, unit, yieldPeriodValue) && group) { + _logger->log_debug("parseRemoteProcessGroup: yieldPeriod => [%d] ms", yieldPeriod.c_str()); + group->setYieldPeriodMsec(yieldPeriodValue); + } + + if (Property::StringToTime(timeout, timeoutValue, unit) + && Property::ConvertTimeUnitToMS(timeoutValue, unit, timeoutValue) && group) { + _logger->log_debug("parseRemoteProcessGroup: timeoutValue => [%d] ms", timeout.c_str()); + group->setTimeOut(yieldPeriodValue); + } + + group->setTransmitting(true); + group->setURL(url); + + } + } + } + } } -void FlowController::parseRemoteProcessGroup(xmlDoc *doc, xmlNode *node, ProcessGroup *parent) -{ +void FlowController::parseConnectionYaml(YAML::Node *connectionsNode, ProcessGroup *parent) { + uuid_t uuid; + Connection *connection = NULL; + + if (!parent) { + _logger->log_error("parseProcessNode: no parent group was provided"); + return; + } + + if (connectionsNode) { + int numConnections = connectionsNode->size(); + if (numConnections < 1) { + throw new std::invalid_argument("There must be at least one connection configured."); + } + + if (connectionsNode->IsSequence()) { + for (YAML::const_iterator iter = connectionsNode->begin(); iter != connectionsNode->end(); ++iter) { +// generate the random UIID + uuid_generate(uuid); + + YAML::Node connectionNode = iter->as<YAML::Node>(); + + std::string name = connectionNode["name"].as<std::string>(); + std::string destName = connectionNode["destination name"].as<std::string>(); + + char uuidStr[37]; + uuid_unparse(_uuid, uuidStr); + + _logger->log_debug("Created connection with UUID %s and name %s", uuidStr, name.c_str()); + connection = this->createConnection(name, uuid); + auto rawRelationship = connectionNode["source relationship name"].as<std::string>(); + Relationship relationship(rawRelationship, ""); + _logger->log_debug("parseConnection: relationship => [%s]", rawRelationship.c_str()); + if (connection) + connection->setRelationship(relationship); + std::string connectionSrcProcName = connectionNode["source name"].as<std::string>(); + Processor *srcProcessor = this->_root->findProcessor(connectionSrcProcName); + _logger->log_debug("I see processor with name %s looking for source with name %s", + this->_root->findProcessor(connectionSrcProcName)->getName().c_str(), + connectionSrcProcName.c_str()); + if (!srcProcessor) { + _logger->log_error("Could not locate a source with name %s to create a connection", + connectionSrcProcName.c_str()); + throw std::invalid_argument( + "Could not locate a source with name %s to create a connection " + connectionSrcProcName); + } + + _logger->log_debug("This processor has UUID of %s", srcProcessor->getUUIDStr().c_str()); + _logger->log_trace("Trying to find dest processor by name %s", destName.c_str()); + Processor *destProcessor = this->_root->findProcessor(destName); + // If we could not find name, try by UUID + if (!destProcessor) { + _logger->log_trace("Now looking up by uuid"); + uuid_t destUuid; + uuid_parse(destName.c_str(), destUuid); + destProcessor = this->_root->findProcessor(destUuid); + } + if (destProcessor) { + std::string destUuid = destProcessor->getUUIDStr(); + if (!destUuid.empty()) { + _logger->log_debug("This destination processor has UUID of %s", + destProcessor->getUUIDStr().c_str()); + } + } else { + _logger->log_debug("!!! === Could not find a destination processor for the connection."); + } + + uuid_t srcUuid; + uuid_t destUuid; + srcProcessor->getUUID(srcUuid); + connection->setSourceProcessorUUID(srcUuid); + destProcessor->getUUID(destUuid); + connection->setDestinationProcessorUUID(destUuid); + + if (connection) { + parent->addConnection(connection); + } + } + } + + if (connection) + parent->addConnection(connection); + + return; + } +} + +void FlowController::parseRemoteProcessGroup(xmlDoc *doc, xmlNode *node, ProcessGroup *parent) { uuid_t uuid; xmlNode *currentNode; ProcessGroup *group = NULL; int64_t yieldPeriod = -1; int64_t timeOut = -1; - // generate the random UIID +// generate the random UIID uuid_generate(uuid); - for (currentNode = node->xmlChildrenNode; currentNode != NULL; currentNode = currentNode->next) - { - if (currentNode->type == XML_ELEMENT_NODE) - { - if (xmlStrcmp(currentNode->name, BAD_CAST "id") == 0) - { + for (currentNode = node->xmlChildrenNode; currentNode != NULL; currentNode = currentNode->next) { + if (currentNode->type == XML_ELEMENT_NODE) { + if (xmlStrcmp(currentNode->name, BAD_CAST "id") == 0) { char *id = (char *) xmlNodeGetContent(currentNode); - if (id) - { + if (id) { _logger->log_debug("parseRootProcessGroup: id => [%s]", id); uuid_parse(id, uuid); xmlFree(id); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "name") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "name") == 0) { char *name = (char *) xmlNodeGetContent(currentNode); - if (name) - { + if (name) { _logger->log_debug("parseRemoteProcessGroup: name => [%s]", name); group = this->createRemoteProcessGroup(name, uuid); - if (group == NULL) - { + if (group == NULL) { xmlFree(name); return; } @@ -397,109 +669,152 @@ void FlowController::parseRemoteProcessGroup(xmlDoc *doc, xmlNode *node, Process parent->addProcessGroup(group); xmlFree(name); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "yieldPeriod") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "yieldPeriod") == 0) { TimeUnit unit; char *temp = (char *) xmlNodeGetContent(currentNode); - if (temp) - { - if (Property::StringToTime(temp, yieldPeriod, unit) && - Property::ConvertTimeUnitToMS(yieldPeriod, unit, yieldPeriod) && group) - { + if (temp) { + if (Property::StringToTime(temp, yieldPeriod, unit) + && Property::ConvertTimeUnitToMS(yieldPeriod, unit, yieldPeriod) && group) { _logger->log_debug("parseRemoteProcessGroup: yieldPeriod => [%d] ms", yieldPeriod); group->setYieldPeriodMsec(yieldPeriod); } xmlFree(temp); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "timeout") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "timeout") == 0) { TimeUnit unit; char *temp = (char *) xmlNodeGetContent(currentNode); - if (temp) - { - if (Property::StringToTime(temp, timeOut, unit) && - Property::ConvertTimeUnitToMS(timeOut, unit, timeOut) && group) - { + if (temp) { + if (Property::StringToTime(temp, timeOut, unit) + && Property::ConvertTimeUnitToMS(timeOut, unit, timeOut) && group) { _logger->log_debug("parseRemoteProcessGroup: timeOut => [%d] ms", timeOut); group->setTimeOut(timeOut); } xmlFree(temp); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "transmitting") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "transmitting") == 0) { char *temp = (char *) xmlNodeGetContent(currentNode); bool transmitting; - if (temp) - { - if (Property::StringToBool(temp, transmitting) && group) - { + if (temp) { + if (Property::StringToBool(temp, transmitting) && group) { _logger->log_debug("parseRemoteProcessGroup: transmitting => [%d]", transmitting); group->setTransmitting(transmitting); } xmlFree(temp); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "inputPort") == 0 && group) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "inputPort") == 0 && group) { this->parsePort(doc, currentNode, group, SEND); - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "outputPort") == 0 && group) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "outputPort") == 0 && group) { this->parsePort(doc, currentNode, group, RECEIVE); } } // if (currentNode->type == XML_ELEMENT_NODE) - } // for node + } // for node } -void FlowController::parseProcessorProperty(xmlDoc *doc, xmlNode *node, Processor *processor) -{ +void FlowController::parseProcessorProperty(xmlDoc *doc, xmlNode *node, Processor *processor) { xmlNode *currentNode; std::string propertyValue; std::string propertyName; - if (!processor) - { + if (!processor) { _logger->log_error("parseProcessorProperty: no parent processor existed"); return; } - for (currentNode = node->xmlChildrenNode; currentNode != NULL; currentNode = currentNode->next) - { - if (currentNode->type == XML_ELEMENT_NODE) - { - if (xmlStrcmp(currentNode->name, BAD_CAST "name") == 0) - { + for (currentNode = node->xmlChildrenNode; currentNode != NULL; currentNode = currentNode->next) { + if (currentNode->type == XML_ELEMENT_NODE) { + if (xmlStrcmp(currentNode->name, BAD_CAST "name") == 0) { char *name = (char *) xmlNodeGetContent(currentNode); - if (name) - { + if (name) { _logger->log_debug("parseProcessorNode: name => [%s]", name); propertyName = name; xmlFree(name); } } - if (xmlStrcmp(currentNode->name, BAD_CAST "value") == 0) - { + if (xmlStrcmp(currentNode->name, BAD_CAST "value") == 0) { char *value = (char *) xmlNodeGetContent(currentNode); - if (value) - { + if (value) { _logger->log_debug("parseProcessorNode: value => [%s]", value); propertyValue = value; xmlFree(value); } } - if (!propertyName.empty() && !propertyValue.empty()) - { + if (!propertyName.empty() && !propertyValue.empty()) { processor->setProperty(propertyName, propertyValue); } } // if (currentNode->type == XML_ELEMENT_NODE) - } // for node + } // for node +} + +void FlowController::parsePortYaml(YAML::Node *portNode, ProcessGroup *parent, TransferDirection direction) { + uuid_t uuid; + Processor *processor = NULL; + RemoteProcessorGroupPort *port = NULL; + + _logger->log_trace("Creating a port from YAML."); + + if (!parent) { + _logger->log_error("parseProcessNode: no parent group existed"); + return; + } + + YAML::Node inputPortsObj = portNode->as<YAML::Node>(); + +// generate the random UIID + uuid_generate(uuid); + + auto portId = inputPortsObj["id"].as<std::string>(); + auto nameStr = inputPortsObj["name"].as<std::string>(); + uuid_parse(portId.c_str(), uuid); + + port = new RemoteProcessorGroupPort(nameStr.c_str(), uuid); + _logger->log_debug("parse input port: name => [%s]", nameStr.c_str()); + _logger->log_debug("parse input port: id => [%s]", portId.c_str()); + + processor = (Processor *) port; + port->setDirection(direction); + port->setTimeOut(parent->getTimeOut()); + port->setTransmitting(true); + processor->setYieldPeriodMsec(parent->getYieldPeriodMsec()); + processor->initialize(); + + // handle port properties + YAML::Node nodeVal = portNode->as<YAML::Node>(); + YAML::Node propertiesNode = nodeVal["Properties"]; + std::vector<Property> properties; + + _logger->log_debug("!!! === Checking out properties for input port...."); + + if (propertiesNode.IsMap() && !propertiesNode.IsNull() && propertiesNode.size() > 0) { + std::map<std::string, std::string> propertiesMap = propertiesNode.as<std::map<std::string, std::string>>(); + _logger->log_debug("Found non-empty properties sequence... interpreting."); + for (std::map<std::string, std::string>::iterator propsIter = propertiesMap.begin(); + propsIter != propertiesMap.end(); propsIter++) { + std::string propertyName = propsIter->first; + std::string propertyValue = propsIter->second; + _logger->log_debug("Detected property %s => %s", propertyName.c_str(), propertyValue.c_str()); + if (!processor->setProperty(propertyName, propertyValue)) { + _logger->log_warn("Received property %s with value %s but is not one of the properties for %s", + propertyName.c_str(), propertyValue.c_str(), nameStr.c_str()); + } + } + } else { + _logger->log_debug("no properties here..."); + } + + // add processor to parent + parent->addProcessor(processor); + processor->setScheduledState(RUNNING); + auto rawMaxConcurrentTasks = inputPortsObj["max concurrent tasks"].as<std::string>(); + int64_t maxConcurrentTasks; + if (Property::StringToInt(rawMaxConcurrentTasks, maxConcurrentTasks)) { + processor->setMaxConcurrentTasks(maxConcurrentTasks); + } + _logger->log_debug("parseProcessorNode: maxConcurrentTasks => [%d]", maxConcurrentTasks); + processor->setMaxConcurrentTasks(maxConcurrentTasks); + } -void FlowController::parsePort(xmlDoc *doc, xmlNode *processorNode, ProcessGroup *parent, TransferDirection direction) -{ +void FlowController::parsePort(xmlDoc *doc, xmlNode *processorNode, ProcessGroup *parent, TransferDirection direction) { char *id = NULL; char *name = NULL; uuid_t uuid; @@ -507,38 +822,29 @@ void FlowController::parsePort(xmlDoc *doc, xmlNode *processorNode, ProcessGroup Processor *processor = NULL; RemoteProcessorGroupPort *port = NULL; - if (!parent) - { + if (!parent) { _logger->log_error("parseProcessNode: no parent group existed"); return; } - // generate the random UIID +// generate the random UIID uuid_generate(uuid); - for (currentNode = processorNode->xmlChildrenNode; currentNode != NULL; currentNode = currentNode->next) - { - if (currentNode->type == XML_ELEMENT_NODE) - { - if (xmlStrcmp(currentNode->name, BAD_CAST "id") == 0) - { + for (currentNode = processorNode->xmlChildrenNode; currentNode != NULL; currentNode = currentNode->next) { + if (currentNode->type == XML_ELEMENT_NODE) { + if (xmlStrcmp(currentNode->name, BAD_CAST "id") == 0) { id = (char *) xmlNodeGetContent(currentNode); - if (id) - { + if (id) { _logger->log_debug("parseProcessorNode: id => [%s]", id); uuid_parse(id, uuid); xmlFree(id); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "name") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "name") == 0) { name = (char *) xmlNodeGetContent(currentNode); - if (name) - { + if (name) { _logger->log_debug("parseProcessorNode: name => [%s]", name); port = new RemoteProcessorGroupPort(name, uuid); processor = (Processor *) port; - if (processor == NULL) - { + if (processor == NULL) { xmlFree(name); return; } @@ -551,55 +857,42 @@ void FlowController::parsePort(xmlDoc *doc, xmlNode *processorNode, ProcessGroup parent->addProcessor(processor); xmlFree(name); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "scheduledState") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "scheduledState") == 0) { char *temp = (char *) xmlNodeGetContent(currentNode); - if (temp) - { + if (temp) { std::string state = temp; - if (state == "DISABLED") - { + if (state == "DISABLED") { _logger->log_debug("parseProcessorNode: scheduledState => [%s]", state.c_str()); processor->setScheduledState(DISABLED); } - if (state == "STOPPED") - { + if (state == "STOPPED") { _logger->log_debug("parseProcessorNode: scheduledState => [%s]", state.c_str()); processor->setScheduledState(STOPPED); } - if (state == "RUNNING") - { + if (state == "RUNNING") { _logger->log_debug("parseProcessorNode: scheduledState => [%s]", state.c_str()); processor->setScheduledState(RUNNING); } xmlFree(temp); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "maxConcurrentTasks") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "maxConcurrentTasks") == 0) { char *temp = (char *) xmlNodeGetContent(currentNode); - if (temp) - { + if (temp) { int64_t maxConcurrentTasks; - if (Property::StringToInt(temp, maxConcurrentTasks)) - { + if (Property::StringToInt(temp, maxConcurrentTasks)) { _logger->log_debug("parseProcessorNode: maxConcurrentTasks => [%d]", maxConcurrentTasks); processor->setMaxConcurrentTasks(maxConcurrentTasks); } xmlFree(temp); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "property") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "property") == 0) { this->parseProcessorProperty(doc, currentNode, processor); } } // if (currentNode->type == XML_ELEMENT_NODE) - } // while node + } // while node } -void FlowController::parseProcessorNode(xmlDoc *doc, xmlNode *processorNode, ProcessGroup *parent) -{ +void FlowController::parseProcessorNode(xmlDoc *doc, xmlNode *processorNode, ProcessGroup *parent) { char *id = NULL; char *name = NULL; int64_t schedulingPeriod = -1; @@ -611,37 +904,28 @@ void FlowController::parseProcessorNode(xmlDoc *doc, xmlNode *processorNode, Pro xmlNode *currentNode; Processor *processor = NULL; - if (!parent) - { + if (!parent) { _logger->log_error("parseProcessNode: no parent group existed"); return; } - // generate the random UIID +// generate the random UIID uuid_generate(uuid); - for (currentNode = processorNode->xmlChildrenNode; currentNode != NULL; currentNode = currentNode->next) - { - if (currentNode->type == XML_ELEMENT_NODE) - { - if (xmlStrcmp(currentNode->name, BAD_CAST "id") == 0) - { + for (currentNode = processorNode->xmlChildrenNode; currentNode != NULL; currentNode = currentNode->next) { + if (currentNode->type == XML_ELEMENT_NODE) { + if (xmlStrcmp(currentNode->name, BAD_CAST "id") == 0) { id = (char *) xmlNodeGetContent(currentNode); - if (id) - { + if (id) { _logger->log_debug("parseProcessorNode: id => [%s]", id); uuid_parse(id, uuid); xmlFree(id); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "name") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "name") == 0) { name = (char *) xmlNodeGetContent(currentNode); - if (name) - { + if (name) { _logger->log_debug("parseProcessorNode: name => [%s]", name); processor = this->createProcessor(name, uuid); - if (processor == NULL) - { + if (processor == NULL) { xmlFree(name); return; } @@ -649,251 +933,211 @@ void FlowController::parseProcessorNode(xmlDoc *doc, xmlNode *processorNode, Pro parent->addProcessor(processor); xmlFree(name); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "schedulingPeriod") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "schedulingPeriod") == 0) { TimeUnit unit; char *temp = (char *) xmlNodeGetContent(currentNode); - if (temp) - { - if (Property::StringToTime(temp, schedulingPeriod, unit) && - Property::ConvertTimeUnitToNS(schedulingPeriod, unit, schedulingPeriod)) - { + if (temp) { + if (Property::StringToTime(temp, schedulingPeriod, unit) + && Property::ConvertTimeUnitToNS(schedulingPeriod, unit, schedulingPeriod)) { _logger->log_debug("parseProcessorNode: schedulingPeriod => [%d] ns", schedulingPeriod); processor->setSchedulingPeriodNano(schedulingPeriod); } xmlFree(temp); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "penalizationPeriod") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "penalizationPeriod") == 0) { TimeUnit unit; char *temp = (char *) xmlNodeGetContent(currentNode); - if (temp) - { - if (Property::StringToTime(temp, penalizationPeriod, unit) && - Property::ConvertTimeUnitToMS(penalizationPeriod, unit, penalizationPeriod)) - { + if (temp) { + if (Property::StringToTime(temp, penalizationPeriod, unit) + && Property::ConvertTimeUnitToMS(penalizationPeriod, unit, penalizationPeriod)) { _logger->log_debug("parseProcessorNode: penalizationPeriod => [%d] ms", penalizationPeriod); processor->setPenalizationPeriodMsec(penalizationPeriod); } xmlFree(temp); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "yieldPeriod") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "yieldPeriod") == 0) { TimeUnit unit; char *temp = (char *) xmlNodeGetContent(currentNode); - if (temp) - { - if (Property::StringToTime(temp, yieldPeriod, unit) && - Property::ConvertTimeUnitToMS(yieldPeriod, unit, yieldPeriod)) - { + if (temp) { + if (Property::StringToTime(temp, yieldPeriod, unit) + && Property::ConvertTimeUnitToMS(yieldPeriod, unit, yieldPeriod)) { _logger->log_debug("parseProcessorNode: yieldPeriod => [%d] ms", yieldPeriod); processor->setYieldPeriodMsec(yieldPeriod); } xmlFree(temp); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "lossTolerant") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "lossTolerant") == 0) { char *temp = (char *) xmlNodeGetContent(currentNode); - if (temp) - { - if (Property::StringToBool(temp, lossTolerant)) - { + if (temp) { + if (Property::StringToBool(temp, lossTolerant)) { _logger->log_debug("parseProcessorNode: lossTolerant => [%d]", lossTolerant); processor->setlossTolerant(lossTolerant); } xmlFree(temp); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "scheduledState") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "scheduledState") == 0) { char *temp = (char *) xmlNodeGetContent(currentNode); - if (temp) - { + if (temp) { std::string state = temp; - if (state == "DISABLED") - { + if (state == "DISABLED") { _logger->log_debug("parseProcessorNode: scheduledState => [%s]", state.c_str()); processor->setScheduledState(DISABLED); } - if (state == "STOPPED") - { + if (state == "STOPPED") { _logger->log_debug("parseProcessorNode: scheduledState => [%s]", state.c_str()); processor->setScheduledState(STOPPED); } - if (state == "RUNNING") - { + if (state == "RUNNING") { _logger->log_debug("parseProcessorNode: scheduledState => [%s]", state.c_str()); processor->setScheduledState(RUNNING); } xmlFree(temp); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "schedulingStrategy") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "schedulingStrategy") == 0) { char *temp = (char *) xmlNodeGetContent(currentNode); - if (temp) - { + if (temp) { std::string strategy = temp; - if (strategy == "TIMER_DRIVEN") - { + if (strategy == "TIMER_DRIVEN") { _logger->log_debug("parseProcessorNode: scheduledStrategy => [%s]", strategy.c_str()); processor->setSchedulingStrategy(TIMER_DRIVEN); } - if (strategy == "EVENT_DRIVEN") - { + if (strategy == "EVENT_DRIVEN") { _logger->log_debug("parseProcessorNode: scheduledStrategy => [%s]", strategy.c_str()); processor->setSchedulingStrategy(EVENT_DRIVEN); } xmlFree(temp); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "maxConcurrentTasks") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "maxConcurrentTasks") == 0) { char *temp = (char *) xmlNodeGetContent(currentNode); - if (temp) - { + if (temp) { int64_t maxConcurrentTasks; - if (Property::StringToInt(temp, maxConcurrentTasks)) - { + if (Property::StringToInt(temp, maxConcurrentTasks)) { _logger->log_debug("parseProcessorNode: maxConcurrentTasks => [%d]", maxConcurrentTasks); processor->setMaxConcurrentTasks(maxConcurrentTasks); } xmlFree(temp); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "runDurationNanos") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "runDurationNanos") == 0) { char *temp = (char *) xmlNodeGetContent(currentNode); - if (temp) - { - if (Property::StringToInt(temp, runDurationNanos)) - { + if (temp) { + if (Property::StringToInt(temp, runDurationNanos)) { _logger->log_debug("parseProcessorNode: runDurationNanos => [%d]", runDurationNanos); processor->setRunDurationNano(runDurationNanos); } xmlFree(temp); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "autoTerminatedRelationship") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "autoTerminatedRelationship") == 0) { char *temp = (char *) xmlNodeGetContent(currentNode); - if (temp) - { + if (temp) { std::string relationshipName = temp; Relationship relationship(relationshipName, ""); std::set<Relationship> relationships; relationships.insert(relationship); processor->setAutoTerminatedRelationships(relationships); - _logger->log_debug("parseProcessorNode: autoTerminatedRelationship => [%s]", relationshipName.c_str()); + _logger->log_debug("parseProcessorNode: autoTerminatedRelationship => [%s]", + relationshipName.c_str()); xmlFree(temp); } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "property") == 0) - { + } else if (xmlStrcmp(currentNode->name, BAD_CAST "property") == 0) { this->parseProcessorProperty(doc, currentNode, processor); } } // if (currentNode->type == XML_ELEMENT_NODE) - } // while node + } // while node } -void FlowController::load() -{ - if (_running) - { +void FlowController::load(ConfigFormat configFormat) { + if (_running) { stop(true); } - if (!_initialized) - { - _logger->log_info("Load Flow Controller from file %s", _xmlFileName.c_str()); - - xmlDoc *doc = xmlReadFile(_xmlFileName.c_str(), NULL, XML_PARSE_NONET); - if (doc == NULL) - { - _logger->log_error("xmlReadFile returned NULL when reading [%s]", _xmlFileName.c_str()); - _initialized = true; - return; - } + if (!_initialized) { + _logger->log_info("Load Flow Controller from file %s", _configurationFileName.c_str()); - xmlNode *root = xmlDocGetRootElement(doc); + if (ConfigFormat::XML == configFormat) { + _logger->log_info("Detected an XML configuration file for processing."); - if (root == NULL) - { - _logger->log_error("Can not get root from XML doc %s", _xmlFileName.c_str()); - xmlFreeDoc(doc); - xmlCleanupParser(); - } + xmlDoc *doc = xmlReadFile(_configurationFileName.c_str(), NULL, XML_PARSE_NONET); + if (doc == NULL) { + _logger->log_error("xmlReadFile returned NULL when reading [%s]", _configurationFileName.c_str()); + _initialized = true; + return; + } - if (xmlStrcmp(root->name, BAD_CAST "flowController") != 0) - { - _logger->log_error("Root name is not flowController for XML doc %s", _xmlFileName.c_str()); - xmlFreeDoc(doc); - xmlCleanupParser(); - return; - } + xmlNode *root = xmlDocGetRootElement(doc); + + if (root == NULL) { + _logger->log_error("Can not get root from XML doc %s", _configurationFileName.c_str()); + xmlFreeDoc(doc); + xmlCleanupParser(); + } + + if (xmlStrcmp(root->name, BAD_CAST "flowController") != 0) { + _logger->log_error("Root name is not flowController for XML doc %s", _configurationFileName.c_str()); + xmlFreeDoc(doc); + xmlCleanupParser(); + return; + } + + xmlNode *currentNode; - xmlNode *currentNode; - - for (currentNode = root->xmlChildrenNode; currentNode != NULL; currentNode = currentNode->next) - { - if (currentNode->type == XML_ELEMENT_NODE) - { - if (xmlStrcmp(currentNode->name, BAD_CAST "rootGroup") == 0) - { - this->parseRootProcessGroup(doc, currentNode); - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "maxTimerDrivenThreadCount") == 0) - { - char *temp = (char *) xmlNodeGetContent(currentNode); - int64_t maxTimerDrivenThreadCount; - if (temp) - { - if (Property::StringToInt(temp, maxTimerDrivenThreadCount)) - { - _logger->log_debug("maxTimerDrivenThreadCount => [%d]", maxTimerDrivenThreadCount); - this->_maxTimerDrivenThreads = maxTimerDrivenThreadCount; + for (currentNode = root->xmlChildrenNode; currentNode != NULL; currentNode = currentNode->next) { + if (currentNode->type == XML_ELEMENT_NODE) { + if (xmlStrcmp(currentNode->name, BAD_CAST "rootGroup") == 0) { + this->parseRootProcessGroup(doc, currentNode); + } else if (xmlStrcmp(currentNode->name, BAD_CAST "maxTimerDrivenThreadCount") == 0) { + char *temp = (char *) xmlNodeGetContent(currentNode); + int64_t maxTimerDrivenThreadCount; + if (temp) { + if (Property::StringToInt(temp, maxTimerDrivenThreadCount)) { + _logger->log_debug("maxTimerDrivenThreadCount => [%d]", maxTimerDrivenThreadCount); + this->_maxTimerDrivenThreads = maxTimerDrivenThreadCount; + } + xmlFree(temp); } - xmlFree(temp); - } - } - else if (xmlStrcmp(currentNode->name, BAD_CAST "maxEventDrivenThreadCount") == 0) - { - char *temp = (char *) xmlNodeGetContent(currentNode); - int64_t maxEventDrivenThreadCount; - if (temp) - { - if (Property::StringToInt(temp, maxEventDrivenThreadCount)) - { - _logger->log_debug("maxEventDrivenThreadCount => [%d]", maxEventDrivenThreadCount); - this->_maxEventDrivenThreads = maxEventDrivenThreadCount; + } else if (xmlStrcmp(currentNode->name, BAD_CAST "maxEventDrivenThreadCount") == 0) { + char *temp = (char *) xmlNodeGetContent(currentNode); + int64_t maxEventDrivenThreadCount; + if (temp) { + if (Property::StringToInt(temp, maxEventDrivenThreadCount)) { + _logger->log_debug("maxEventDrivenThreadCount => [%d]", maxEventDrivenThreadCount); + this->_maxEventDrivenThreads = maxEventDrivenThreadCount; + } + xmlFree(temp); } - xmlFree(temp); } - } - } // type == XML_ELEMENT_NODE - } // for + } // type == XML_ELEMENT_NODE + } // for + + xmlFreeDoc(doc); + xmlCleanupParser(); + _initialized = true; + } else if (ConfigFormat::YAML == configFormat) { + _logger->log_info("Detected a YAML configuration file for processing."); - xmlFreeDoc(doc); - xmlCleanupParser(); - _initialized = true; + YAML::Node flow = YAML::LoadFile(_configurationFileName); + + YAML::Node flowControllerNode = flow["Flow Controller"]; + YAML::Node processorsNode = flow[CONFIG_YAML_PROCESSORS_KEY]; + YAML::Node connectionsNode = flow["Connections"]; + YAML::Node remoteProcessingGroupNode = flow["Remote Processing Groups"]; + + // Create the root process group + parseRootProcessGroupYaml(flowControllerNode); + parseProcessorNodeYaml(processorsNode, this->_root); + parseRemoteProcessGroupYaml(&remoteProcessingGroupNode, this->_root); + parseConnectionYaml(&connectionsNode, this->_root); + + _initialized = true; + } } } -bool FlowController::start() -{ - if (!_initialized) - { +bool FlowController::start() { + if (!_initialized) { _logger->log_error("Can not start Flow Controller because it has not been initialized"); return false; - } - else - { - if (!_running) - { + } else { + if (!_running) { _logger->log_info("Start Flow Controller"); this->_timerScheduler.start(); if (this->_root)
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/cf813d4d/src/ProcessGroup.cpp ---------------------------------------------------------------------- diff --git a/src/ProcessGroup.cpp b/src/ProcessGroup.cpp index d44cc99..70ee9d7 100644 --- a/src/ProcessGroup.cpp +++ b/src/ProcessGroup.cpp @@ -218,6 +218,29 @@ Processor *ProcessGroup::findProcessor(uuid_t uuid) return ret; } +Processor *ProcessGroup::findProcessor(std::string processorName) +{ + Processor *ret = NULL; + + for (std::set<Processor *>::iterator it = _processors.begin(); it != _processors.end(); ++it) + { + Processor *processor(*it); + _logger->log_debug("Current processor is %s", processor->getName().c_str()); + if (processor->getName() == processorName) + return processor; + } + + for (std::set<ProcessGroup *>::iterator it = _childProcessGroups.begin(); it != _childProcessGroups.end(); ++it) + { + ProcessGroup *processGroup(*it); + Processor *processor = processGroup->findProcessor(processorName); + if (processor) + return processor; + } + + return ret; +} + void ProcessGroup::updatePropertyValue(std::string processorName, std::string propertyName, std::string propertyValue) { std::lock_guard<std::mutex> lock(_mtx); http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/cf813d4d/src/ProcessSession.cpp ---------------------------------------------------------------------- diff --git a/src/ProcessSession.cpp b/src/ProcessSession.cpp index 2628ae3..a6b0a19 100644 --- a/src/ProcessSession.cpp +++ b/src/ProcessSession.cpp @@ -462,7 +462,8 @@ void ProcessSession::commit() if (!_processContext->getProcessor()->isAutoTerminated(relationship)) { // Not autoterminate, we should have the connect - throw Exception(PROCESS_SESSION_EXCEPTION, "Connect empty for non auto terminated relationship"); + std::string message = "Connect empty for non auto terminated relationship" + relationship.getName(); + throw Exception(PROCESS_SESSION_EXCEPTION, message.c_str()); } else { @@ -521,7 +522,8 @@ void ProcessSession::commit() if (!_processContext->getProcessor()->isAutoTerminated(relationship)) { // Not autoterminate, we should have the connect - throw Exception(PROCESS_SESSION_EXCEPTION, "Connect empty for non auto terminated relationship"); + std::string message = "Connect empty for non auto terminated relationship " + relationship.getName(); + throw Exception(PROCESS_SESSION_EXCEPTION, message.c_str()); } else { http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/cf813d4d/src/Processor.cpp ---------------------------------------------------------------------- diff --git a/src/Processor.cpp b/src/Processor.cpp index d5e1a6a..cc136dc 100644 --- a/src/Processor.cpp +++ b/src/Processor.cpp @@ -208,14 +208,16 @@ bool Processor::getProperty(std::string name, std::string &value) bool Processor::setProperty(std::string name, std::string value) { + std::lock_guard<std::mutex> lock(_mtx); std::map<std::string, Property>::iterator it = _properties.find(name); + if (it != _properties.end()) { Property item = it->second; item.setValue(value); _properties[item.getName()] = item; - _logger->log_info("Processor %s property name %s value %s", _name.c_str(), item.getName().c_str(), item.getValue().c_str()); + _logger->log_info("Processor %s property name %s value %s", _name.c_str(), item.getName().c_str(), value.c_str()); return true; } else http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/cf813d4d/thirdparty/uuid/libuuid.a ---------------------------------------------------------------------- diff --git a/thirdparty/uuid/libuuid.a b/thirdparty/uuid/libuuid.a deleted file mode 100644 index 761b642..0000000 Binary files a/thirdparty/uuid/libuuid.a and /dev/null differ http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/cf813d4d/thirdparty/uuid/tst_uuid ---------------------------------------------------------------------- diff --git a/thirdparty/uuid/tst_uuid b/thirdparty/uuid/tst_uuid index e067cb2..4343c86 100755 Binary files a/thirdparty/uuid/tst_uuid and b/thirdparty/uuid/tst_uuid differ http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/cf813d4d/thirdparty/yaml-cpp-yaml-cpp-0.5.3/.clang-format ---------------------------------------------------------------------- diff --git a/thirdparty/yaml-cpp-yaml-cpp-0.5.3/.clang-format b/thirdparty/yaml-cpp-yaml-cpp-0.5.3/.clang-format new file mode 100644 index 0000000..d6e1b22 --- /dev/null +++ b/thirdparty/yaml-cpp-yaml-cpp-0.5.3/.clang-format @@ -0,0 +1,47 @@ +--- +# BasedOnStyle: Google +AccessModifierOffset: -1 +ConstructorInitializerIndentWidth: 4 +AlignEscapedNewlinesLeft: true +AlignTrailingComments: true +AllowAllParametersOfDeclarationOnNextLine: true +AllowShortIfStatementsOnASingleLine: false +AllowShortLoopsOnASingleLine: false +AlwaysBreakTemplateDeclarations: true +AlwaysBreakBeforeMultilineStrings: true +BreakBeforeBinaryOperators: false +BreakBeforeTernaryOperators: true +BreakConstructorInitializersBeforeComma: false +BinPackParameters: true +ColumnLimit: 80 +ConstructorInitializerAllOnOneLineOrOnePerLine: true +DerivePointerBinding: true +ExperimentalAutoDetectBinPacking: false +IndentCaseLabels: true +MaxEmptyLinesToKeep: 1 +NamespaceIndentation: None +ObjCSpaceBeforeProtocolList: false +PenaltyBreakBeforeFirstCallParameter: 1 +PenaltyBreakComment: 60 +PenaltyBreakString: 1000 +PenaltyBreakFirstLessLess: 120 +PenaltyExcessCharacter: 1000000 +PenaltyReturnTypeOnItsOwnLine: 200 +PointerBindsToType: true +SpacesBeforeTrailingComments: 2 +Cpp11BracedListStyle: true +Standard: Auto +IndentWidth: 2 +TabWidth: 8 +UseTab: Never +BreakBeforeBraces: Attach +IndentFunctionDeclarationAfterType: true +SpacesInParentheses: false +SpacesInAngles: false +SpaceInEmptyParentheses: false +SpacesInCStyleCastParentheses: false +SpaceAfterControlStatementKeyword: true +SpaceBeforeAssignmentOperators: true +ContinuationIndentWidth: 4 +... + http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/cf813d4d/thirdparty/yaml-cpp-yaml-cpp-0.5.3/.gitignore ---------------------------------------------------------------------- diff --git a/thirdparty/yaml-cpp-yaml-cpp-0.5.3/.gitignore b/thirdparty/yaml-cpp-yaml-cpp-0.5.3/.gitignore new file mode 100644 index 0000000..567609b --- /dev/null +++ b/thirdparty/yaml-cpp-yaml-cpp-0.5.3/.gitignore @@ -0,0 +1 @@ +build/ http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/cf813d4d/thirdparty/yaml-cpp-yaml-cpp-0.5.3/CMakeLists.txt ---------------------------------------------------------------------- diff --git a/thirdparty/yaml-cpp-yaml-cpp-0.5.3/CMakeLists.txt b/thirdparty/yaml-cpp-yaml-cpp-0.5.3/CMakeLists.txt new file mode 100644 index 0000000..d4a8e29 --- /dev/null +++ b/thirdparty/yaml-cpp-yaml-cpp-0.5.3/CMakeLists.txt @@ -0,0 +1,340 @@ +### +### CMake settings +### +## Due to Mac OSX we need to keep compatibility with CMake 2.6 +# see http://www.cmake.org/Wiki/CMake_Policies +cmake_minimum_required(VERSION 2.6) +# see http://www.cmake.org/cmake/help/cmake-2-8-docs.html#policy:CMP0012 +if(POLICY CMP0012) + cmake_policy(SET CMP0012 OLD) +endif() +# see http://www.cmake.org/cmake/help/cmake-2-8-docs.html#policy:CMP0015 +if(POLICY CMP0015) + cmake_policy(SET CMP0015 OLD) +endif() + +include(CheckCXXCompilerFlag) + + +### +### Project settings +### +project(YAML_CPP) + +set(YAML_CPP_VERSION_MAJOR "0") +set(YAML_CPP_VERSION_MINOR "5") +set(YAML_CPP_VERSION_PATCH "3") +set(YAML_CPP_VERSION "${YAML_CPP_VERSION_MAJOR}.${YAML_CPP_VERSION_MINOR}.${YAML_CPP_VERSION_PATCH}") + +enable_testing() + + +### +### Project options +### +## Project stuff +option(YAML_CPP_BUILD_TOOLS "Enable testing and parse tools" ON) +option(YAML_CPP_BUILD_CONTRIB "Enable contrib stuff in library" ON) + +## Build options +# --> General +# see http://www.cmake.org/cmake/help/cmake2.6docs.html#variable:BUILD_SHARED_LIBS +# http://www.cmake.org/cmake/help/cmake2.6docs.html#command:add_library +option(BUILD_SHARED_LIBS "Build Shared Libraries" OFF) + +# --> Apple +option(APPLE_UNIVERSAL_BIN "Apple: Build universal binary" OFF) + +# --> Microsoft Visual C++ +# see http://msdn.microsoft.com/en-us/library/aa278396(v=VS.60).aspx +# http://msdn.microsoft.com/en-us/library/2kzt1wy3(v=VS.71).aspx +option(MSVC_SHARED_RT "MSVC: Build with shared runtime libs (/MD)" ON) +option(MSVC_STHREADED_RT "MSVC: Build with single-threaded static runtime libs (/ML until VS .NET 2003)" OFF) + +### +### Sources, headers, directories and libs +### + +# From http://www.cmake.org/pipermail/cmake/2010-March/035992.html: +# function to collect all the sources from sub-directories +# into a single list +function(add_sources) + get_property(is_defined GLOBAL PROPERTY SRCS_LIST DEFINED) + if(NOT is_defined) + define_property(GLOBAL PROPERTY SRCS_LIST + BRIEF_DOCS "List of source files" + FULL_DOCS "List of all source files in the entire project") + endif() + # make absolute paths + set(SRCS) + foreach(s IN LISTS ARGN) + if(NOT IS_ABSOLUTE "${s}") + get_filename_component(s "${s}" ABSOLUTE) + endif() + list(APPEND SRCS "${s}") + endforeach() + # append to global list + set_property(GLOBAL APPEND PROPERTY SRCS_LIST "${SRCS}") +endfunction(add_sources) + +set(header_directory "include/yaml-cpp/") + +file(GLOB sources "src/[a-zA-Z]*.cpp") +file(GLOB_RECURSE public_headers "include/yaml-cpp/[a-zA-Z]*.h") +file(GLOB private_headers "src/[a-zA-Z]*.h") + +if(YAML_CPP_BUILD_CONTRIB) + file(GLOB contrib_sources "src/contrib/[a-zA-Z]*.cpp") + file(GLOB contrib_public_headers "include/yaml-cpp/contrib/[a-zA-Z]*.h") + file(GLOB contrib_private_headers "src/contrib/[a-zA-Z]*.h") +else() + add_definitions(-DYAML_CPP_NO_CONTRIB) +endif() + +set(library_sources + ${sources} + ${public_headers} + ${private_headers} + ${contrib_sources} + ${contrib_public_headers} + ${contrib_private_headers} +) +add_sources(${library_sources}) + +if(VERBOSE) + message(STATUS "sources: ${sources}") + message(STATUS "public_headers: ${public_headers}") + message(STATUS "private_headers: ${private_headers}") + message(STATUS "contrib_sources: ${contrib_sources}") + message(STATUS "contrib_public_headers: ${contrib_public_headers}") + message(STATUS "contrib_private_headers: ${contrib_private_headers}") +endif() + +include_directories(${YAML_CPP_SOURCE_DIR}/src) +include_directories(${YAML_CPP_SOURCE_DIR}/include) + +find_package(Boost REQUIRED) +include_directories(${Boost_INCLUDE_DIRS}) + + +### +### General compilation settings +### +set(yaml_c_flags ${CMAKE_C_FLAGS}) +set(yaml_cxx_flags ${CMAKE_CXX_FLAGS}) + +if(BUILD_SHARED_LIBS) + set(LABEL_SUFFIX "shared") +else() + set(LABEL_SUFFIX "static") +endif() + +if(APPLE) + if(APPLE_UNIVERSAL_BIN) + set(CMAKE_OSX_ARCHITECTURES ppc;i386) + endif() +endif() + +if(IPHONE) + set(CMAKE_OSX_SYSROOT "iphoneos4.2") + set(CMAKE_OSX_ARCHITECTURES "armv6;armv7") +endif() + +if(WIN32) + if(BUILD_SHARED_LIBS) + add_definitions(-D${PROJECT_NAME}_DLL) # use or build Windows DLL + endif() + if(CMAKE_INSTALL_PREFIX_INITIALIZED_TO_DEFAULT) + set(CMAKE_INSTALL_PREFIX "C:/") + endif() +endif() + +# GCC or Clang specialities +if("${CMAKE_CXX_COMPILER_ID}" MATCHES "GNU" OR + "${CMAKE_CXX_COMPILER_ID}" MATCHES "Clang") + ### General stuff + if(WIN32) + set(CMAKE_SHARED_LIBRARY_PREFIX "") # DLLs do not have a "lib" prefix + set(CMAKE_IMPORT_LIBRARY_PREFIX "") # same for DLL import libs + set(CMAKE_LINK_DEF_FILE_FLAG "") # CMake workaround (2.8.3) + endif() + + ### Project stuff + if(NOT CMAKE_CONFIGURATION_TYPES AND NOT CMAKE_BUILD_TYPE) + set(CMAKE_BUILD_TYPE Release) + endif() + # + set(CMAKE_CXX_FLAGS_RELEASE "-O2") + set(CMAKE_CXX_FLAGS_RELWITHDEBINFO "-O2 -g") + set(CMAKE_CXX_FLAGS_DEBUG "-g") + set(CMAKE_CXX_FLAGS_MINSIZEREL "-Os") + # + set(GCC_EXTRA_OPTIONS "") + # + set(FLAG_TESTED "-Wextra") + check_cxx_compiler_flag(${FLAG_TESTED} FLAG_WEXTRA) + if(FLAG_WEXTRA) + set(GCC_EXTRA_OPTIONS "${GCC_EXTRA_OPTIONS} ${FLAG_TESTED}") + endif() + # + set(yaml_cxx_flags "-Wall ${GCC_EXTRA_OPTIONS} -pedantic -Wno-long-long ${yaml_cxx_flags}") + + ### Make specific + if(${CMAKE_BUILD_TOOL} MATCHES make OR ${CMAKE_BUILD_TOOL} MATCHES gmake) + add_custom_target(debuggable $(MAKE) clean + COMMAND ${CMAKE_COMMAND} -DCMAKE_BUILD_TYPE=Debug ${CMAKE_SOURCE_DIR} + COMMENT "Adjusting settings for debug compilation" + VERBATIM) + add_custom_target(releasable $(MAKE) clean + COMMAND ${CMAKE_COMMAND} -DCMAKE_BUILD_TYPE=Release ${CMAKE_SOURCE_DIR} + COMMENT "Adjusting settings for release compilation" + VERBATIM) + endif() +endif() + +# Microsoft VisualC++ specialities +if(MSVC) + ### General stuff + # a) Change MSVC runtime library settings (/MD[d], /MT[d], /ML[d] (single-threaded until VS 2003)) + # plus set lib suffix for later use and project label accordingly + # see http://msdn.microsoft.com/en-us/library/aa278396(v=VS.60).aspx + # http://msdn.microsoft.com/en-us/library/2kzt1wy3(v=VS.71).aspx + set(LIB_RT_SUFFIX "md") # CMake defaults to /MD for MSVC + set(LIB_RT_OPTION "/MD") + # + if(NOT MSVC_SHARED_RT) # User wants to have static runtime libraries (/MT, /ML) + if(MSVC_STHREADED_RT) # User wants to have old single-threaded static runtime libraries + set(LIB_RT_SUFFIX "ml") + set(LIB_RT_OPTION "/ML") + if(NOT ${MSVC_VERSION} LESS 1400) + message(FATAL_ERROR "Single-threaded static runtime libraries (/ML) only available until VS .NET 2003 (7.1).") + endif() + else() + set(LIB_RT_SUFFIX "mt") + set(LIB_RT_OPTION "/MT") + endif() + + # correct linker options + foreach(flag_var yaml_c_flags yaml_cxx_flags) + foreach(config_name "" DEBUG RELEASE MINSIZEREL RELWITHDEBINFO) + set(var_name "${flag_var}") + if(NOT "${config_name}" STREQUAL "") + set(var_name "${var_name}_${config_name}") + endif() + string(REPLACE "/MD" "${LIB_RT_OPTION}" ${var_name} "${${var_name}}") + endforeach() + endforeach() + endif() + # + set(LABEL_SUFFIX "${LABEL_SUFFIX} ${LIB_RT_SUFFIX}") + + # b) Change prefix for static libraries + set(CMAKE_STATIC_LIBRARY_PREFIX "lib") # to distinguish static libraries from DLL import libs + + # c) Correct suffixes for static libraries + if(NOT BUILD_SHARED_LIBS) + ### General stuff + set(LIB_TARGET_SUFFIX "${LIB_SUFFIX}${LIB_RT_SUFFIX}") + endif() + + ### Project stuff + # /W3 = set warning level; see http://msdn.microsoft.com/en-us/library/thxezb7y.aspx + # /wd4127 = disable warning C4127 "conditional expression is constant"; see http://msdn.microsoft.com/en-us/library/6t66728h.aspx + # /wd4355 = disable warning C4355 "'this' : used in base member initializer list"; http://msdn.microsoft.com/en-us/library/3c594ae3.aspx + set(yaml_cxx_flags "/W3 /wd4127 /wd4355 ${yaml_cxx_flags}") +endif() + + +### +### General install settings +### +if(WIN32) + set(_library_dir bin) # .dll are in PATH, like executables +else() + set(_library_dir lib) +endif() + +set(INCLUDE_INSTALL_ROOT_DIR include) + +set(INCLUDE_INSTALL_DIR ${INCLUDE_INSTALL_ROOT_DIR}/yaml-cpp) +set(LIB_INSTALL_DIR "${_library_dir}${LIB_SUFFIX}") + +set(_INSTALL_DESTINATIONS + RUNTIME DESTINATION bin + LIBRARY DESTINATION ${LIB_INSTALL_DIR} + ARCHIVE DESTINATION "lib${LIB_SUFFIX}" +) + + +### +### Library +### +add_library(yaml-cpp ${library_sources}) +set_target_properties(yaml-cpp PROPERTIES + COMPILE_FLAGS "${yaml_c_flags} ${yaml_cxx_flags}" +) + +set_target_properties(yaml-cpp PROPERTIES + VERSION "${YAML_CPP_VERSION}" + SOVERSION "${YAML_CPP_VERSION_MAJOR}.${YAML_CPP_VERSION_MINOR}" + PROJECT_LABEL "yaml-cpp ${LABEL_SUFFIX}" +) + +if(IPHONE) + set_target_properties(yaml-cpp PROPERTIES + XCODE_ATTRIBUTE_IPHONEOS_DEPLOYMENT_TARGET "3.0" + ) +endif() + +if(MSVC) + if(NOT BUILD_SHARED_LIBS) + # correct library names + set_target_properties(yaml-cpp PROPERTIES + DEBUG_POSTFIX "${LIB_TARGET_SUFFIX}d" + RELEASE_POSTFIX "${LIB_TARGET_SUFFIX}" + MINSIZEREL_POSTFIX "${LIB_TARGET_SUFFIX}" + RELWITHDEBINFO_POSTFIX "${LIB_TARGET_SUFFIX}" + ) + endif() +endif() + +install(TARGETS yaml-cpp ${_INSTALL_DESTINATIONS}) +install( + DIRECTORY ${header_directory} + DESTINATION ${INCLUDE_INSTALL_DIR} + FILES_MATCHING PATTERN "*.h" +) + +export( + TARGETS yaml-cpp + FILE "${PROJECT_BINARY_DIR}/yaml-cpp-targets.cmake") +export(PACKAGE yaml-cpp) +set(EXPORT_TARGETS yaml-cpp CACHE INTERNAL "export targets") + +set(CONFIG_INCLUDE_DIRS "${YAML_CPP_SOURCE_DIR}/include") +configure_file(${CMAKE_CURRENT_SOURCE_DIR}/yaml-cpp-config.cmake.in + "${PROJECT_BINARY_DIR}/yaml-cpp-config.cmake" @ONLY) +configure_file(${CMAKE_CURRENT_SOURCE_DIR}/yaml-cpp-config-version.cmake.in + "${PROJECT_BINARY_DIR}/yaml-cpp-config-version.cmake" @ONLY) + +if(UNIX) + set(PC_FILE ${CMAKE_BINARY_DIR}/yaml-cpp.pc) + configure_file("yaml-cpp.pc.cmake" ${PC_FILE} @ONLY) + install(FILES ${PC_FILE} DESTINATION ${LIB_INSTALL_DIR}/pkgconfig) +endif() + + +### +### Extras +### +if(YAML_CPP_BUILD_TOOLS) + add_subdirectory(test) + add_subdirectory(util) +endif() + +### Formatting +get_property(all_sources GLOBAL PROPERTY SRCS_LIST) +add_custom_target(format + COMMAND clang-format --style=file -i ${all_sources} + COMMENT "Running clang-format" + VERBATIM) http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/cf813d4d/thirdparty/yaml-cpp-yaml-cpp-0.5.3/CONTRIBUTING.md ---------------------------------------------------------------------- diff --git a/thirdparty/yaml-cpp-yaml-cpp-0.5.3/CONTRIBUTING.md b/thirdparty/yaml-cpp-yaml-cpp-0.5.3/CONTRIBUTING.md new file mode 100644 index 0000000..41f50dc --- /dev/null +++ b/thirdparty/yaml-cpp-yaml-cpp-0.5.3/CONTRIBUTING.md @@ -0,0 +1,17 @@ +# Style + +This project is formatted with [clang-format][fmt] using the style file at the root of the repository. Please run clang-format before sending a pull request. + +In general, try to follow the style of surrounding code. + +[fmt]: http://clang.llvm.org/docs/ClangFormat.html + +# Tests + +Please verify the tests pass by running the target `tests/run_tests`. + +If you are adding functionality, add tests accordingly. + +# Pull request process + +Every pull request undergoes a code review. Unfortunately, github's code review process isn't great, but we'll manage. During the code review, if you make changes, add new commits to the pull request for each change. Once the code review is complete, rebase against the master branch and squash into a single commit. http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/cf813d4d/thirdparty/yaml-cpp-yaml-cpp-0.5.3/LICENSE ---------------------------------------------------------------------- diff --git a/thirdparty/yaml-cpp-yaml-cpp-0.5.3/LICENSE b/thirdparty/yaml-cpp-yaml-cpp-0.5.3/LICENSE new file mode 100644 index 0000000..991fdbb --- /dev/null +++ b/thirdparty/yaml-cpp-yaml-cpp-0.5.3/LICENSE @@ -0,0 +1,19 @@ +Copyright (c) 2008-2015 Jesse Beder. + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in +all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +THE SOFTWARE. http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/cf813d4d/thirdparty/yaml-cpp-yaml-cpp-0.5.3/Makefile ---------------------------------------------------------------------- diff --git a/thirdparty/yaml-cpp-yaml-cpp-0.5.3/Makefile b/thirdparty/yaml-cpp-yaml-cpp-0.5.3/Makefile new file mode 100644 index 0000000..f23f477 --- /dev/null +++ b/thirdparty/yaml-cpp-yaml-cpp-0.5.3/Makefile @@ -0,0 +1,40 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License + +CFLAGS = -Wall +INCLUDES = -I./include + +CPP_FILES := $(wildcard src/*.cpp) +OBJ_FILES := $(addprefix build/,$(notdir $(CPP_FILES:.cpp=.o))) + +all: lib/libyaml-cpp.a + +lib: + mkdir -p ./lib + +build: + mkdir -p ./build + +lib/libyaml-cpp.a: $(OBJ_FILES) + mkdir -p ./lib + ar crs $@ $^ + +build/%.o: src/%.cpp + mkdir -p ./build + g++ -Os $(INCLUDES) $(CC_FLAGS) -c -o $@ $< + +clean: + rm -rf ./lib ./build http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/cf813d4d/thirdparty/yaml-cpp-yaml-cpp-0.5.3/README.md ---------------------------------------------------------------------- diff --git a/thirdparty/yaml-cpp-yaml-cpp-0.5.3/README.md b/thirdparty/yaml-cpp-yaml-cpp-0.5.3/README.md new file mode 100644 index 0000000..54263ef --- /dev/null +++ b/thirdparty/yaml-cpp-yaml-cpp-0.5.3/README.md @@ -0,0 +1,52 @@ +# yaml-cpp + +yaml-cpp is a [YAML](http://www.yaml.org/) parser and emitter in C++ matching the [YAML 1.2 spec](http://www.yaml.org/spec/1.2/spec.html). + +To get a feel for how it can be used, see the [Tutorial](https://github.com/jbeder/yaml-cpp/wiki/Tutorial) or [How to Emit YAML](https://github.com/jbeder/yaml-cpp/wiki/How-To-Emit-YAML). For the old API (version < 0.5.0), see [How To Parse A Document](https://github.com/jbeder/yaml-cpp/wiki/How-To-Parse-A-Document-(Old-API)). + + +# Problems? # + +If you find a bug, post an [issue](https://github.com/jbeder/yaml-cpp/issues)! If you have questions about how to use yaml-cpp, please post it on http://stackoverflow.com and tag it `yaml-cpp`. + +# How to Build # + +yaml-cpp uses [CMake](http://www.cmake.org) to support cross-platform building. The basic steps to build are: + +1. Download and install [CMake](http://www.cmake.org) (Resources -> Download). + +**Note:** If you don't use the provided installer for your platform, make sure that you add CMake's bin folder to your path. + +2. Navigate into the source directory, and type: + +``` +mkdir build +cd build +``` + +3. Run CMake. The basic syntax is: + +``` +cmake [-G generator] [-DBUILD_SHARED_LIBS=ON|OFF] .. +``` + + * The `generator` is whatever type of build system you'd like to use. To see a full list of generators on your platform, just run `cmake` (with no arguments). For example: + * On Windows, you might use "Visual Studio 12 2013" to generate a Visual Studio 2013 solution + * On OS X, you might use "Xcode" to generate an Xcode project + * On a UNIX-y system, simply omit the option to generate a makefile + + * yaml-cpp defaults to building a static library, but you may build a shared library by specifying `-DBUILD_SHARED_LIBS=ON`. + + * For more options on customizing the build, see the [CMakeLists.txt](https://github.com/jbeder/yaml-cpp/blob/master/CMakeLists.txt) file. + +4. Build it! + +5. To clean up, just remove the `build` directory. + +# Recent Release # + +[yaml-cpp 0.5.2](https://github.com/jbeder/yaml-cpp/releases/tag/release-0.5.2) has been released! This is a bug fix release. + +[yaml-cpp 0.3.0](https://github.com/jbeder/yaml-cpp/releases/tag/release-0.3.0) is still available if you want the old API. + +**The old API will continue to be supported, and will still receive bugfixes!** The 0.3.x and 0.4.x versions will be old API releases, and 0.5.x and above will all be new API releases. http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/cf813d4d/thirdparty/yaml-cpp-yaml-cpp-0.5.3/include/yaml-cpp/anchor.h ---------------------------------------------------------------------- diff --git a/thirdparty/yaml-cpp-yaml-cpp-0.5.3/include/yaml-cpp/anchor.h b/thirdparty/yaml-cpp-yaml-cpp-0.5.3/include/yaml-cpp/anchor.h new file mode 100644 index 0000000..06759c7 --- /dev/null +++ b/thirdparty/yaml-cpp-yaml-cpp-0.5.3/include/yaml-cpp/anchor.h @@ -0,0 +1,17 @@ +#ifndef ANCHOR_H_62B23520_7C8E_11DE_8A39_0800200C9A66 +#define ANCHOR_H_62B23520_7C8E_11DE_8A39_0800200C9A66 + +#if defined(_MSC_VER) || \ + (defined(__GNUC__) && (__GNUC__ == 3 && __GNUC_MINOR__ >= 4) || \ + (__GNUC__ >= 4)) // GCC supports "pragma once" correctly since 3.4 +#pragma once +#endif + +#include <cstddef> + +namespace YAML { +typedef std::size_t anchor_t; +const anchor_t NullAnchor = 0; +} + +#endif // ANCHOR_H_62B23520_7C8E_11DE_8A39_0800200C9A66 http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/cf813d4d/thirdparty/yaml-cpp-yaml-cpp-0.5.3/include/yaml-cpp/binary.h ---------------------------------------------------------------------- diff --git a/thirdparty/yaml-cpp-yaml-cpp-0.5.3/include/yaml-cpp/binary.h b/thirdparty/yaml-cpp-yaml-cpp-0.5.3/include/yaml-cpp/binary.h new file mode 100644 index 0000000..29d5dbd --- /dev/null +++ b/thirdparty/yaml-cpp-yaml-cpp-0.5.3/include/yaml-cpp/binary.h @@ -0,0 +1,67 @@ +#ifndef BASE64_H_62B23520_7C8E_11DE_8A39_0800200C9A66 +#define BASE64_H_62B23520_7C8E_11DE_8A39_0800200C9A66 + +#if defined(_MSC_VER) || \ + (defined(__GNUC__) && (__GNUC__ == 3 && __GNUC_MINOR__ >= 4) || \ + (__GNUC__ >= 4)) // GCC supports "pragma once" correctly since 3.4 +#pragma once +#endif + +#include <string> +#include <vector> + +#include "yaml-cpp/dll.h" + +namespace YAML { +YAML_CPP_API std::string EncodeBase64(const unsigned char *data, + std::size_t size); +YAML_CPP_API std::vector<unsigned char> DecodeBase64(const std::string &input); + +class YAML_CPP_API Binary { + public: + Binary() : m_unownedData(0), m_unownedSize(0) {} + Binary(const unsigned char *data_, std::size_t size_) + : m_unownedData(data_), m_unownedSize(size_) {} + + bool owned() const { return !m_unownedData; } + std::size_t size() const { return owned() ? m_data.size() : m_unownedSize; } + const unsigned char *data() const { + return owned() ? &m_data[0] : m_unownedData; + } + + void swap(std::vector<unsigned char> &rhs) { + if (m_unownedData) { + m_data.swap(rhs); + rhs.clear(); + rhs.resize(m_unownedSize); + std::copy(m_unownedData, m_unownedData + m_unownedSize, rhs.begin()); + m_unownedData = 0; + m_unownedSize = 0; + } else { + m_data.swap(rhs); + } + } + + bool operator==(const Binary &rhs) const { + const std::size_t s = size(); + if (s != rhs.size()) + return false; + const unsigned char *d1 = data(); + const unsigned char *d2 = rhs.data(); + for (std::size_t i = 0; i < s; i++) { + if (*d1++ != *d2++) + return false; + } + return true; + } + + bool operator!=(const Binary &rhs) const { return !(*this == rhs); } + + private: + std::vector<unsigned char> m_data; + const unsigned char *m_unownedData; + std::size_t m_unownedSize; +}; +} + +#endif // BASE64_H_62B23520_7C8E_11DE_8A39_0800200C9A66 http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/cf813d4d/thirdparty/yaml-cpp-yaml-cpp-0.5.3/include/yaml-cpp/contrib/anchordict.h ---------------------------------------------------------------------- diff --git a/thirdparty/yaml-cpp-yaml-cpp-0.5.3/include/yaml-cpp/contrib/anchordict.h b/thirdparty/yaml-cpp-yaml-cpp-0.5.3/include/yaml-cpp/contrib/anchordict.h new file mode 100644 index 0000000..b4677e2 --- /dev/null +++ b/thirdparty/yaml-cpp-yaml-cpp-0.5.3/include/yaml-cpp/contrib/anchordict.h @@ -0,0 +1,37 @@ +#ifndef ANCHORDICT_H_62B23520_7C8E_11DE_8A39_0800200C9A66 +#define ANCHORDICT_H_62B23520_7C8E_11DE_8A39_0800200C9A66 + +#if defined(_MSC_VER) || \ + (defined(__GNUC__) && (__GNUC__ == 3 && __GNUC_MINOR__ >= 4) || \ + (__GNUC__ >= 4)) // GCC supports "pragma once" correctly since 3.4 +#pragma once +#endif + +#include <vector> + +#include "../anchor.h" + +namespace YAML { +/// AnchorDict +/// . An object that stores and retrieves values correlating to anchor_t +/// values. +/// . Efficient implementation that can make assumptions about how anchor_t +/// values are assigned by the Parser class. +template <class T> +class AnchorDict { + public: + void Register(anchor_t anchor, T value) { + if (anchor > m_data.size()) { + m_data.resize(anchor); + } + m_data[anchor - 1] = value; + } + + T Get(anchor_t anchor) const { return m_data[anchor - 1]; } + + private: + std::vector<T> m_data; +}; +} + +#endif // ANCHORDICT_H_62B23520_7C8E_11DE_8A39_0800200C9A66 http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/cf813d4d/thirdparty/yaml-cpp-yaml-cpp-0.5.3/include/yaml-cpp/contrib/graphbuilder.h ---------------------------------------------------------------------- diff --git a/thirdparty/yaml-cpp-yaml-cpp-0.5.3/include/yaml-cpp/contrib/graphbuilder.h b/thirdparty/yaml-cpp-yaml-cpp-0.5.3/include/yaml-cpp/contrib/graphbuilder.h new file mode 100644 index 0000000..7c2159b --- /dev/null +++ b/thirdparty/yaml-cpp-yaml-cpp-0.5.3/include/yaml-cpp/contrib/graphbuilder.h @@ -0,0 +1,147 @@ +#ifndef GRAPHBUILDER_H_62B23520_7C8E_11DE_8A39_0800200C9A66 +#define GRAPHBUILDER_H_62B23520_7C8E_11DE_8A39_0800200C9A66 + +#if defined(_MSC_VER) || \ + (defined(__GNUC__) && (__GNUC__ == 3 && __GNUC_MINOR__ >= 4) || \ + (__GNUC__ >= 4)) // GCC supports "pragma once" correctly since 3.4 +#pragma once +#endif + +#include "yaml-cpp/mark.h" +#include <string> + +namespace YAML { +class Parser; + +// GraphBuilderInterface +// . Abstraction of node creation +// . pParentNode is always NULL or the return value of one of the NewXXX() +// functions. +class GraphBuilderInterface { + public: + // Create and return a new node with a null value. + virtual void *NewNull(const Mark &mark, void *pParentNode) = 0; + + // Create and return a new node with the given tag and value. + virtual void *NewScalar(const Mark &mark, const std::string &tag, + void *pParentNode, const std::string &value) = 0; + + // Create and return a new sequence node + virtual void *NewSequence(const Mark &mark, const std::string &tag, + void *pParentNode) = 0; + + // Add pNode to pSequence. pNode was created with one of the NewXxx() + // functions and pSequence with NewSequence(). + virtual void AppendToSequence(void *pSequence, void *pNode) = 0; + + // Note that no moew entries will be added to pSequence + virtual void SequenceComplete(void *pSequence) { (void)pSequence; } + + // Create and return a new map node + virtual void *NewMap(const Mark &mark, const std::string &tag, + void *pParentNode) = 0; + + // Add the pKeyNode => pValueNode mapping to pMap. pKeyNode and pValueNode + // were created with one of the NewXxx() methods and pMap with NewMap(). + virtual void AssignInMap(void *pMap, void *pKeyNode, void *pValueNode) = 0; + + // Note that no more assignments will be made in pMap + virtual void MapComplete(void *pMap) { (void)pMap; } + + // Return the node that should be used in place of an alias referencing + // pNode (pNode by default) + virtual void *AnchorReference(const Mark &mark, void *pNode) { + (void)mark; + return pNode; + } +}; + +// Typesafe wrapper for GraphBuilderInterface. Assumes that Impl defines +// Node, Sequence, and Map types. Sequence and Map must derive from Node +// (unless Node is defined as void). Impl must also implement function with +// all of the same names as the virtual functions in GraphBuilderInterface +// -- including the ones with default implementations -- but with the +// prototypes changed to accept an explicit Node*, Sequence*, or Map* where +// appropriate. +template <class Impl> +class GraphBuilder : public GraphBuilderInterface { + public: + typedef typename Impl::Node Node; + typedef typename Impl::Sequence Sequence; + typedef typename Impl::Map Map; + + GraphBuilder(Impl &impl) : m_impl(impl) { + Map *pMap = NULL; + Sequence *pSeq = NULL; + Node *pNode = NULL; + + // Type consistency checks + pNode = pMap; + pNode = pSeq; + } + + GraphBuilderInterface &AsBuilderInterface() { return *this; } + + virtual void *NewNull(const Mark &mark, void *pParentNode) { + return CheckType<Node>(m_impl.NewNull(mark, AsNode(pParentNode))); + } + + virtual void *NewScalar(const Mark &mark, const std::string &tag, + void *pParentNode, const std::string &value) { + return CheckType<Node>( + m_impl.NewScalar(mark, tag, AsNode(pParentNode), value)); + } + + virtual void *NewSequence(const Mark &mark, const std::string &tag, + void *pParentNode) { + return CheckType<Sequence>( + m_impl.NewSequence(mark, tag, AsNode(pParentNode))); + } + virtual void AppendToSequence(void *pSequence, void *pNode) { + m_impl.AppendToSequence(AsSequence(pSequence), AsNode(pNode)); + } + virtual void SequenceComplete(void *pSequence) { + m_impl.SequenceComplete(AsSequence(pSequence)); + } + + virtual void *NewMap(const Mark &mark, const std::string &tag, + void *pParentNode) { + return CheckType<Map>(m_impl.NewMap(mark, tag, AsNode(pParentNode))); + } + virtual void AssignInMap(void *pMap, void *pKeyNode, void *pValueNode) { + m_impl.AssignInMap(AsMap(pMap), AsNode(pKeyNode), AsNode(pValueNode)); + } + virtual void MapComplete(void *pMap) { m_impl.MapComplete(AsMap(pMap)); } + + virtual void *AnchorReference(const Mark &mark, void *pNode) { + return CheckType<Node>(m_impl.AnchorReference(mark, AsNode(pNode))); + } + + private: + Impl &m_impl; + + // Static check for pointer to T + template <class T, class U> + static T *CheckType(U *p) { + return p; + } + + static Node *AsNode(void *pNode) { return static_cast<Node *>(pNode); } + static Sequence *AsSequence(void *pSeq) { + return static_cast<Sequence *>(pSeq); + } + static Map *AsMap(void *pMap) { return static_cast<Map *>(pMap); } +}; + +void *BuildGraphOfNextDocument(Parser &parser, + GraphBuilderInterface &graphBuilder); + +template <class Impl> +typename Impl::Node *BuildGraphOfNextDocument(Parser &parser, Impl &impl) { + GraphBuilder<Impl> graphBuilder(impl); + return static_cast<typename Impl::Node *>( + BuildGraphOfNextDocument(parser, graphBuilder)); +} +} + +#endif // GRAPHBUILDER_H_62B23520_7C8E_11DE_8A39_0800200C9A66
