http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/b02af540/libminifi/src/ListenSyslog.cpp ---------------------------------------------------------------------- diff --git a/libminifi/src/ListenSyslog.cpp b/libminifi/src/ListenSyslog.cpp new file mode 100644 index 0000000..ace37d7 --- /dev/null +++ b/libminifi/src/ListenSyslog.cpp @@ -0,0 +1,342 @@ +/** + * @file ListenSyslog.cpp + * ListenSyslog class implementation + * + * 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. + */ +#include <queue> +#include <stdio.h> +#include <string> +#include "TimeUtil.h" +#include "ListenSyslog.h" +#include "ProcessContext.h" +#include "ProcessSession.h" + +const std::string ListenSyslog::ProcessorName("ListenSyslog"); +Property ListenSyslog::RecvBufSize("Receive Buffer Size", "The size of each buffer used to receive Syslog messages.", "65507 B"); +Property ListenSyslog::MaxSocketBufSize("Max Size of Socket Buffer", "The maximum size of the socket buffer that should be used.", "1 MB"); +Property ListenSyslog::MaxConnections("Max Number of TCP Connections", "The maximum number of concurrent connections to accept Syslog messages in TCP mode.", "2"); +Property ListenSyslog::MaxBatchSize("Max Batch Size", + "The maximum number of Syslog events to add to a single FlowFile.", "1"); +Property ListenSyslog::MessageDelimiter("Message Delimiter", + "Specifies the delimiter to place between Syslog messages when multiple messages are bundled together (see <Max Batch Size> property).", "\n"); +Property ListenSyslog::ParseMessages("Parse Messages", + "Indicates if the processor should parse the Syslog messages. If set to false, each outgoing FlowFile will only.", "false"); +Property ListenSyslog::Protocol("Protocol", "The protocol for Syslog communication.", "UDP"); +Property ListenSyslog::Port("Port", "The port for Syslog communication.", "514"); +Relationship ListenSyslog::Success("success", "All files are routed to success"); +Relationship ListenSyslog::Invalid("invalid", "SysLog message format invalid"); + +void ListenSyslog::initialize() +{ + //! Set the supported properties + std::set<Property> properties; + properties.insert(RecvBufSize); + properties.insert(MaxSocketBufSize); + properties.insert(MaxConnections); + properties.insert(MaxBatchSize); + properties.insert(MessageDelimiter); + properties.insert(ParseMessages); + properties.insert(Protocol); + properties.insert(Port); + setSupportedProperties(properties); + //! Set the supported relationships + std::set<Relationship> relationships; + relationships.insert(Success); + relationships.insert(Invalid); + setSupportedRelationships(relationships); +} + +void ListenSyslog::startSocketThread() +{ + if (_thread != NULL) + return; + + _logger->log_info("ListenSysLog Socket Thread Start"); + _serverTheadRunning = true; + _thread = new std::thread(run, this); + _thread->detach(); +} + +void ListenSyslog::run(ListenSyslog *process) +{ + process->runThread(); +} + +void ListenSyslog::runThread() +{ + while (_serverTheadRunning) + { + if (_resetServerSocket) + { + _resetServerSocket = false; + // need to reset the socket + std::vector<int>::iterator it; + for (it = _clientSockets.begin(); it != _clientSockets.end(); ++it) + { + int clientSocket = *it; + close(clientSocket); + } + _clientSockets.clear(); + if (_serverSocket > 0) + { + close(_serverSocket); + _serverSocket = 0; + } + } + + if (_serverSocket <= 0) + { + uint16_t portno = _port; + struct sockaddr_in serv_addr; + int sockfd; + if (_protocol == "TCP") + sockfd = socket(AF_INET, SOCK_STREAM, 0); + else + sockfd = socket(AF_INET, SOCK_DGRAM, 0); + if (sockfd < 0) + { + _logger->log_info("ListenSysLog Server socket creation failed"); + break; + } + bzero((char *) &serv_addr, sizeof(serv_addr)); + serv_addr.sin_family = AF_INET; + serv_addr.sin_addr.s_addr = INADDR_ANY; + serv_addr.sin_port = htons(portno); + if (bind(sockfd, (struct sockaddr *) &serv_addr, + sizeof(serv_addr)) < 0) + { + _logger->log_error("ListenSysLog Server socket bind failed"); + break; + } + if (_protocol == "TCP") + listen(sockfd,5); + _serverSocket = sockfd; + _logger->log_error("ListenSysLog Server socket %d bind OK to port %d", _serverSocket, portno); + } + FD_ZERO(&_readfds); + FD_SET(_serverSocket, &_readfds); + _maxFds = _serverSocket; + std::vector<int>::iterator it; + for (it = _clientSockets.begin(); it != _clientSockets.end(); ++it) + { + int clientSocket = *it; + if (clientSocket >= _maxFds) + _maxFds = clientSocket; + FD_SET(clientSocket, &_readfds); + } + fd_set fds; + struct timeval tv; + int retval; + fds = _readfds; + tv.tv_sec = 0; + // 100 msec + tv.tv_usec = 100000; + retval = select(_maxFds+1, &fds, NULL, NULL, &tv); + if (retval < 0) + break; + if (retval == 0) + continue; + if (FD_ISSET(_serverSocket, &fds)) + { + // server socket, either we have UDP datagram or TCP connection request + if (_protocol == "TCP") + { + socklen_t clilen; + struct sockaddr_in cli_addr; + clilen = sizeof(cli_addr); + int newsockfd = accept(_serverSocket, + (struct sockaddr *) &cli_addr, + &clilen); + if (newsockfd > 0) + { + if (_clientSockets.size() < _maxConnections) + { + _clientSockets.push_back(newsockfd); + _logger->log_info("ListenSysLog new client socket %d connection", newsockfd); + continue; + } + else + { + close(newsockfd); + } + } + } + else + { + socklen_t clilen; + struct sockaddr_in cli_addr; + clilen = sizeof(cli_addr); + int recvlen = recvfrom(_serverSocket, _buffer, sizeof(_buffer), 0, + (struct sockaddr *)&cli_addr, &clilen); + if (recvlen > 0 && (recvlen + getEventQueueByteSize()) <= _recvBufSize) + { + uint8_t *payload = new uint8_t[recvlen]; + memcpy(payload, _buffer, recvlen); + putEvent(payload, recvlen); + } + } + } + it = _clientSockets.begin(); + while (it != _clientSockets.end()) + { + int clientSocket = *it; + if (FD_ISSET(clientSocket, &fds)) + { + int recvlen = readline(clientSocket, (char *)_buffer, sizeof(_buffer)); + if (recvlen <= 0) + { + close(clientSocket); + _logger->log_info("ListenSysLog client socket %d close", clientSocket); + it = _clientSockets.erase(it); + } + else + { + if ((recvlen + getEventQueueByteSize()) <= _recvBufSize) + { + uint8_t *payload = new uint8_t[recvlen]; + memcpy(payload, _buffer, recvlen); + putEvent(payload, recvlen); + } + ++it; + } + } + } + } + return; +} + + +int ListenSyslog::readline( int fd, char *bufptr, size_t len ) +{ + char *bufx = bufptr; + static char *bp; + static int cnt = 0; + static char b[ 2048 ]; + char c; + + while ( --len > 0 ) + { + if ( --cnt <= 0 ) + { + cnt = recv( fd, b, sizeof( b ), 0 ); + if ( cnt < 0 ) + { + if ( errno == EINTR ) + { + len++; /* the while will decrement */ + continue; + } + return -1; + } + if ( cnt == 0 ) + return 0; + bp = b; + } + c = *bp++; + *bufptr++ = c; + if ( c == '\n' ) + { + *bufptr = '\n'; + return bufptr - bufx + 1; + } + } + return -1; +} + +void ListenSyslog::onTrigger(ProcessContext *context, ProcessSession *session) +{ + std::string value; + bool needResetServerSocket = false; + if (context->getProperty(Protocol.getName(), value)) + { + if (_protocol != value) + needResetServerSocket = true; + _protocol = value; + } + if (context->getProperty(RecvBufSize.getName(), value)) + { + Property::StringToInt(value, _recvBufSize); + } + if (context->getProperty(MaxSocketBufSize.getName(), value)) + { + Property::StringToInt(value, _maxSocketBufSize); + } + if (context->getProperty(MaxConnections.getName(), value)) + { + Property::StringToInt(value, _maxConnections); + } + if (context->getProperty(MessageDelimiter.getName(), value)) + { + _messageDelimiter = value; + } + if (context->getProperty(ParseMessages.getName(), value)) + { + Property::StringToBool(value, _parseMessages); + } + if (context->getProperty(Port.getName(), value)) + { + int64_t oldPort = _port; + Property::StringToInt(value, _port); + if (_port != oldPort) + needResetServerSocket = true; + } + if (context->getProperty(MaxBatchSize.getName(), value)) + { + Property::StringToInt(value, _maxBatchSize); + } + + if (needResetServerSocket) + _resetServerSocket = true; + + startSocketThread(); + + // read from the event queue + if (isEventQueueEmpty()) + { + context->yield(); + return; + } + + std::queue<SysLogEvent> eventQueue; + pollEvent(eventQueue, _maxBatchSize); + bool firstEvent = true; + FlowFileRecord *flowFile = NULL; + while(!eventQueue.empty()) + { + SysLogEvent event = eventQueue.front(); + eventQueue.pop(); + if (firstEvent) + { + flowFile = session->create(); + if (!flowFile) + return; + ListenSyslog::WriteCallback callback((char *)event.payload, event.len); + session->write(flowFile, &callback); + delete[] event.payload; + firstEvent = false; + } + else + { + ListenSyslog::WriteCallback callback((char *)event.payload, event.len); + session->append(flowFile, &callback); + delete[] event.payload; + } + } + flowFile->addAttribute("syslog.protocol", _protocol); + flowFile->addAttribute("syslog.port", std::to_string(_port)); + session->transfer(flowFile, Success); +}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/b02af540/libminifi/src/LogAttribute.cpp ---------------------------------------------------------------------- diff --git a/libminifi/src/LogAttribute.cpp b/libminifi/src/LogAttribute.cpp new file mode 100644 index 0000000..82130f8 --- /dev/null +++ b/libminifi/src/LogAttribute.cpp @@ -0,0 +1,158 @@ +/** + * @file LogAttribute.cpp + * LogAttribute class implementation + * + * 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. + */ +#include <vector> +#include <queue> +#include <map> +#include <set> +#include <sys/time.h> +#include <time.h> +#include <sstream> +#include <string.h> +#include <iostream> + +#include "TimeUtil.h" +#include "LogAttribute.h" +#include "ProcessContext.h" +#include "ProcessSession.h" + +const std::string LogAttribute::ProcessorName("LogAttribute"); +Property LogAttribute::LogLevel("Log Level", "The Log Level to use when logging the Attributes", "info"); +Property LogAttribute::AttributesToLog("Attributes to Log", "A comma-separated list of Attributes to Log. If not specified, all attributes will be logged.", ""); +Property LogAttribute::AttributesToIgnore("Attributes to Ignore", "A comma-separated list of Attributes to ignore. If not specified, no attributes will be ignored.", ""); +Property LogAttribute::LogPayload("Log Payload", + "If true, the FlowFile's payload will be logged, in addition to its attributes; otherwise, just the Attributes will be logged.", "false"); +Property LogAttribute::LogPrefix("Log prefix", + "Log prefix appended to the log lines. It helps to distinguish the output of multiple LogAttribute processors.", ""); +Relationship LogAttribute::Success("success", "success operational on the flow record"); + +void LogAttribute::initialize() +{ + //! Set the supported properties + std::set<Property> properties; + properties.insert(LogLevel); + properties.insert(AttributesToLog); + properties.insert(AttributesToIgnore); + properties.insert(LogPayload); + properties.insert(LogPrefix); + setSupportedProperties(properties); + //! Set the supported relationships + std::set<Relationship> relationships; + relationships.insert(Success); + setSupportedRelationships(relationships); +} + +void LogAttribute::onTrigger(ProcessContext *context, ProcessSession *session) +{ + std::string dashLine = "--------------------------------------------------"; + LogAttrLevel level = LogAttrLevelInfo; + bool logPayload = false; + std::ostringstream message; + + FlowFileRecord *flow = session->get(); + + if (!flow) + return; + + std::string value; + if (context->getProperty(LogLevel.getName(), value)) + { + logLevelStringToEnum(value, level); + } + if (context->getProperty(LogPrefix.getName(), value)) + { + dashLine = "-----" + value + "-----"; + } + if (context->getProperty(LogPayload.getName(), value)) + { + Property::StringToBool(value, logPayload); + } + + message << "Logging for flow file " << "\n"; + message << dashLine; + message << "\nStandard FlowFile Attributes"; + message << "\n" << "UUID:" << flow->getUUIDStr(); + message << "\n" << "EntryDate:" << getTimeStr(flow->getEntryDate()); + message << "\n" << "lineageStartDate:" << getTimeStr(flow->getlineageStartDate()); + message << "\n" << "Size:" << flow->getSize() << " Offset:" << flow->getOffset(); + message << "\nFlowFile Attributes Map Content"; + std::map<std::string, std::string> attrs = flow->getAttributes(); + std::map<std::string, std::string>::iterator it; + for (it = attrs.begin(); it!= attrs.end(); it++) + { + message << "\n" << "key:" << it->first << " value:" << it->second; + } + message << "\nFlowFile Resource Claim Content"; + ResourceClaim *claim = flow->getResourceClaim(); + if (claim) + { + message << "\n" << "Content Claim:" << claim->getContentFullPath(); + } + if (logPayload && flow->getSize() <= 1024*1024) + { + message << "\n" << "Payload:" << "\n"; + ReadCallback callback(flow->getSize()); + session->read(flow, &callback); + for (unsigned int i = 0, j = 0; i < callback._readSize; i++) + { + char temp[8]; + sprintf(temp, "%02x ", (unsigned char) (callback._buffer[i])); + message << temp; + j++; + if (j == 16) + { + message << '\n'; + j = 0; + } + } + } + message << "\n" << dashLine << std::ends; + std::string output = message.str(); + + switch (level) + { + case LogAttrLevelInfo: + _logger->log_info("%s", output.c_str()); + break; + case LogAttrLevelDebug: + _logger->log_debug("%s", output.c_str()); + break; + case LogAttrLevelError: + _logger->log_error("%s", output.c_str()); + break; + case LogAttrLevelTrace: + _logger->log_trace("%s", output.c_str()); + break; + case LogAttrLevelWarn: + _logger->log_warn("%s", output.c_str()); + break; + default: + break; + } + + // Test Import + /* + FlowFileRecord *importRecord = session->create(); + session->import(claim->getContentFullPath(), importRecord); + session->transfer(importRecord, Success); */ + + + // Transfer to the relationship + session->transfer(flow, Success); +} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/b02af540/libminifi/src/Logger.cpp ---------------------------------------------------------------------- diff --git a/libminifi/src/Logger.cpp b/libminifi/src/Logger.cpp new file mode 100644 index 0000000..984f609 --- /dev/null +++ b/libminifi/src/Logger.cpp @@ -0,0 +1,27 @@ +/** + * @file Logger.cpp + * Logger class implementation + * + * 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. + */ +#include <vector> +#include <queue> +#include <map> + +#include "Logger.h" + +Logger *Logger::_logger(NULL); + http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/b02af540/libminifi/src/ProcessGroup.cpp ---------------------------------------------------------------------- diff --git a/libminifi/src/ProcessGroup.cpp b/libminifi/src/ProcessGroup.cpp new file mode 100644 index 0000000..70ee9d7 --- /dev/null +++ b/libminifi/src/ProcessGroup.cpp @@ -0,0 +1,314 @@ +/** + * @file ProcessGroup.cpp + * ProcessGroup class implementation + * + * 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. + */ +#include <vector> +#include <queue> +#include <map> +#include <set> +#include <sys/time.h> +#include <time.h> +#include <chrono> +#include <thread> + +#include "ProcessGroup.h" +#include "Processor.h" + +ProcessGroup::ProcessGroup(ProcessGroupType type, std::string name, uuid_t uuid, ProcessGroup *parent) +: _name(name), + _type(type), + _parentProcessGroup(parent) +{ + if (!uuid) + // Generate the global UUID for the flow record + uuid_generate(_uuid); + else + uuid_copy(_uuid, uuid); + + _yieldPeriodMsec = 0; + _transmitting = false; + + _logger = Logger::getLogger(); + _logger->log_info("ProcessGroup %s created", _name.c_str()); +} + +ProcessGroup::~ProcessGroup() +{ + for (std::set<Connection *>::iterator it = _connections.begin(); it != _connections.end(); ++it) + { + Connection *connection = *it; + connection->drain(); + delete connection; + } + + for (std::set<ProcessGroup *>::iterator it = _childProcessGroups.begin(); it != _childProcessGroups.end(); ++it) + { + ProcessGroup *processGroup(*it); + delete processGroup; + } + + for (std::set<Processor *>::iterator it = _processors.begin(); it != _processors.end(); ++it) + { + Processor *processor(*it); + delete processor; + } +} + +bool ProcessGroup::isRootProcessGroup() +{ + std::lock_guard<std::mutex> lock(_mtx); + return (_type == ROOT_PROCESS_GROUP); +} + +void ProcessGroup::addProcessor(Processor *processor) +{ + std::lock_guard<std::mutex> lock(_mtx); + + if (_processors.find(processor) == _processors.end()) + { + // We do not have the same processor in this process group yet + _processors.insert(processor); + _logger->log_info("Add processor %s into process group %s", + processor->getName().c_str(), _name.c_str()); + } +} + +void ProcessGroup::removeProcessor(Processor *processor) +{ + std::lock_guard<std::mutex> lock(_mtx); + + if (_processors.find(processor) != _processors.end()) + { + // We do have the same processor in this process group yet + _processors.erase(processor); + _logger->log_info("Remove processor %s from process group %s", + processor->getName().c_str(), _name.c_str()); + } +} + +void ProcessGroup::addProcessGroup(ProcessGroup *child) +{ + std::lock_guard<std::mutex> lock(_mtx); + + if (_childProcessGroups.find(child) == _childProcessGroups.end()) + { + // We do not have the same child process group in this process group yet + _childProcessGroups.insert(child); + _logger->log_info("Add child process group %s into process group %s", + child->getName().c_str(), _name.c_str()); + } +} + +void ProcessGroup::removeProcessGroup(ProcessGroup *child) +{ + std::lock_guard<std::mutex> lock(_mtx); + + if (_childProcessGroups.find(child) != _childProcessGroups.end()) + { + // We do have the same child process group in this process group yet + _childProcessGroups.erase(child); + _logger->log_info("Remove child process group %s from process group %s", + child->getName().c_str(), _name.c_str()); + } +} + +void ProcessGroup::startProcessing(TimerDrivenSchedulingAgent *timeScheduler) +{ + std::lock_guard<std::mutex> lock(_mtx); + + try + { + // Start all the processor node, input and output ports + for (std::set<Processor *>::iterator it = _processors.begin(); it != _processors.end(); ++it) + { + Processor *processor(*it); + if (!processor->isRunning() && processor->getScheduledState() != DISABLED) + { + if (processor->getSchedulingStrategy() == TIMER_DRIVEN) + timeScheduler->schedule(processor); + } + } + + for (std::set<ProcessGroup *>::iterator it = _childProcessGroups.begin(); it != _childProcessGroups.end(); ++it) + { + ProcessGroup *processGroup(*it); + processGroup->startProcessing(timeScheduler); + } + } + catch (std::exception &exception) + { + _logger->log_debug("Caught Exception %s", exception.what()); + throw; + } + catch (...) + { + _logger->log_debug("Caught Exception during process group start processing"); + throw; + } +} + +void ProcessGroup::stopProcessing(TimerDrivenSchedulingAgent *timeScheduler) +{ + std::lock_guard<std::mutex> lock(_mtx); + + try + { + // Stop all the processor node, input and output ports + for (std::set<Processor *>::iterator it = _processors.begin(); it != _processors.end(); ++it) + { + Processor *processor(*it); + if (processor->getSchedulingStrategy() == TIMER_DRIVEN) + timeScheduler->unschedule(processor); + } + + for (std::set<ProcessGroup *>::iterator it = _childProcessGroups.begin(); it != _childProcessGroups.end(); ++it) + { + ProcessGroup *processGroup(*it); + processGroup->stopProcessing(timeScheduler); + } + } + catch (std::exception &exception) + { + _logger->log_debug("Caught Exception %s", exception.what()); + throw; + } + catch (...) + { + _logger->log_debug("Caught Exception during process group stop processing"); + throw; + } +} + +Processor *ProcessGroup::findProcessor(uuid_t uuid) +{ + Processor *ret = NULL; + // std::lock_guard<std::mutex> lock(_mtx); + + for (std::set<Processor *>::iterator it = _processors.begin(); it != _processors.end(); ++it) + { + Processor *processor(*it); + uuid_t processorUUID; + if (processor->getUUID(processorUUID) && uuid_compare(processorUUID, uuid) == 0) + return processor; + } + + for (std::set<ProcessGroup *>::iterator it = _childProcessGroups.begin(); it != _childProcessGroups.end(); ++it) + { + ProcessGroup *processGroup(*it); + Processor *processor = processGroup->findProcessor(uuid); + if (processor) + return processor; + } + + 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); + + for (std::set<Processor *>::iterator it = _processors.begin(); it != _processors.end(); ++it) + { + Processor *processor(*it); + if (processor->getName() == processorName) + { + processor->setProperty(propertyName, propertyValue); + } + } + + for (std::set<ProcessGroup *>::iterator it = _childProcessGroups.begin(); it != _childProcessGroups.end(); ++it) + { + ProcessGroup *processGroup(*it); + processGroup->updatePropertyValue(processorName, propertyName, propertyValue); + } + + return; +} + +void ProcessGroup::addConnection(Connection *connection) +{ + std::lock_guard<std::mutex> lock(_mtx); + + if (_connections.find(connection) == _connections.end()) + { + // We do not have the same connection in this process group yet + _connections.insert(connection); + _logger->log_info("Add connection %s into process group %s", + connection->getName().c_str(), _name.c_str()); + uuid_t sourceUUID; + Processor *source = NULL; + connection->getSourceProcessorUUID(sourceUUID); + source = this->findProcessor(sourceUUID); + if (source) + source->addConnection(connection); + Processor *destination = NULL; + uuid_t destinationUUID; + connection->getDestinationProcessorUUID(destinationUUID); + destination = this->findProcessor(destinationUUID); + if (destination && destination != source) + destination->addConnection(connection); + } +} + +void ProcessGroup::removeConnection(Connection *connection) +{ + std::lock_guard<std::mutex> lock(_mtx); + + if (_connections.find(connection) != _connections.end()) + { + // We do not have the same connection in this process group yet + _connections.erase(connection); + _logger->log_info("Remove connection %s into process group %s", + connection->getName().c_str(), _name.c_str()); + uuid_t sourceUUID; + Processor *source = NULL; + connection->getSourceProcessorUUID(sourceUUID); + source = this->findProcessor(sourceUUID); + if (source) + source->removeConnection(connection); + Processor *destination = NULL; + uuid_t destinationUUID; + connection->getDestinationProcessorUUID(destinationUUID); + destination = this->findProcessor(destinationUUID); + if (destination && destination != source) + destination->removeConnection(connection); + } +} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/b02af540/libminifi/src/ProcessSession.cpp ---------------------------------------------------------------------- diff --git a/libminifi/src/ProcessSession.cpp b/libminifi/src/ProcessSession.cpp new file mode 100644 index 0000000..4f526c3 --- /dev/null +++ b/libminifi/src/ProcessSession.cpp @@ -0,0 +1,731 @@ +/** + * @file ProcessSession.cpp + * ProcessSession class implementation + * + * 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. + */ +#include <vector> +#include <queue> +#include <map> +#include <set> +#include <sys/time.h> +#include <time.h> +#include <chrono> +#include <thread> +#include <iostream> + +#include "ProcessSession.h" + +FlowFileRecord* ProcessSession::create() +{ + std::map<std::string, std::string> empty; + FlowFileRecord *record = new FlowFileRecord(empty); + + if (record) + { + _addedFlowFiles[record->getUUIDStr()] = record; + _logger->log_debug("Create FlowFile with UUID %s", record->getUUIDStr().c_str()); + } + + return record; +} + +FlowFileRecord* ProcessSession::create(FlowFileRecord *parent) +{ + FlowFileRecord *record = this->create(); + if (record) + { + // Copy attributes + std::map<std::string, std::string> parentAttributes = parent->getAttributes(); + std::map<std::string, std::string>::iterator it; + for (it = parentAttributes.begin(); it!= parentAttributes.end(); it++) + { + if (it->first == FlowAttributeKey(ALTERNATE_IDENTIFIER) || + it->first == FlowAttributeKey(DISCARD_REASON) || + it->first == FlowAttributeKey(UUID)) + // Do not copy special attributes from parent + continue; + record->setAttribute(it->first, it->second); + } + record->_lineageStartDate = parent->_lineageStartDate; + record->_lineageIdentifiers = parent->_lineageIdentifiers; + record->_lineageIdentifiers.insert(parent->_uuidStr); + + } + return record; +} + +FlowFileRecord* ProcessSession::clone(FlowFileRecord *parent) +{ + FlowFileRecord *record = this->create(parent); + if (record) + { + // Copy Resource Claim + record->_claim = parent->_claim; + if (record->_claim) + { + record->_offset = parent->_offset; + record->_size = parent->_size; + record->_claim->increaseFlowFileRecordOwnedCount(); + } + } + return record; +} + +FlowFileRecord* ProcessSession::cloneDuringTransfer(FlowFileRecord *parent) +{ + std::map<std::string, std::string> empty; + FlowFileRecord *record = new FlowFileRecord(empty); + + if (record) + { + this->_clonedFlowFiles[record->getUUIDStr()] = record; + _logger->log_debug("Clone FlowFile with UUID %s during transfer", record->getUUIDStr().c_str()); + // Copy attributes + std::map<std::string, std::string> parentAttributes = parent->getAttributes(); + std::map<std::string, std::string>::iterator it; + for (it = parentAttributes.begin(); it!= parentAttributes.end(); it++) + { + if (it->first == FlowAttributeKey(ALTERNATE_IDENTIFIER) || + it->first == FlowAttributeKey(DISCARD_REASON) || + it->first == FlowAttributeKey(UUID)) + // Do not copy special attributes from parent + continue; + record->setAttribute(it->first, it->second); + } + record->_lineageStartDate = parent->_lineageStartDate; + record->_lineageIdentifiers = parent->_lineageIdentifiers; + record->_lineageIdentifiers.insert(parent->_uuidStr); + + // Copy Resource Claim + record->_claim = parent->_claim; + if (record->_claim) + { + record->_offset = parent->_offset; + record->_size = parent->_size; + record->_claim->increaseFlowFileRecordOwnedCount(); + } + } + + return record; +} + +FlowFileRecord* ProcessSession::clone(FlowFileRecord *parent, long offset, long size) +{ + FlowFileRecord *record = this->create(parent); + if (record) + { + if (parent->_claim) + { + if ((offset + size) > (long) parent->_size) + { + // Set offset and size + _logger->log_error("clone offset %d and size %d exceed parent size %d", + offset, size, parent->_size); + // Remove the Add FlowFile for the session + std::map<std::string, FlowFileRecord *>::iterator it = + this->_addedFlowFiles.find(record->getUUIDStr()); + if (it != this->_addedFlowFiles.end()) + this->_addedFlowFiles.erase(record->getUUIDStr()); + delete record; + return NULL; + } + record->_offset = parent->_offset + parent->_offset; + record->_size = size; + // Copy Resource Claim + record->_claim = parent->_claim; + record->_claim->increaseFlowFileRecordOwnedCount(); + } + } + return record; +} + +void ProcessSession::remove(FlowFileRecord *flow) +{ + flow->_markedDelete = true; + _deletedFlowFiles[flow->getUUIDStr()] = flow; +} + +void ProcessSession::putAttribute(FlowFileRecord *flow, std::string key, std::string value) +{ + flow->setAttribute(key, value); +} + +void ProcessSession::removeAttribute(FlowFileRecord *flow, std::string key) +{ + flow->removeAttribute(key); +} + +void ProcessSession::penalize(FlowFileRecord *flow) +{ + flow->_penaltyExpirationMs = getTimeMillis() + this->_processContext->getProcessor()->getPenalizationPeriodMsec(); +} + +void ProcessSession::transfer(FlowFileRecord *flow, Relationship relationship) +{ + _transferRelationship[flow->getUUIDStr()] = relationship; +} + +void ProcessSession::write(FlowFileRecord *flow, OutputStreamCallback *callback) +{ + ResourceClaim *claim = NULL; + + claim = new ResourceClaim(DEFAULT_CONTENT_DIRECTORY); + + try + { + std::ofstream fs; + fs.open(claim->getContentFullPath().c_str(), std::fstream::out | std::fstream::binary | std::fstream::trunc); + if (fs.is_open()) + { + // Call the callback to write the content + callback->process(&fs); + if (fs.good() && fs.tellp() >= 0) + { + flow->_size = fs.tellp(); + flow->_offset = 0; + if (flow->_claim) + { + // Remove the old claim + flow->_claim->decreaseFlowFileRecordOwnedCount(); + flow->_claim = NULL; + } + flow->_claim = claim; + claim->increaseFlowFileRecordOwnedCount(); + /* + _logger->log_debug("Write offset %d length %d into content %s for FlowFile UUID %s", + flow->_offset, flow->_size, flow->_claim->getContentFullPath().c_str(), flow->getUUIDStr().c_str()); */ + fs.close(); + } + else + { + fs.close(); + throw Exception(FILE_OPERATION_EXCEPTION, "File Write Error"); + } + } + else + { + throw Exception(FILE_OPERATION_EXCEPTION, "File Open Error"); + } + } + catch (std::exception &exception) + { + if (flow && flow->_claim == claim) + { + flow->_claim->decreaseFlowFileRecordOwnedCount(); + flow->_claim = NULL; + } + if (claim) + delete claim; + _logger->log_debug("Caught Exception %s", exception.what()); + throw; + } + catch (...) + { + if (flow && flow->_claim == claim) + { + flow->_claim->decreaseFlowFileRecordOwnedCount(); + flow->_claim = NULL; + } + if (claim) + delete claim; + _logger->log_debug("Caught Exception during process session write"); + throw; + } +} + +void ProcessSession::append(FlowFileRecord *flow, OutputStreamCallback *callback) +{ + ResourceClaim *claim = NULL; + + if (flow->_claim == NULL) + { + // No existed claim for append, we need to create new claim + return write(flow, callback); + } + + claim = flow->_claim; + + try + { + std::ofstream fs; + fs.open(claim->getContentFullPath().c_str(), std::fstream::out | std::fstream::binary | std::fstream::app); + if (fs.is_open()) + { + // Call the callback to write the content + std::streampos oldPos = fs.tellp(); + callback->process(&fs); + if (fs.good() && fs.tellp() >= 0) + { + uint64_t appendSize = fs.tellp() - oldPos; + flow->_size += appendSize; + /* + _logger->log_debug("Append offset %d extra length %d to new size %d into content %s for FlowFile UUID %s", + flow->_offset, appendSize, flow->_size, claim->getContentFullPath().c_str(), flow->getUUIDStr().c_str()); */ + fs.close(); + } + else + { + fs.close(); + throw Exception(FILE_OPERATION_EXCEPTION, "File Write Error"); + } + } + else + { + throw Exception(FILE_OPERATION_EXCEPTION, "File Open Error"); + } + } + catch (std::exception &exception) + { + _logger->log_debug("Caught Exception %s", exception.what()); + throw; + } + catch (...) + { + _logger->log_debug("Caught Exception during process session append"); + throw; + } +} + +void ProcessSession::read(FlowFileRecord *flow, InputStreamCallback *callback) +{ + try + { + ResourceClaim *claim = NULL; + if (flow->_claim == NULL) + { + // No existed claim for read, we throw exception + throw Exception(FILE_OPERATION_EXCEPTION, "No Content Claim existed for read"); + } + + claim = flow->_claim; + std::ifstream fs; + fs.open(claim->getContentFullPath().c_str(), std::fstream::in | std::fstream::binary); + if (fs.is_open()) + { + fs.seekg(flow->_offset, fs.beg); + + if (fs.good()) + { + callback->process(&fs); + /* + _logger->log_debug("Read offset %d size %d content %s for FlowFile UUID %s", + flow->_offset, flow->_size, claim->getContentFullPath().c_str(), flow->getUUIDStr().c_str()); */ + fs.close(); + } + else + { + fs.close(); + throw Exception(FILE_OPERATION_EXCEPTION, "File Read Error"); + } + } + else + { + throw Exception(FILE_OPERATION_EXCEPTION, "File Open Error"); + } + } + catch (std::exception &exception) + { + _logger->log_debug("Caught Exception %s", exception.what()); + throw; + } + catch (...) + { + _logger->log_debug("Caught Exception during process session read"); + throw; + } +} + +void ProcessSession::import(std::string source, FlowFileRecord *flow, bool keepSource, uint64_t offset) +{ + ResourceClaim *claim = NULL; + + claim = new ResourceClaim(DEFAULT_CONTENT_DIRECTORY); + char *buf = NULL; + int size = 4096; + buf = new char [size]; + + try + { + std::ofstream fs; + fs.open(claim->getContentFullPath().c_str(), std::fstream::out | std::fstream::binary | std::fstream::trunc); + std::ifstream input; + input.open(source.c_str(), std::fstream::in | std::fstream::binary); + + if (fs.is_open() && input.is_open()) + { + // Open the source file and stream to the flow file + input.seekg(offset, fs.beg); + while (input.good()) + { + input.read(buf, size); + if (input) + fs.write(buf, size); + else + fs.write(buf, input.gcount()); + } + + if (fs.good() && fs.tellp() >= 0) + { + flow->_size = fs.tellp(); + flow->_offset = 0; + if (flow->_claim) + { + // Remove the old claim + flow->_claim->decreaseFlowFileRecordOwnedCount(); + flow->_claim = NULL; + } + flow->_claim = claim; + claim->increaseFlowFileRecordOwnedCount(); + /* + _logger->log_debug("Import offset %d length %d into content %s for FlowFile UUID %s", + flow->_offset, flow->_size, flow->_claim->getContentFullPath().c_str(), flow->getUUIDStr().c_str()); */ + fs.close(); + input.close(); + if (!keepSource) + std::remove(source.c_str()); + } + else + { + fs.close(); + input.close(); + throw Exception(FILE_OPERATION_EXCEPTION, "File Import Error"); + } + } + else + { + throw Exception(FILE_OPERATION_EXCEPTION, "File Import Error"); + } + + delete[] buf; + } + catch (std::exception &exception) + { + if (flow && flow->_claim == claim) + { + flow->_claim->decreaseFlowFileRecordOwnedCount(); + flow->_claim = NULL; + } + if (claim) + delete claim; + _logger->log_debug("Caught Exception %s", exception.what()); + delete[] buf; + throw; + } + catch (...) + { + if (flow && flow->_claim == claim) + { + flow->_claim->decreaseFlowFileRecordOwnedCount(); + flow->_claim = NULL; + } + if (claim) + delete claim; + _logger->log_debug("Caught Exception during process session write"); + delete[] buf; + throw; + } +} + +void ProcessSession::commit() +{ + try + { + // First we clone the flow record based on the transfered relationship for updated flow record + std::map<std::string, FlowFileRecord *>::iterator it; + for (it = _updatedFlowFiles.begin(); it!= _updatedFlowFiles.end(); it++) + { + FlowFileRecord *record = it->second; + if (record->_markedDelete) + continue; + std::map<std::string, Relationship>::iterator itRelationship = + this->_transferRelationship.find(record->getUUIDStr()); + if (itRelationship != _transferRelationship.end()) + { + Relationship relationship = itRelationship->second; + // Find the relationship, we need to find the connections for that relationship + std::set<Connection *> connections = + _processContext->getProcessor()->getOutGoingConnections(relationship.getName()); + if (connections.empty()) + { + // No connection + if (!_processContext->getProcessor()->isAutoTerminated(relationship)) + { + // Not autoterminate, we should have the connect + std::string message = "Connect empty for non auto terminated relationship" + relationship.getName(); + throw Exception(PROCESS_SESSION_EXCEPTION, message.c_str()); + } + else + { + // Autoterminated + remove(record); + } + } + else + { + // We connections, clone the flow and assign the connection accordingly + for (std::set<Connection *>::iterator itConnection = connections.begin(); itConnection != connections.end(); ++itConnection) + { + Connection *connection(*itConnection); + if (itConnection == connections.begin()) + { + // First connection which the flow need be routed to + record->_connection = connection; + } + else + { + // Clone the flow file and route to the connection + FlowFileRecord *cloneRecord; + cloneRecord = this->cloneDuringTransfer(record); + if (cloneRecord) + cloneRecord->_connection = connection; + else + throw Exception(PROCESS_SESSION_EXCEPTION, "Can not clone the flow for transfer"); + } + } + } + } + else + { + // Can not find relationship for the flow + throw Exception(PROCESS_SESSION_EXCEPTION, "Can not find the transfer relationship for the flow"); + } + } + + // Do the samething for added flow file + for (it = _addedFlowFiles.begin(); it!= _addedFlowFiles.end(); it++) + { + FlowFileRecord *record = it->second; + if (record->_markedDelete) + continue; + std::map<std::string, Relationship>::iterator itRelationship = + this->_transferRelationship.find(record->getUUIDStr()); + if (itRelationship != _transferRelationship.end()) + { + Relationship relationship = itRelationship->second; + // Find the relationship, we need to find the connections for that relationship + std::set<Connection *> connections = + _processContext->getProcessor()->getOutGoingConnections(relationship.getName()); + if (connections.empty()) + { + // No connection + if (!_processContext->getProcessor()->isAutoTerminated(relationship)) + { + // Not autoterminate, we should have the connect + std::string message = "Connect empty for non auto terminated relationship " + relationship.getName(); + throw Exception(PROCESS_SESSION_EXCEPTION, message.c_str()); + } + else + { + // Autoterminated + remove(record); + } + } + else + { + // We connections, clone the flow and assign the connection accordingly + for (std::set<Connection *>::iterator itConnection = connections.begin(); itConnection != connections.end(); ++itConnection) + { + Connection *connection(*itConnection); + if (itConnection == connections.begin()) + { + // First connection which the flow need be routed to + record->_connection = connection; + } + else + { + // Clone the flow file and route to the connection + FlowFileRecord *cloneRecord; + cloneRecord = this->cloneDuringTransfer(record); + if (cloneRecord) + cloneRecord->_connection = connection; + else + throw Exception(PROCESS_SESSION_EXCEPTION, "Can not clone the flow for transfer"); + } + } + } + } + else + { + // Can not find relationship for the flow + throw Exception(PROCESS_SESSION_EXCEPTION, "Can not find the transfer relationship for the flow"); + } + } + + // Complete process the added and update flow files for the session, send the flow file to its queue + for (it = _updatedFlowFiles.begin(); it!= _updatedFlowFiles.end(); it++) + { + FlowFileRecord *record = it->second; + if (record->_markedDelete) + { + continue; + } + if (record->_connection) + record->_connection->put(record); + else + delete record; + } + for (it = _addedFlowFiles.begin(); it!= _addedFlowFiles.end(); it++) + { + FlowFileRecord *record = it->second; + if (record->_markedDelete) + { + continue; + } + if (record->_connection) + record->_connection->put(record); + else + delete record; + } + // Process the clone flow files + for (it = _clonedFlowFiles.begin(); it!= _clonedFlowFiles.end(); it++) + { + FlowFileRecord *record = it->second; + if (record->_markedDelete) + { + continue; + } + if (record->_connection) + record->_connection->put(record); + else + delete record; + } + // Delete the deleted flow files + for (it = _deletedFlowFiles.begin(); it!= _deletedFlowFiles.end(); it++) + { + FlowFileRecord *record = it->second; + delete record; + } + // Delete the snapshot + for (it = _originalFlowFiles.begin(); it!= _originalFlowFiles.end(); it++) + { + FlowFileRecord *record = it->second; + delete record; + } + // All done + _updatedFlowFiles.clear(); + _addedFlowFiles.clear(); + _clonedFlowFiles.clear(); + _deletedFlowFiles.clear(); + _originalFlowFiles.clear(); + _logger->log_trace("ProcessSession committed for %s", _processContext->getProcessor()->getName().c_str()); + } + catch (std::exception &exception) + { + _logger->log_debug("Caught Exception %s", exception.what()); + throw; + } + catch (...) + { + _logger->log_debug("Caught Exception during process session commit"); + throw; + } +} + + +void ProcessSession::rollback() +{ + try + { + std::map<std::string, FlowFileRecord *>::iterator it; + // Requeue the snapshot of the flowfile back + for (it = _originalFlowFiles.begin(); it!= _originalFlowFiles.end(); it++) + { + FlowFileRecord *record = it->second; + if (record->_orginalConnection) + { + record->_snapshot = false; + record->_orginalConnection->put(record); + } + else + delete record; + } + _originalFlowFiles.clear(); + // Process the clone flow files + for (it = _clonedFlowFiles.begin(); it!= _clonedFlowFiles.end(); it++) + { + FlowFileRecord *record = it->second; + delete record; + } + _clonedFlowFiles.clear(); + for (it = _addedFlowFiles.begin(); it!= _addedFlowFiles.end(); it++) + { + FlowFileRecord *record = it->second; + delete record; + } + _addedFlowFiles.clear(); + for (it = _updatedFlowFiles.begin(); it!= _updatedFlowFiles.end(); it++) + { + FlowFileRecord *record = it->second; + delete record; + } + _updatedFlowFiles.clear(); + _deletedFlowFiles.clear(); + _logger->log_trace("ProcessSession rollback for %s", _processContext->getProcessor()->getName().c_str()); + } + catch (std::exception &exception) + { + _logger->log_debug("Caught Exception %s", exception.what()); + throw; + } + catch (...) + { + _logger->log_debug("Caught Exception during process session roll back"); + throw; + } +} + +FlowFileRecord *ProcessSession::get() +{ + Connection *first = _processContext->getProcessor()->getNextIncomingConnection(); + + if (first == NULL) + return NULL; + + Connection *current = first; + + do + { + std::set<FlowFileRecord *> expired; + FlowFileRecord *ret = current->poll(expired); + if (expired.size() > 0) + { + // Remove expired flow record + for (std::set<FlowFileRecord *>::iterator it = expired.begin(); it != expired.end(); ++it) + { + delete (*it); + } + } + if (ret) + { + // add the flow record to the current process session update map + ret->_markedDelete = false; + _updatedFlowFiles[ret->getUUIDStr()] = ret; + std::map<std::string, std::string> empty; + FlowFileRecord *snapshot = new FlowFileRecord(empty); + _logger->log_debug("Create Snapshot FlowFile with UUID %s", snapshot->getUUIDStr().c_str()); + snapshot->duplicate(ret); + // save a snapshot + _originalFlowFiles[snapshot->getUUIDStr()] = snapshot; + return ret; + } + current = _processContext->getProcessor()->getNextIncomingConnection(); + } + while (current != NULL && current != first); + + return NULL; +} + http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/b02af540/libminifi/src/Processor.cpp ---------------------------------------------------------------------- diff --git a/libminifi/src/Processor.cpp b/libminifi/src/Processor.cpp new file mode 100644 index 0000000..cc136dc --- /dev/null +++ b/libminifi/src/Processor.cpp @@ -0,0 +1,451 @@ +/** + * @file Processor.cpp + * Processor class implementation + * + * 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. + */ +#include <vector> +#include <queue> +#include <map> +#include <set> +#include <sys/time.h> +#include <time.h> +#include <chrono> +#include <thread> + +#include "Processor.h" +#include "ProcessContext.h" +#include "ProcessSession.h" + +Processor::Processor(std::string name, uuid_t uuid) +: _name(name) +{ + if (!uuid) + // Generate the global UUID for the flow record + uuid_generate(_uuid); + else + uuid_copy(_uuid, uuid); + + char uuidStr[37]; + uuid_unparse(_uuid, uuidStr); + _uuidStr = uuidStr; + + // Setup the default values + _state = DISABLED; + _strategy = TIMER_DRIVEN; + _lossTolerant = false; + _triggerWhenEmpty = false; + _schedulingPeriodNano = MINIMUM_SCHEDULING_NANOS; + _runDurantionNano = 0; + _yieldPeriodMsec = DEFAULT_YIELD_PERIOD_SECONDS * 1000; + _penalizationPeriodMsec = DEFAULT_PENALIZATION_PERIOD_SECONDS * 1000; + _maxConcurrentTasks = 1; + _activeTasks = 0; + _yieldExpiration = 0; + _incomingConnectionsIter = this->_incomingConnections.begin(); + _logger = Logger::getLogger(); + + _logger->log_info("Processor %s created UUID %s", _name.c_str(), _uuidStr.c_str()); +} + +Processor::~Processor() +{ + +} + +bool Processor::isRunning() +{ + return (_state == RUNNING && _activeTasks > 0); +} + +bool Processor::setSupportedProperties(std::set<Property> properties) +{ + if (isRunning()) + { + _logger->log_info("Can not set processor property while the process %s is running", + _name.c_str()); + return false; + } + + std::lock_guard<std::mutex> lock(_mtx); + + _properties.clear(); + for (std::set<Property>::iterator it = properties.begin(); it != properties.end(); ++it) + { + Property item(*it); + _properties[item.getName()] = item; + _logger->log_info("Processor %s supported property name %s", _name.c_str(), item.getName().c_str()); + } + + return true; +} + +bool Processor::setSupportedRelationships(std::set<Relationship> relationships) +{ + if (isRunning()) + { + _logger->log_info("Can not set processor supported relationship while the process %s is running", + _name.c_str()); + return false; + } + + std::lock_guard<std::mutex> lock(_mtx); + + _relationships.clear(); + for (std::set<Relationship>::iterator it = relationships.begin(); it != relationships.end(); ++it) + { + Relationship item(*it); + _relationships[item.getName()] = item; + _logger->log_info("Processor %s supported relationship name %s", _name.c_str(), item.getName().c_str()); + } + + return true; +} + +bool Processor::setAutoTerminatedRelationships(std::set<Relationship> relationships) +{ + if (isRunning()) + { + _logger->log_info("Can not set processor auto terminated relationship while the process %s is running", + _name.c_str()); + return false; + } + + std::lock_guard<std::mutex> lock(_mtx); + + _autoTerminatedRelationships.clear(); + for (std::set<Relationship>::iterator it = relationships.begin(); it != relationships.end(); ++it) + { + Relationship item(*it); + _autoTerminatedRelationships[item.getName()] = item; + _logger->log_info("Processor %s auto terminated relationship name %s", _name.c_str(), item.getName().c_str()); + } + + return true; +} + +bool Processor::isAutoTerminated(Relationship relationship) +{ + bool isRun = isRunning(); + + if (!isRun) + _mtx.lock(); + + std::map<std::string, Relationship>::iterator it = _autoTerminatedRelationships.find(relationship.getName()); + if (it != _autoTerminatedRelationships.end()) + { + if (!isRun) + _mtx.unlock(); + return true; + } + else + { + if (!isRun) + _mtx.unlock(); + return false; + } +} + +bool Processor::isSupportedRelationship(Relationship relationship) +{ + bool isRun = isRunning(); + + if (!isRun) + _mtx.lock(); + + std::map<std::string, Relationship>::iterator it = _relationships.find(relationship.getName()); + if (it != _relationships.end()) + { + if (!isRun) + _mtx.unlock(); + return true; + } + else + { + if (!isRun) + _mtx.unlock(); + return false; + } +} + +bool Processor::getProperty(std::string name, std::string &value) +{ + bool isRun = isRunning(); + + if (!isRun) + // Because set property only allowed in non running state, we need to obtain lock avoid rack condition + _mtx.lock(); + + std::map<std::string, Property>::iterator it = _properties.find(name); + if (it != _properties.end()) + { + Property item = it->second; + value = item.getValue(); + if (!isRun) + _mtx.unlock(); + return true; + } + else + { + if (!isRun) + _mtx.unlock(); + return false; + } +} + +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(), value.c_str()); + return true; + } + else + { + return false; + } +} + +std::set<Connection *> Processor::getOutGoingConnections(std::string relationship) +{ + std::set<Connection *> empty; + + std::map<std::string, std::set<Connection *>>::iterator it = _outGoingConnections.find(relationship); + if (it != _outGoingConnections.end()) + { + return _outGoingConnections[relationship]; + } + else + { + return empty; + } +} + +bool Processor::addConnection(Connection *connection) +{ + bool ret = false; + + if (isRunning()) + { + _logger->log_info("Can not add connection while the process %s is running", + _name.c_str()); + return false; + } + + std::lock_guard<std::mutex> lock(_mtx); + + uuid_t srcUUID; + uuid_t destUUID; + + connection->getSourceProcessorUUID(srcUUID); + connection->getDestinationProcessorUUID(destUUID); + + if (uuid_compare(_uuid, destUUID) == 0) + { + // Connection is destination to the current processor + if (_incomingConnections.find(connection) == _incomingConnections.end()) + { + _incomingConnections.insert(connection); + connection->setDestinationProcessor(this); + _logger->log_info("Add connection %s into Processor %s incoming connection", + connection->getName().c_str(), _name.c_str()); + _incomingConnectionsIter = this->_incomingConnections.begin(); + ret = true; + } + } + + if (uuid_compare(_uuid, srcUUID) == 0) + { + std::string relationship = connection->getRelationship().getName(); + // Connection is source from the current processor + std::map<std::string, std::set<Connection *>>::iterator it = + _outGoingConnections.find(relationship); + if (it != _outGoingConnections.end()) + { + // We already has connection for this relationship + std::set<Connection *> existedConnection = it->second; + if (existedConnection.find(connection) == existedConnection.end()) + { + // We do not have the same connection for this relationship yet + existedConnection.insert(connection); + connection->setSourceProcessor(this); + _outGoingConnections[relationship] = existedConnection; + _logger->log_info("Add connection %s into Processor %s outgoing connection for relationship %s", + connection->getName().c_str(), _name.c_str(), relationship.c_str()); + ret = true; + } + } + else + { + // We do not have any outgoing connection for this relationship yet + std::set<Connection *> newConnection; + newConnection.insert(connection); + connection->setSourceProcessor(this); + _outGoingConnections[relationship] = newConnection; + _logger->log_info("Add connection %s into Processor %s outgoing connection for relationship %s", + connection->getName().c_str(), _name.c_str(), relationship.c_str()); + ret = true; + } + } + + return ret; +} + +void Processor::removeConnection(Connection *connection) +{ + if (isRunning()) + { + _logger->log_info("Can not remove connection while the process %s is running", + _name.c_str()); + return; + } + + std::lock_guard<std::mutex> lock(_mtx); + + uuid_t srcUUID; + uuid_t destUUID; + + connection->getSourceProcessorUUID(srcUUID); + connection->getDestinationProcessorUUID(destUUID); + + if (uuid_compare(_uuid, destUUID) == 0) + { + // Connection is destination to the current processor + if (_incomingConnections.find(connection) != _incomingConnections.end()) + { + _incomingConnections.erase(connection); + connection->setDestinationProcessor(NULL); + _logger->log_info("Remove connection %s into Processor %s incoming connection", + connection->getName().c_str(), _name.c_str()); + _incomingConnectionsIter = this->_incomingConnections.begin(); + } + } + + if (uuid_compare(_uuid, srcUUID) == 0) + { + std::string relationship = connection->getRelationship().getName(); + // Connection is source from the current processor + std::map<std::string, std::set<Connection *>>::iterator it = + _outGoingConnections.find(relationship); + if (it == _outGoingConnections.end()) + { + return; + } + else + { + if (_outGoingConnections[relationship].find(connection) != _outGoingConnections[relationship].end()) + { + _outGoingConnections[relationship].erase(connection); + connection->setSourceProcessor(NULL); + _logger->log_info("Remove connection %s into Processor %s outgoing connection for relationship %s", + connection->getName().c_str(), _name.c_str(), relationship.c_str()); + } + } + } +} + +Connection *Processor::getNextIncomingConnection() +{ + std::lock_guard<std::mutex> lock(_mtx); + + if (_incomingConnections.size() == 0) + return NULL; + + if (_incomingConnectionsIter == _incomingConnections.end()) + _incomingConnectionsIter = _incomingConnections.begin(); + + Connection *ret = *_incomingConnectionsIter; + _incomingConnectionsIter++; + + if (_incomingConnectionsIter == _incomingConnections.end()) + _incomingConnectionsIter = _incomingConnections.begin(); + + return ret; +} + +bool Processor::flowFilesQueued() +{ + std::lock_guard<std::mutex> lock(_mtx); + + if (_incomingConnections.size() == 0) + return false; + + for (std::set<Connection *>::iterator it = _incomingConnections.begin(); it != _incomingConnections.end(); ++it) + { + Connection *connection = *it; + if (connection->getQueueSize() > 0) + return true; + } + + return false; +} + +bool Processor::flowFilesOutGoingFull() +{ + std::lock_guard<std::mutex> lock(_mtx); + + std::map<std::string, std::set<Connection *>>::iterator it; + + for (it = _outGoingConnections.begin(); it != _outGoingConnections.end(); ++it) + { + // We already has connection for this relationship + std::set<Connection *> existedConnection = it->second; + for (std::set<Connection *>::iterator itConnection = existedConnection.begin(); itConnection != existedConnection.end(); ++itConnection) + { + Connection *connection = *itConnection; + if (connection->isFull()) + return true; + } + } + + return false; +} + +void Processor::onTrigger() +{ + ProcessContext *context = new ProcessContext(this); + ProcessSession *session = new ProcessSession(context); + try { + // Call the child onTrigger function + this->onTrigger(context, session); + session->commit(); + delete session; + delete context; + } + catch (std::exception &exception) + { + _logger->log_debug("Caught Exception %s", exception.what()); + session->rollback(); + delete session; + delete context; + throw; + } + catch (...) + { + _logger->log_debug("Caught Exception Processor::onTrigger"); + session->rollback(); + delete session; + delete context; + throw; + } +} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/b02af540/libminifi/src/RealTimeDataCollector.cpp ---------------------------------------------------------------------- diff --git a/libminifi/src/RealTimeDataCollector.cpp b/libminifi/src/RealTimeDataCollector.cpp new file mode 100644 index 0000000..c7118ff --- /dev/null +++ b/libminifi/src/RealTimeDataCollector.cpp @@ -0,0 +1,482 @@ +/** + * @file RealTimeDataCollector.cpp + * RealTimeDataCollector class implementation + * + * 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. + */ +#include <vector> +#include <queue> +#include <map> +#include <set> +#include <sys/time.h> +#include <time.h> +#include <chrono> +#include <thread> +#include <random> +#include <netinet/tcp.h> + +#include "RealTimeDataCollector.h" +#include "ProcessContext.h" +#include "ProcessSession.h" + +const std::string RealTimeDataCollector::ProcessorName("RealTimeDataCollector"); +Property RealTimeDataCollector::FILENAME("File Name", "File Name for the real time processor to process", "data.osp"); +Property RealTimeDataCollector::REALTIMESERVERNAME("Real Time Server Name", "Real Time Server Name", "localhost"); +Property RealTimeDataCollector::REALTIMESERVERPORT("Real Time Server Port", "Real Time Server Port", "10000"); +Property RealTimeDataCollector::BATCHSERVERNAME("Batch Server Name", "Batch Server Name", "localhost"); +Property RealTimeDataCollector::BATCHSERVERPORT("Batch Server Port", "Batch Server Port", "10001"); +Property RealTimeDataCollector::ITERATION("Iteration", + "If true, sample osp file will be iterated", "true"); +Property RealTimeDataCollector::REALTIMEMSGID("Real Time Message ID", "Real Time Message ID", "41"); +Property RealTimeDataCollector::BATCHMSGID("Batch Message ID", "Batch Message ID", "172, 30, 48"); +Property RealTimeDataCollector::REALTIMEINTERVAL("Real Time Interval", "Real Time Data Collection Interval in msec", "10 ms"); +Property RealTimeDataCollector::BATCHINTERVAL("Batch Time Interval", "Batch Processing Interval in msec", "100 ms"); +Property RealTimeDataCollector::BATCHMAXBUFFERSIZE("Batch Max Buffer Size", "Batch Buffer Maximum size in bytes", "262144"); +Relationship RealTimeDataCollector::Success("success", "success operational on the flow record"); + +void RealTimeDataCollector::initialize() +{ + //! Set the supported properties + std::set<Property> properties; + properties.insert(FILENAME); + properties.insert(REALTIMESERVERNAME); + properties.insert(REALTIMESERVERPORT); + properties.insert(BATCHSERVERNAME); + properties.insert(BATCHSERVERPORT); + properties.insert(ITERATION); + properties.insert(REALTIMEMSGID); + properties.insert(BATCHMSGID); + properties.insert(REALTIMEINTERVAL); + properties.insert(BATCHINTERVAL); + properties.insert(BATCHMAXBUFFERSIZE); + + setSupportedProperties(properties); + //! Set the supported relationships + std::set<Relationship> relationships; + relationships.insert(Success); + setSupportedRelationships(relationships); + +} + +int RealTimeDataCollector::connectServer(const char *host, uint16_t port) +{ + in_addr_t addr; + int sock = 0; + struct hostent *h; +#ifdef __MACH__ + h = gethostbyname(host); +#else + char buf[1024]; + struct hostent he; + int hh_errno; + gethostbyname_r(host, &he, buf, sizeof(buf), &h, &hh_errno); +#endif + memcpy((char *) &addr, h->h_addr_list[0], h->h_length); + sock = socket(AF_INET, SOCK_STREAM, 0); + if (sock < 0) + { + _logger->log_error("Could not create socket to hostName %s", host); + return 0; + } + +#ifndef __MACH__ + int opt = 1; + bool nagle_off = true; + + if (nagle_off) + { + if (setsockopt(sock, SOL_TCP, TCP_NODELAY, (void *)&opt, sizeof(opt)) < 0) + { + _logger->log_error("setsockopt() TCP_NODELAY failed"); + close(sock); + return 0; + } + if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, + (char *)&opt, sizeof(opt)) < 0) + { + _logger->log_error("setsockopt() SO_REUSEADDR failed"); + close(sock); + return 0; + } + } + + int sndsize = 256*1024; + if (setsockopt(sock, SOL_SOCKET, SO_SNDBUF, (char *)&sndsize, (int)sizeof(sndsize)) < 0) + { + _logger->log_error("setsockopt() SO_SNDBUF failed"); + close(sock); + return 0; + } +#endif + + struct sockaddr_in sa; + socklen_t socklen; + int status; + + //TODO bind socket to the interface + memset(&sa, 0, sizeof(sa)); + sa.sin_family = AF_INET; + sa.sin_addr.s_addr = htonl(INADDR_ANY); + sa.sin_port = htons(0); + socklen = sizeof(sa); + if (bind(sock, (struct sockaddr *)&sa, socklen) < 0) + { + _logger->log_error("socket bind failed"); + close(sock); + return 0; + } + + memset(&sa, 0, sizeof(sa)); + sa.sin_family = AF_INET; + sa.sin_addr.s_addr = addr; + sa.sin_port = htons(port); + socklen = sizeof(sa); + + status = connect(sock, (struct sockaddr *)&sa, socklen); + + if (status < 0) + { + _logger->log_error("socket connect failed to %s %d", host, port); + close(sock); + return 0; + } + + _logger->log_info("socket %d connect to server %s port %d success", sock, host, port); + + return sock; +} + +int RealTimeDataCollector::sendData(int socket, const char *buf, int buflen) +{ + int ret = 0, bytes = 0; + + while (bytes < buflen) + { + ret = send(socket, buf+bytes, buflen-bytes, 0); + //check for errors + if (ret == -1) + { + return ret; + } + bytes+=ret; + } + + if (ret) + _logger->log_debug("Send data size %d over socket %d", buflen, socket); + + return ret; +} + +void RealTimeDataCollector::onTriggerRealTime(ProcessContext *context, ProcessSession *session) +{ + if (_realTimeAccumulated >= this->_realTimeInterval) + { + std::string value; + if (this->getProperty(REALTIMEMSGID.getName(), value)) + { + this->_realTimeMsgID.clear(); + this->_logger->log_info("Real Time Msg IDs %s", value.c_str()); + std::stringstream lineStream(value); + std::string cell; + + while(std::getline(lineStream, cell, ',')) + { + this->_realTimeMsgID.push_back(cell); + // this->_logger->log_debug("Real Time Msg ID %s", cell.c_str()); + } + } + if (this->getProperty(BATCHMSGID.getName(), value)) + { + this->_batchMsgID.clear(); + this->_logger->log_info("Batch Msg IDs %s", value.c_str()); + std::stringstream lineStream(value); + std::string cell; + + while(std::getline(lineStream, cell, ',')) + { + cell = Property::trim(cell); + this->_batchMsgID.push_back(cell); + // this->_logger->log_debug("Batch Msg ID %s", cell.c_str()); + } + } + // _logger->log_info("onTriggerRealTime"); + // Open the file + if (!this->_fileStream.is_open()) + { + _fileStream.open(this->_fileName.c_str(), std::ifstream::in); + if (this->_fileStream.is_open()) + _logger->log_debug("open %s", _fileName.c_str()); + } + if (!_fileStream.good()) + { + _logger->log_error("load data file failed %s", _fileName.c_str()); + return; + } + if (this->_fileStream.is_open()) + { + std::string line; + + while (std::getline(_fileStream, line)) + { + line += "\n"; + std::stringstream lineStream(line); + std::string cell; + if (std::getline(lineStream, cell, ',')) + { + cell = Property::trim(cell); + // Check whether it match to the batch traffic + for (std::vector<std::string>::iterator it = _batchMsgID.begin(); it != _batchMsgID.end(); ++it) + { + if (cell == *it) + { + // push the batch data to the queue + std::lock_guard<std::mutex> lock(_mtx); + while ((_queuedDataSize + line.size()) > _batchMaxBufferSize) + { + std::string item = _queue.front(); + _queuedDataSize -= item.size(); + _logger->log_debug("Pop item size %d from batch queue, queue buffer size %d", item.size(), _queuedDataSize); + _queue.pop(); + } + _queue.push(line); + _queuedDataSize += line.size(); + _logger->log_debug("Push batch msg ID %s into batch queue, queue buffer size %d", cell.c_str(), _queuedDataSize); + } + } + bool findRealTime = false; + // Check whether it match to the real time traffic + for (std::vector<std::string>::iterator it = _realTimeMsgID.begin(); it != _realTimeMsgID.end(); ++it) + { + if (cell == *it) + { + int status = 0; + if (this->_realTimeSocket <= 0) + { + // Connect the LTE socket + uint16_t port = _realTimeServerPort; + this->_realTimeSocket = connectServer(_realTimeServerName.c_str(), port); + } + if (this->_realTimeSocket) + { + // try to send the data + status = sendData(_realTimeSocket, line.data(), line.size()); + if (status < 0) + { + close(_realTimeSocket); + _realTimeSocket = 0; + } + } + if (this->_realTimeSocket <= 0 || status < 0) + { + // push the batch data to the queue + std::lock_guard<std::mutex> lock(_mtx); + while ((_queuedDataSize + line.size()) > _batchMaxBufferSize) + { + std::string item = _queue.front(); + _queuedDataSize -= item.size(); + _logger->log_debug("Pop item size %d from batch queue, queue buffer size %d", item.size(), _queuedDataSize); + _queue.pop(); + } + _queue.push(line); + _queuedDataSize += line.size(); + _logger->log_debug("Push real time msg ID %s into batch queue, queue buffer size %d", cell.c_str(), _queuedDataSize); + } + // find real time + findRealTime = true; + } // cell + } // for real time pattern + if (findRealTime) + // we break the while once we find the first real time + break; + } // if get line + } // while + if (_fileStream.eof()) + { + _fileStream.close(); + } + } // if open + _realTimeAccumulated = 0; + } + _realTimeAccumulated += context->getProcessor()->getSchedulingPeriodNano(); +} + +void RealTimeDataCollector::onTriggerBatch(ProcessContext *context, ProcessSession *session) +{ + if (_batchAcccumulated >= this->_batchInterval) + { + // _logger->log_info("onTriggerBatch"); + // dequeue the batch and send over WIFI + int status = 0; + if (this->_batchSocket <= 0) + { + // Connect the WIFI socket + uint16_t port = _batchServerPort; + this->_batchSocket = connectServer(_batchServerName.c_str(), port); + } + if (this->_batchSocket) + { + std::lock_guard<std::mutex> lock(_mtx); + + while (!_queue.empty()) + { + std::string line = _queue.front(); + status = sendData(_batchSocket, line.data(), line.size()); + _queue.pop(); + _queuedDataSize -= line.size(); + if (status < 0) + { + close(_batchSocket); + _batchSocket = 0; + break; + } + } + } + _batchAcccumulated = 0; + } + _batchAcccumulated += context->getProcessor()->getSchedulingPeriodNano(); +} + +void RealTimeDataCollector::onTrigger(ProcessContext *context, ProcessSession *session) +{ + std::thread::id id = std::this_thread::get_id(); + + if (id == _realTimeThreadId) + return onTriggerRealTime(context, session); + else if (id == _batchThreadId) + return onTriggerBatch(context, session); + else + { + std::lock_guard<std::mutex> lock(_mtx); + if (!this->_firstInvoking) + { + this->_fileName = "data.osp"; + std::string value; + if (this->getProperty(FILENAME.getName(), value)) + { + this->_fileName = value; + this->_logger->log_info("Data Collector File Name %s", _fileName.c_str()); + } + this->_realTimeServerName = "localhost"; + if (this->getProperty(REALTIMESERVERNAME.getName(), value)) + { + this->_realTimeServerName = value; + this->_logger->log_info("Real Time Server Name %s", this->_realTimeServerName.c_str()); + } + this->_realTimeServerPort = 10000; + if (this->getProperty(REALTIMESERVERPORT.getName(), value)) + { + Property::StringToInt(value, _realTimeServerPort); + this->_logger->log_info("Real Time Server Port %d", _realTimeServerPort); + } + if (this->getProperty(BATCHSERVERNAME.getName(), value)) + { + this->_batchServerName = value; + this->_logger->log_info("Batch Server Name %s", this->_batchServerName.c_str()); + } + this->_batchServerPort = 10001; + if (this->getProperty(BATCHSERVERPORT.getName(), value)) + { + Property::StringToInt(value, _batchServerPort); + this->_logger->log_info("Batch Server Port %d", _batchServerPort); + } + if (this->getProperty(ITERATION.getName(), value)) + { + Property::StringToBool(value, this->_iteration); + _logger->log_info("Iteration %d", _iteration); + } + this->_realTimeInterval = 10000000; //10 msec + if (this->getProperty(REALTIMEINTERVAL.getName(), value)) + { + TimeUnit unit; + if (Property::StringToTime(value, _realTimeInterval, unit) && + Property::ConvertTimeUnitToNS(_realTimeInterval, unit, _realTimeInterval)) + { + _logger->log_info("Real Time Interval: [%d] ns", _realTimeInterval); + } + } + this->_batchInterval = 100000000; //100 msec + if (this->getProperty(BATCHINTERVAL.getName(), value)) + { + TimeUnit unit; + if (Property::StringToTime(value, _batchInterval, unit) && + Property::ConvertTimeUnitToNS(_batchInterval, unit, _batchInterval)) + { + _logger->log_info("Batch Time Interval: [%d] ns", _batchInterval); + } + } + this->_batchMaxBufferSize = 256*1024; + if (this->getProperty(BATCHMAXBUFFERSIZE.getName(), value)) + { + Property::StringToInt(value, _batchMaxBufferSize); + this->_logger->log_info("Batch Max Buffer Size %d", _batchMaxBufferSize); + } + if (this->getProperty(REALTIMEMSGID.getName(), value)) + { + this->_logger->log_info("Real Time Msg IDs %s", value.c_str()); + std::stringstream lineStream(value); + std::string cell; + + while(std::getline(lineStream, cell, ',')) + { + this->_realTimeMsgID.push_back(cell); + this->_logger->log_info("Real Time Msg ID %s", cell.c_str()); + } + } + if (this->getProperty(BATCHMSGID.getName(), value)) + { + this->_logger->log_info("Batch Msg IDs %s", value.c_str()); + std::stringstream lineStream(value); + std::string cell; + + while(std::getline(lineStream, cell, ',')) + { + cell = Property::trim(cell); + this->_batchMsgID.push_back(cell); + this->_logger->log_info("Batch Msg ID %s", cell.c_str()); + } + } + // Connect the LTE socket + uint16_t port = _realTimeServerPort; + + this->_realTimeSocket = connectServer(_realTimeServerName.c_str(), port); + + // Connect the WIFI socket + port = _batchServerPort; + + this->_batchSocket = connectServer(_batchServerName.c_str(), port); + + // Open the file + _fileStream.open(this->_fileName.c_str(), std::ifstream::in); + if (!_fileStream.good()) + { + _logger->log_error("load data file failed %s", _fileName.c_str()); + return; + } + else + { + _logger->log_debug("open %s", _fileName.c_str()); + } + _realTimeThreadId = id; + this->_firstInvoking = true; + } + else + { + if (id != _realTimeThreadId) + _batchThreadId = id; + this->_firstInvoking = false; + } + } +} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/b02af540/libminifi/src/RemoteProcessorGroupPort.cpp ---------------------------------------------------------------------- diff --git a/libminifi/src/RemoteProcessorGroupPort.cpp b/libminifi/src/RemoteProcessorGroupPort.cpp new file mode 100644 index 0000000..9d849ae --- /dev/null +++ b/libminifi/src/RemoteProcessorGroupPort.cpp @@ -0,0 +1,100 @@ +/** + * @file RemoteProcessorGroupPort.cpp + * RemoteProcessorGroupPort class implementation + * + * 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. + */ +#include <vector> +#include <queue> +#include <map> +#include <set> +#include <sys/time.h> +#include <time.h> +#include <sstream> +#include <string.h> +#include <iostream> + +#include "TimeUtil.h" +#include "RemoteProcessorGroupPort.h" +#include "ProcessContext.h" +#include "ProcessSession.h" + +const std::string RemoteProcessorGroupPort::ProcessorName("RemoteProcessorGroupPort"); +Property RemoteProcessorGroupPort::hostName("Host Name", "Remote Host Name.", "localhost"); +Property RemoteProcessorGroupPort::port("Port", "Remote Port", "9999"); +Relationship RemoteProcessorGroupPort::relation; + +void RemoteProcessorGroupPort::initialize() +{ + //! Set the supported properties + std::set<Property> properties; + properties.insert(hostName); + properties.insert(port); + setSupportedProperties(properties); + //! Set the supported relationships + std::set<Relationship> relationships; + relationships.insert(relation); + setSupportedRelationships(relationships); +} + +void RemoteProcessorGroupPort::onTrigger(ProcessContext *context, ProcessSession *session) +{ + std::string value; + + if (!_transmitting) + return; + + std::string host = _peer->getHostName(); + uint16_t sport = _peer->getPort(); + int64_t lvalue; + bool needReset = false; + + if (context->getProperty(hostName.getName(), value)) + { + host = value; + } + if (context->getProperty(port.getName(), value) && Property::StringToInt(value, lvalue)) + { + sport = (uint16_t) lvalue; + } + if (host != _peer->getHostName()) + { + _peer->setHostName(host); + needReset= true; + } + if (sport != _peer->getPort()) + { + _peer->setPort(sport); + needReset = true; + } + if (needReset) + _protocol->tearDown(); + + if (!_protocol->bootstrap()) + { + // bootstrap the client protocol if needeed + context->yield(); + _logger->log_error("Site2Site bootstrap failed yield period %d peer timeout %d", context->getProcessor()->getYieldPeriodMsec(), _protocol->getPeer()->getTimeOut()); + return; + } + + if (_direction == RECEIVE) + _protocol->receiveFlowFiles(context, session); + else + _protocol->transferFlowFiles(context, session); + + return; +} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/b02af540/libminifi/src/ResourceClaim.cpp ---------------------------------------------------------------------- diff --git a/libminifi/src/ResourceClaim.cpp b/libminifi/src/ResourceClaim.cpp new file mode 100644 index 0000000..3c22ac9 --- /dev/null +++ b/libminifi/src/ResourceClaim.cpp @@ -0,0 +1,45 @@ +/** + * @file ResourceClaim.cpp + * ResourceClaim class implementation + * + * 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. + */ +#include <vector> +#include <queue> +#include <map> + +#include "ResourceClaim.h" + +std::atomic<uint64_t> ResourceClaim::_localResourceClaimNumber(0); + +ResourceClaim::ResourceClaim(const std::string contentDirectory) +: _id(_localResourceClaimNumber.load()), + _flowFileRecordOwnedCount(0) +{ + char uuidStr[37]; + + // Generate the global UUID for the resource claim + uuid_generate(_uuid); + // Increase the local ID for the resource claim + ++_localResourceClaimNumber; + uuid_unparse(_uuid, uuidStr); + // Create the full content path for the content + _contentFullPath = contentDirectory + "/" + uuidStr; + + _configure = Configure::getConfigure(); + _logger = Logger::getLogger(); + _logger->log_debug("Resource Claim created %s", _contentFullPath.c_str()); +} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/b02af540/libminifi/src/SchedulingAgent.cpp ---------------------------------------------------------------------- diff --git a/libminifi/src/SchedulingAgent.cpp b/libminifi/src/SchedulingAgent.cpp new file mode 100644 index 0000000..211c328 --- /dev/null +++ b/libminifi/src/SchedulingAgent.cpp @@ -0,0 +1,86 @@ +/** + * @file SchedulingAgent.cpp + * SchedulingAgent class implementation + * + * 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. + */ +#include <chrono> +#include <thread> +#include <iostream> +#include "Exception.h" +#include "SchedulingAgent.h" + +bool SchedulingAgent::hasWorkToDo(Processor *processor) +{ + // Whether it has work to do + if (processor->getTriggerWhenEmpty() || !processor->hasIncomingConnections() || + processor->flowFilesQueued()) + return true; + else + return false; +} + +bool SchedulingAgent::hasTooMuchOutGoing(Processor *processor) +{ + return processor->flowFilesOutGoingFull(); +} + +bool SchedulingAgent::onTrigger(Processor *processor) +{ + if (processor->isYield()) + return false; + + // No need to yield, reset yield expiration to 0 + processor->clearYield(); + + if (!hasWorkToDo(processor)) + // No work to do, yield + return true; + + if(hasTooMuchOutGoing(processor)) + // need to apply backpressure + return true; + + //TODO runDuration + + processor->incrementActiveTasks(); + try + { + processor->onTrigger(); + processor->decrementActiveTask(); + } + catch (Exception &exception) + { + // Normal exception + _logger->log_debug("Caught Exception %s", exception.what()); + processor->decrementActiveTask(); + } + catch (std::exception &exception) + { + _logger->log_debug("Caught Exception %s", exception.what()); + processor->yield(_administrativeYieldDuration); + processor->decrementActiveTask(); + } + catch (...) + { + _logger->log_debug("Caught Exception during SchedulingAgent::onTrigger"); + processor->yield(_administrativeYieldDuration); + processor->decrementActiveTask(); + } + + return false; +} +
