http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/b02af540/src/Site2SiteClientProtocol.cpp ---------------------------------------------------------------------- diff --git a/src/Site2SiteClientProtocol.cpp b/src/Site2SiteClientProtocol.cpp deleted file mode 100644 index 88ea78a..0000000 --- a/src/Site2SiteClientProtocol.cpp +++ /dev/null @@ -1,1313 +0,0 @@ -/** - * @file Site2SiteProtocol.cpp - * Site2SiteProtocol 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 <sys/time.h> -#include <stdio.h> -#include <time.h> -#include <chrono> -#include <thread> -#include <random> -#include <netinet/tcp.h> -#include <iostream> -#include "Site2SitePeer.h" -#include "Site2SiteClientProtocol.h" - -bool Site2SiteClientProtocol::establish() -{ - if (_peerState != IDLE) - { - _logger->log_error("Site2Site peer state is not idle while try to establish"); - return false; - } - - bool ret = _peer->Open(); - - if (!ret) - { - _logger->log_error("Site2Site peer socket open failed"); - return false; - } - - // Negotiate the version - ret = initiateResourceNegotiation(); - - if (!ret) - { - _logger->log_error("Site2Site Protocol Version Negotiation failed"); - /* - _peer->yield(); - tearDown(); */ - return false; - } - - _logger->log_info("Site2Site socket established"); - _peerState = ESTABLISHED; - - return true; -} - -bool Site2SiteClientProtocol::initiateResourceNegotiation() -{ - // Negotiate the version - if (_peerState != IDLE) - { - _logger->log_error("Site2Site peer state is not idle while initiateResourceNegotiation"); - return false; - } - - _logger->log_info("Negotiate protocol version with destination port %s current version %d", _portIdStr.c_str(), _currentVersion); - - int ret = _peer->writeUTF(this->getResourceName()); - - if (ret <= 0) - { - // tearDown(); - return false; - } - - ret = _peer->write(_currentVersion); - - if (ret <= 0) - { - // tearDown(); - return false; - } - - uint8_t statusCode; - ret = _peer->read(statusCode); - - if (ret <= 0) - { - // tearDown(); - return false; - } - - switch (statusCode) - { - case RESOURCE_OK: - _logger->log_info("Site2Site Protocol Negotiate protocol version OK"); - return true; - case DIFFERENT_RESOURCE_VERSION: - uint32_t serverVersion; - ret = _peer->read(serverVersion); - if (ret <= 0) - { - // tearDown(); - return false; - } - _logger->log_info("Site2Site Server Response asked for a different protocol version %d", serverVersion); - for (unsigned int i = (_currentVersionIndex + 1); i < sizeof(_supportedVersion)/sizeof(uint32_t); i++) - { - if (serverVersion >= _supportedVersion[i]) - { - _currentVersion = _supportedVersion[i]; - _currentVersionIndex = i; - return initiateResourceNegotiation(); - } - } - ret = -1; - // tearDown(); - return false; - case NEGOTIATED_ABORT: - _logger->log_info("Site2Site Negotiate protocol response ABORT"); - ret = -1; - // tearDown(); - return false; - default: - _logger->log_info("Negotiate protocol response unknown code %d", statusCode); - return true; - } - - return true; -} - -bool Site2SiteClientProtocol::initiateCodecResourceNegotiation() -{ - // Negotiate the version - if (_peerState != HANDSHAKED) - { - _logger->log_error("Site2Site peer state is not handshaked while initiateCodecResourceNegotiation"); - return false; - } - - _logger->log_info("Negotiate Codec version with destination port %s current version %d", _portIdStr.c_str(), _currentCodecVersion); - - int ret = _peer->writeUTF(this->getCodecResourceName()); - - if (ret <= 0) - { - // tearDown(); - return false; - } - - ret = _peer->write(_currentCodecVersion); - - if (ret <= 0) - { - // tearDown(); - return false; - } - - uint8_t statusCode; - ret = _peer->read(statusCode); - - if (ret <= 0) - { - // tearDown(); - return false; - } - - switch (statusCode) - { - case RESOURCE_OK: - _logger->log_info("Site2Site Codec Negotiate version OK"); - return true; - case DIFFERENT_RESOURCE_VERSION: - uint32_t serverVersion; - ret = _peer->read(serverVersion); - if (ret <= 0) - { - // tearDown(); - return false; - } - _logger->log_info("Site2Site Server Response asked for a different codec version %d", serverVersion); - for (unsigned int i = (_currentCodecVersionIndex + 1); i < sizeof(_supportedCodecVersion)/sizeof(uint32_t); i++) - { - if (serverVersion >= _supportedCodecVersion[i]) - { - _currentCodecVersion = _supportedCodecVersion[i]; - _currentCodecVersionIndex = i; - return initiateCodecResourceNegotiation(); - } - } - ret = -1; - // tearDown(); - return false; - case NEGOTIATED_ABORT: - _logger->log_info("Site2Site Codec Negotiate response ABORT"); - ret = -1; - // tearDown(); - return false; - default: - _logger->log_info("Negotiate Codec response unknown code %d", statusCode); - return true; - } - - return true; -} - -bool Site2SiteClientProtocol::handShake() -{ - if (_peerState != ESTABLISHED) - { - _logger->log_error("Site2Site peer state is not established while handshake"); - return false; - } - _logger->log_info("Site2Site Protocol Perform hand shake with destination port %s", _portIdStr.c_str()); - uuid_t uuid; - // Generate the global UUID for the com identify - uuid_generate(uuid); - char uuidStr[37]; - uuid_unparse(uuid, uuidStr); - _commsIdentifier = uuidStr; - - int ret = _peer->writeUTF(_commsIdentifier); - - if (ret <= 0) - { - // tearDown(); - return false; - } - - std::map<std::string, std::string> properties; - properties[HandShakePropertyStr[GZIP]] = "false"; - properties[HandShakePropertyStr[PORT_IDENTIFIER]] = _portIdStr; - properties[HandShakePropertyStr[REQUEST_EXPIRATION_MILLIS]] = std::to_string(this->_timeOut); - if (this->_currentVersion >= 5) - { - if (this->_batchCount > 0) - properties[HandShakePropertyStr[BATCH_COUNT]] = std::to_string(this->_batchCount); - if (this->_batchSize > 0) - properties[HandShakePropertyStr[BATCH_SIZE]] = std::to_string(this->_batchSize); - if (this->_batchDuration > 0) - properties[HandShakePropertyStr[BATCH_DURATION]] = std::to_string(this->_batchDuration); - } - - if (_currentVersion >= 3) - { - ret = _peer->writeUTF(_peer->getURL()); - if (ret <= 0) - { - // tearDown(); - return false; - } - } - - uint32_t size = properties.size(); - ret = _peer->write(size); - if (ret <= 0) - { - // tearDown(); - return false; - } - - std::map<std::string, std::string>::iterator it; - for (it = properties.begin(); it!= properties.end(); it++) - { - ret = _peer->writeUTF(it->first); - if (ret <= 0) - { - // tearDown(); - return false; - } - ret = _peer->writeUTF(it->second); - if (ret <= 0) - { - // tearDown(); - return false; - } - _logger->log_info("Site2Site Protocol Send handshake properties %s %s", it->first.c_str(), it->second.c_str()); - } - - RespondCode code; - std::string message; - - ret = this->readRespond(code, message); - - if (ret <= 0) - { - // tearDown(); - return false; - } - - switch (code) - { - case PROPERTIES_OK: - _logger->log_info("Site2Site HandShake Completed"); - _peerState = HANDSHAKED; - return true; - case PORT_NOT_IN_VALID_STATE: - case UNKNOWN_PORT: - case PORTS_DESTINATION_FULL: - _logger->log_error("Site2Site HandShake Failed because destination port is either invalid or full"); - ret = -1; - /* - _peer->yield(); - tearDown(); */ - return false; - default: - _logger->log_info("HandShake Failed because of unknown respond code %d", code); - ret = -1; - /* - _peer->yield(); - tearDown(); */ - return false; - } - - return false; -} - -void Site2SiteClientProtocol::tearDown() -{ - if (_peerState >= ESTABLISHED) - { - _logger->log_info("Site2Site Protocol tearDown"); - // need to write shutdown request - writeRequestType(SHUTDOWN); - } - - std::map<std::string, Transaction *>::iterator it; - for (it = _transactionMap.begin(); it!= _transactionMap.end(); it++) - { - delete it->second; - } - _transactionMap.clear(); - _peer->Close(); - _peerState = IDLE; -} - -int Site2SiteClientProtocol::writeRequestType(RequestType type) -{ - if (type >= MAX_REQUEST_TYPE) - return -1; - - return _peer->writeUTF(RequestTypeStr[type]); -} - -int Site2SiteClientProtocol::readRequestType(RequestType &type) -{ - std::string requestTypeStr; - - int ret = _peer->readUTF(requestTypeStr); - - if (ret <= 0) - return ret; - - for (int i = (int) NEGOTIATE_FLOWFILE_CODEC; i <= (int) SHUTDOWN; i++) - { - if (RequestTypeStr[i] == requestTypeStr) - { - type = (RequestType) i; - return ret; - } - } - - return -1; -} - -int Site2SiteClientProtocol::readRespond(RespondCode &code, std::string &message) -{ - uint8_t firstByte; - - int ret = _peer->read(firstByte); - - if (ret <= 0 || firstByte != CODE_SEQUENCE_VALUE_1) - return -1; - - uint8_t secondByte; - - ret = _peer->read(secondByte); - - if (ret <= 0 || secondByte != CODE_SEQUENCE_VALUE_2) - return -1; - - uint8_t thirdByte; - - ret = _peer->read(thirdByte); - - if (ret <= 0) - return ret; - - code = (RespondCode) thirdByte; - - RespondCodeContext *resCode = this->getRespondCodeContext(code); - - if ( resCode == NULL) - { - // Not a valid respond code - return -1; - } - if (resCode->hasDescription) - { - ret = _peer->readUTF(message); - if (ret <= 0) - return -1; - } - return 3 + message.size(); -} - -int Site2SiteClientProtocol::writeRespond(RespondCode code, std::string message) -{ - RespondCodeContext *resCode = this->getRespondCodeContext(code); - - if (resCode == NULL) - { - // Not a valid respond code - return -1; - } - - uint8_t codeSeq[3]; - codeSeq[0] = CODE_SEQUENCE_VALUE_1; - codeSeq[1] = CODE_SEQUENCE_VALUE_2; - codeSeq[2] = (uint8_t) code; - - int ret = _peer->write(codeSeq, 3); - - if (ret != 3) - return -1; - - if (resCode->hasDescription) - { - ret = _peer->writeUTF(message); - if (ret > 0) - return (3 + ret); - else - return ret; - } - else - return 3; -} - -bool Site2SiteClientProtocol::negotiateCodec() -{ - if (_peerState != HANDSHAKED) - { - _logger->log_error("Site2Site peer state is not handshaked while negotiate codec"); - return false; - } - - _logger->log_info("Site2Site Protocol Negotiate Codec with destination port %s", _portIdStr.c_str()); - - int status = this->writeRequestType(NEGOTIATE_FLOWFILE_CODEC); - - if (status <= 0) - { - // tearDown(); - return false; - } - - // Negotiate the codec version - bool ret = initiateCodecResourceNegotiation(); - - if (!ret) - { - _logger->log_error("Site2Site Codec Version Negotiation failed"); - /* - _peer->yield(); - tearDown(); */ - return false; - } - - _logger->log_info("Site2Site Codec Completed and move to READY state for data transfer"); - _peerState = READY; - - return true; -} - -bool Site2SiteClientProtocol::bootstrap() -{ - if (_peerState == READY) - return true; - - tearDown(); - - if (establish() && handShake() && negotiateCodec()) - { - _logger->log_info("Site2Site Ready For data transaction"); - return true; - } - else - { - _peer->yield(); - tearDown(); - return false; - } -} - -Transaction* Site2SiteClientProtocol::createTransaction(std::string &transactionID, TransferDirection direction) -{ - int ret; - bool dataAvailable; - Transaction *transaction = NULL; - - if (_peerState != READY) - { - bootstrap(); - } - - if (_peerState != READY) - { - return NULL; - } - - if (direction == RECEIVE) - { - ret = writeRequestType(RECEIVE_FLOWFILES); - - if (ret <= 0) - { - // tearDown(); - return NULL; - } - - RespondCode code; - std::string message; - - ret = readRespond(code, message); - - if (ret <= 0) - { - // tearDown(); - return NULL; - } - - switch (code) - { - case MORE_DATA: - dataAvailable = true; - _logger->log_info("Site2Site peer indicates that data is available"); - transaction = new Transaction(direction); - _transactionMap[transaction->getUUIDStr()] = transaction; - transactionID = transaction->getUUIDStr(); - transaction->setDataAvailable(dataAvailable); - _logger->log_info("Site2Site create transaction %s", transaction->getUUIDStr().c_str()); - return transaction; - case NO_MORE_DATA: - dataAvailable = false; - _logger->log_info("Site2Site peer indicates that no data is available"); - transaction = new Transaction(direction); - _transactionMap[transaction->getUUIDStr()] = transaction; - transactionID = transaction->getUUIDStr(); - transaction->setDataAvailable(dataAvailable); - _logger->log_info("Site2Site create transaction %s", transaction->getUUIDStr().c_str()); - return transaction; - default: - _logger->log_info("Site2Site got unexpected response %d when asking for data", code); - // tearDown(); - return NULL; - } - } - else - { - ret = writeRequestType(SEND_FLOWFILES); - - if (ret <= 0) - { - // tearDown(); - return NULL; - } - else - { - transaction = new Transaction(direction); - _transactionMap[transaction->getUUIDStr()] = transaction; - transactionID = transaction->getUUIDStr(); - _logger->log_info("Site2Site create transaction %s", transaction->getUUIDStr().c_str()); - return transaction; - } - } -} - -bool Site2SiteClientProtocol::receive(std::string transactionID, DataPacket *packet, bool &eof) -{ - int ret; - Transaction *transaction = NULL; - - if (_peerState != READY) - { - bootstrap(); - } - - if (_peerState != READY) - { - return false; - } - - std::map<std::string, Transaction *>::iterator it = this->_transactionMap.find(transactionID); - - if (it == _transactionMap.end()) - { - return false; - } - else - { - transaction = it->second; - } - - if (transaction->getState() != TRANSACTION_STARTED && transaction->getState() != DATA_EXCHANGED) - { - _logger->log_info("Site2Site transaction %s is not at started or exchanged state", transactionID.c_str()); - return false; - } - - if (transaction->getDirection() != RECEIVE) - { - _logger->log_info("Site2Site transaction %s direction is wrong", transactionID.c_str()); - return false; - } - - if (!transaction->isDataAvailable()) - { - eof = true; - return true; - } - - if (transaction->_transfers > 0) - { - // if we already has transfer before, check to see whether another one is available - RespondCode code; - std::string message; - - ret = readRespond(code, message); - - if (ret <= 0) - { - return false; - } - if (code == CONTINUE_TRANSACTION) - { - _logger->log_info("Site2Site transaction %s peer indicate continue transaction", transactionID.c_str()); - transaction->_dataAvailable = true; - } - else if (code == FINISH_TRANSACTION) - { - _logger->log_info("Site2Site transaction %s peer indicate finish transaction", transactionID.c_str()); - transaction->_dataAvailable = false; - } - else - { - _logger->log_info("Site2Site transaction %s peer indicate wrong respond code %d", transactionID.c_str(), code); - return false; - } - } - - if (!transaction->isDataAvailable()) - { - eof = true; - return true; - } - - // start to read the packet - uint32_t numAttributes; - ret = _peer->read(numAttributes, &transaction->_crc); - if (ret <= 0 || numAttributes > MAX_NUM_ATTRIBUTES) - { - return false; - } - - // read the attributes - for (unsigned int i = 0; i < numAttributes; i++) - { - std::string key; - std::string value; - ret = _peer->readUTF(key, true, &transaction->_crc); - if (ret <= 0) - { - return false; - } - ret = _peer->readUTF(value, true, &transaction->_crc); - if (ret <= 0) - { - return false; - } - packet->_attributes[key] = value; - _logger->log_info("Site2Site transaction %s receives attribute key %s value %s", transactionID.c_str(), key.c_str(), value.c_str()); - } - - uint64_t len; - ret = _peer->read(len, &transaction->_crc); - if (ret <= 0) - { - return false; - } - - packet->_size = len; - transaction->_transfers++; - transaction->_state = DATA_EXCHANGED; - transaction->_bytes += len; - _logger->log_info("Site2Site transaction %s receives flow record %d, total length %d", transactionID.c_str(), - transaction->_transfers, transaction->_bytes); - - return true; -} - -bool Site2SiteClientProtocol::send(std::string transactionID, DataPacket *packet, FlowFileRecord *flowFile, ProcessSession *session) -{ - int ret; - Transaction *transaction = NULL; - - if (_peerState != READY) - { - bootstrap(); - } - - if (_peerState != READY) - { - return false; - } - - std::map<std::string, Transaction *>::iterator it = this->_transactionMap.find(transactionID); - - if (it == _transactionMap.end()) - { - return false; - } - else - { - transaction = it->second; - } - - if (transaction->getState() != TRANSACTION_STARTED && transaction->getState() != DATA_EXCHANGED) - { - _logger->log_info("Site2Site transaction %s is not at started or exchanged state", transactionID.c_str()); - return false; - } - - if (transaction->getDirection() != SEND) - { - _logger->log_info("Site2Site transaction %s direction is wrong", transactionID.c_str()); - return false; - } - - if (transaction->_transfers > 0) - { - ret = writeRespond(CONTINUE_TRANSACTION, "CONTINUE_TRANSACTION"); - if (ret <= 0) - { - return false; - } - } - - // start to read the packet - uint32_t numAttributes = packet->_attributes.size(); - ret = _peer->write(numAttributes, &transaction->_crc); - if (ret != 4) - { - return false; - } - - std::map<std::string, std::string>::iterator itAttribute; - for (itAttribute = packet->_attributes.begin(); itAttribute!= packet->_attributes.end(); itAttribute++) - { - ret = _peer->writeUTF(itAttribute->first, true, &transaction->_crc); - if (ret <= 0) - { - return false; - } - ret = _peer->writeUTF(itAttribute->second, true, &transaction->_crc); - if (ret <= 0) - { - return false; - } - _logger->log_info("Site2Site transaction %s send attribute key %s value %s", transactionID.c_str(), - itAttribute->first.c_str(), itAttribute->second.c_str()); - } - - uint64_t len = flowFile->getSize() ; - ret = _peer->write(len, &transaction->_crc); - if (ret != 8) - { - return false; - } - - if (flowFile->getSize()) - { - Site2SiteClientProtocol::ReadCallback callback(packet); - session->read(flowFile, &callback); - if (flowFile->getSize() != packet->_size) - { - return false; - } - } - - transaction->_transfers++; - transaction->_state = DATA_EXCHANGED; - transaction->_bytes += len; - _logger->log_info("Site2Site transaction %s send flow record %d, total length %d", transactionID.c_str(), - transaction->_transfers, transaction->_bytes); - - return true; -} - -void Site2SiteClientProtocol::receiveFlowFiles(ProcessContext *context, ProcessSession *session) -{ - uint64_t bytes = 0; - int transfers = 0; - Transaction *transaction = NULL; - - if (_peerState != READY) - { - bootstrap(); - } - - if (_peerState != READY) - { - context->yield(); - tearDown(); - throw Exception(SITE2SITE_EXCEPTION, "Can not establish handshake with peer"); - return; - } - - // Create the transaction - std::string transactionID; - transaction = createTransaction(transactionID, RECEIVE); - - if (transaction == NULL) - { - context->yield(); - tearDown(); - throw Exception(SITE2SITE_EXCEPTION, "Can not create transaction"); - return; - } - - try - { - while (true) - { - std::map<std::string, std::string> empty; - DataPacket packet(this, transaction, empty); - bool eof = false; - - if (!receive(transactionID, &packet, eof)) - { - throw Exception(SITE2SITE_EXCEPTION, "Receive Failed"); - return; - } - if (eof) - { - // transaction done - break; - } - FlowFileRecord *flowFile = session->create(); - if (!flowFile) - { - throw Exception(SITE2SITE_EXCEPTION, "Flow File Creation Failed"); - return; - } - std::map<std::string, std::string>::iterator it; - for (it = packet._attributes.begin(); it!= packet._attributes.end(); it++) - { - flowFile->addAttribute(it->first, it->second); - } - - if (packet._size > 0) - { - Site2SiteClientProtocol::WriteCallback callback(&packet); - session->write(flowFile, &callback); - if (flowFile->getSize() != packet._size) - { - throw Exception(SITE2SITE_EXCEPTION, "Receive Size Not Right"); - return; - } - } - Relationship relation; // undefined relationship - session->transfer(flowFile, relation); - // receive the transfer for the flow record - bytes += packet._size; - transfers++; - } // while true - - if (!confirm(transactionID)) - { - throw Exception(SITE2SITE_EXCEPTION, "Confirm Transaction Failed"); - return; - } - if (!complete(transactionID)) - { - throw Exception(SITE2SITE_EXCEPTION, "Complete Transaction Failed"); - return; - } - _logger->log_info("Site2Site transaction %s successfully receive flow record %d, content bytes %d", - transactionID.c_str(), transfers, bytes); - // we yield the receive if we did not get anything - if (transfers == 0) - context->yield(); - } - catch (std::exception &exception) - { - if (transaction) - deleteTransaction(transactionID); - context->yield(); - tearDown(); - _logger->log_debug("Caught Exception %s", exception.what()); - throw; - } - catch (...) - { - if (transaction) - deleteTransaction(transactionID); - context->yield(); - tearDown(); - _logger->log_debug("Caught Exception during Site2SiteClientProtocol::receiveFlowFiles"); - throw; - } - - deleteTransaction(transactionID); - - return; -} - -bool Site2SiteClientProtocol::confirm(std::string transactionID) -{ - int ret; - Transaction *transaction = NULL; - - if (_peerState != READY) - { - bootstrap(); - } - - if (_peerState != READY) - { - return false; - } - - std::map<std::string, Transaction *>::iterator it = this->_transactionMap.find(transactionID); - - if (it == _transactionMap.end()) - { - return false; - } - else - { - transaction = it->second; - } - - if (transaction->getState() == TRANSACTION_STARTED && !transaction->isDataAvailable() && - transaction->getDirection() == RECEIVE) - { - transaction->_state = TRANSACTION_CONFIRMED; - return true; - } - - if (transaction->getState() != DATA_EXCHANGED) - return false; - - if (transaction->getDirection() == RECEIVE) - { - if (transaction->isDataAvailable()) - return false; - // we received a FINISH_TRANSACTION indicator. Send back a CONFIRM_TRANSACTION message - // to peer so that we can verify that the connection is still open. This is a two-phase commit, - // which helps to prevent the chances of data duplication. Without doing this, we may commit the - // session and then when we send the response back to the peer, the peer may have timed out and may not - // be listening. As a result, it will re-send the data. By doing this two-phase commit, we narrow the - // Critical Section involved in this transaction so that rather than the Critical Section being the - // time window involved in the entire transaction, it is reduced to a simple round-trip conversation. - long crcValue = transaction->getCRC(); - std::string crc = std::to_string(crcValue); - _logger->log_info("Site2Site Send confirm with CRC %d to transaction %s", transaction->getCRC(), - transactionID.c_str()); - ret = writeRespond(CONFIRM_TRANSACTION, crc); - if (ret <= 0) - return false; - RespondCode code; - std::string message; - readRespond(code, message); - if (ret <= 0) - return false; - - if (code == CONFIRM_TRANSACTION) - { - _logger->log_info("Site2Site transaction %s peer confirm transaction", transactionID.c_str()); - transaction->_state = TRANSACTION_CONFIRMED; - return true; - } - else if (code == BAD_CHECKSUM) - { - _logger->log_info("Site2Site transaction %s peer indicate bad checksum", transactionID.c_str()); - /* - transaction->_state = TRANSACTION_CONFIRMED; - return true; */ - return false; - } - else - { - _logger->log_info("Site2Site transaction %s peer unknown respond code %d", - transactionID.c_str(), code); - return false; - } - } - else - { - _logger->log_info("Site2Site Send FINISH TRANSACTION for transaction %s", - transactionID.c_str()); - ret = writeRespond(FINISH_TRANSACTION, "FINISH_TRANSACTION"); - if (ret <= 0) - return false; - RespondCode code; - std::string message; - readRespond(code, message); - if (ret <= 0) - return false; - - // we've sent a FINISH_TRANSACTION. Now we'll wait for the peer to send a 'Confirm Transaction' response - if (code == CONFIRM_TRANSACTION) - { - _logger->log_info("Site2Site transaction %s peer confirm transaction with CRC %s", transactionID.c_str(), message.c_str()); - if (this->_currentVersion > 3) - { - long crcValue = transaction->getCRC(); - std::string crc = std::to_string(crcValue); - if (message == crc) - { - _logger->log_info("Site2Site transaction %s CRC matched", transactionID.c_str()); - ret = writeRespond(CONFIRM_TRANSACTION, "CONFIRM_TRANSACTION"); - if (ret <= 0) - return false; - transaction->_state = TRANSACTION_CONFIRMED; - return true; - } - else - { - _logger->log_info("Site2Site transaction %s CRC not matched %s", transactionID.c_str(), crc.c_str()); - ret = writeRespond(BAD_CHECKSUM, "BAD_CHECKSUM"); - /* - ret = writeRespond(CONFIRM_TRANSACTION, "CONFIRM_TRANSACTION"); - if (ret <= 0) - return false; - transaction->_state = TRANSACTION_CONFIRMED; - return true; */ - return false; - } - } - ret = writeRespond(CONFIRM_TRANSACTION, "CONFIRM_TRANSACTION"); - if (ret <= 0) - return false; - transaction->_state = TRANSACTION_CONFIRMED; - return true; - } - else - { - _logger->log_info("Site2Site transaction %s peer unknown respond code %d", - transactionID.c_str(), code); - return false; - } - return false; - } -} - -void Site2SiteClientProtocol::cancel(std::string transactionID) -{ - Transaction *transaction = NULL; - - if (_peerState != READY) - { - return; - } - - std::map<std::string, Transaction *>::iterator it = this->_transactionMap.find(transactionID); - - if (it == _transactionMap.end()) - { - return; - } - else - { - transaction = it->second; - } - - if (transaction->getState() == TRANSACTION_CANCELED || transaction->getState() == TRANSACTION_COMPLETED - || transaction->getState() == TRANSACTION_ERROR) - { - return; - } - - this->writeRespond(CANCEL_TRANSACTION, "Cancel"); - transaction->_state = TRANSACTION_CANCELED; - - tearDown(); - return; -} - -void Site2SiteClientProtocol::deleteTransaction(std::string transactionID) -{ - Transaction *transaction = NULL; - - std::map<std::string, Transaction *>::iterator it = this->_transactionMap.find(transactionID); - - if (it == _transactionMap.end()) - { - return; - } - else - { - transaction = it->second; - } - - _logger->log_info("Site2Site delete transaction %s", transaction->getUUIDStr().c_str()); - delete transaction; - _transactionMap.erase(transactionID); -} - -void Site2SiteClientProtocol::error(std::string transactionID) -{ - Transaction *transaction = NULL; - - std::map<std::string, Transaction *>::iterator it = this->_transactionMap.find(transactionID); - - if (it == _transactionMap.end()) - { - return; - } - else - { - transaction = it->second; - } - - transaction->_state = TRANSACTION_ERROR; - tearDown(); - return; -} - -//! Complete the transaction -bool Site2SiteClientProtocol::complete(std::string transactionID) -{ - int ret; - Transaction *transaction = NULL; - - if (_peerState != READY) - { - bootstrap(); - } - - if (_peerState != READY) - { - return false; - } - - std::map<std::string, Transaction *>::iterator it = this->_transactionMap.find(transactionID); - - if (it == _transactionMap.end()) - { - return false; - } - else - { - transaction = it->second; - } - - if (transaction->getState() != TRANSACTION_CONFIRMED) - { - return false; - } - - if (transaction->getDirection() == RECEIVE) - { - if (transaction->_transfers == 0) - { - transaction->_state = TRANSACTION_COMPLETED; - return true; - } - else - { - _logger->log_info("Site2Site transaction %s send finished", transactionID.c_str()); - ret = this->writeRespond(TRANSACTION_FINISHED, "Finished"); - if (ret <= 0) - return false; - else - { - transaction->_state = TRANSACTION_COMPLETED; - return true; - } - } - } - else - { - RespondCode code; - std::string message; - int ret; - - ret = readRespond(code, message); - - if (ret <= 0) - return false; - - if (code == TRANSACTION_FINISHED) - { - _logger->log_info("Site2Site transaction %s peer finished transaction", transactionID.c_str()); - transaction->_state = TRANSACTION_COMPLETED; - return true; - } - else - { - _logger->log_info("Site2Site transaction %s peer unknown respond code %d", - transactionID.c_str(), code); - return false; - } - } -} - -void Site2SiteClientProtocol::transferFlowFiles(ProcessContext *context, ProcessSession *session) -{ - FlowFileRecord *flow = session->get(); - Transaction *transaction = NULL; - - if (!flow) - return; - - if (_peerState != READY) - { - bootstrap(); - } - - if (_peerState != READY) - { - context->yield(); - tearDown(); - throw Exception(SITE2SITE_EXCEPTION, "Can not establish handshake with peer"); - return; - } - - // Create the transaction - std::string transactionID; - transaction = createTransaction(transactionID, SEND); - - if (transaction == NULL) - { - context->yield(); - tearDown(); - throw Exception(SITE2SITE_EXCEPTION, "Can not create transaction"); - return; - } - - bool continueTransaction = true; - uint64_t startSendingNanos = getTimeNano(); - - try - { - while (continueTransaction) - { - DataPacket packet(this, transaction, flow->getAttributes()); - - if (!send(transactionID, &packet, flow, session)) - { - throw Exception(SITE2SITE_EXCEPTION, "Send Failed"); - return; - } - _logger->log_info("Site2Site transaction %s send flow record %s", - transactionID.c_str(), flow->getUUIDStr().c_str()); - session->remove(flow); - - uint64_t transferNanos = getTimeNano() - startSendingNanos; - if (transferNanos > _batchSendNanos) - break; - - flow = session->get(); - if (!flow) - { - continueTransaction = false; - } - } // while true - - if (!confirm(transactionID)) - { - throw Exception(SITE2SITE_EXCEPTION, "Confirm Failed"); - return; - } - if (!complete(transactionID)) - { - throw Exception(SITE2SITE_EXCEPTION, "Complete Failed"); - return; - } - _logger->log_info("Site2Site transaction %s successfully send flow record %d, content bytes %d", - transactionID.c_str(), transaction->_transfers, transaction->_bytes); - } - catch (std::exception &exception) - { - if (transaction) - deleteTransaction(transactionID); - context->yield(); - tearDown(); - _logger->log_debug("Caught Exception %s", exception.what()); - throw; - } - catch (...) - { - if (transaction) - deleteTransaction(transactionID); - context->yield(); - tearDown(); - _logger->log_debug("Caught Exception during Site2SiteClientProtocol::transferFlowFiles"); - throw; - } - - deleteTransaction(transactionID); - - return; -}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/b02af540/src/Site2SitePeer.cpp ---------------------------------------------------------------------- diff --git a/src/Site2SitePeer.cpp b/src/Site2SitePeer.cpp deleted file mode 100644 index 48e19d0..0000000 --- a/src/Site2SitePeer.cpp +++ /dev/null @@ -1,435 +0,0 @@ -/** - * @file Site2SitePeer.cpp - * Site2SitePeer 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 <sys/time.h> -#include <stdio.h> -#include <time.h> -#include <chrono> -#include <thread> -#include <random> -#include <netinet/tcp.h> -#include <iostream> -#include "Site2SitePeer.h" - -//! CRC tables -std::atomic<bool> CRC32::tableInit(false); -unsigned int CRC32::table[256]; - -bool Site2SitePeer::Open() -{ - in_addr_t addr; - int sock = 0; - struct hostent *h; - const char *host; - uint16_t port; - - host = this->_host.c_str(); - port = this->_port; - - if (strlen(host) == 0) - return false; - -#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); - this->yield(); - return false; - } - -#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); - this->yield(); - return false; - } - if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, - (char *)&opt, sizeof(opt)) < 0) - { - _logger->log_error("setsockopt() SO_REUSEADDR failed"); - close(sock); - this->yield(); - return false; - } - } - - 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); - this->yield(); - return false; - } - int rcvsize = 256*1024; - if (setsockopt(sock, SOL_SOCKET, SO_RCVBUF, (char *)&rcvsize, (int)sizeof(rcvsize)) < 0) - { - _logger->log_error("setsockopt() SO_RCVBUF failed"); - close(sock); - this->yield(); - return false; - } -#endif - - struct sockaddr_in sa; - socklen_t socklen; - int status; - - 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); - this->yield(); - return false; - } - - 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); - this->yield(); - return false; - } - - _logger->log_info("Site2Site Peer socket %d connect to server %s port %d success", sock, host, port); - - _socket = sock; - - status = sendData((uint8_t *) MAGIC_BYTES, sizeof(MAGIC_BYTES)); - - if (status <= 0) - { - Close(); - return false; - } - - return true; -} - -void Site2SitePeer::Close() -{ - if (_socket) - { - _logger->log_info("Site2Site Peer socket %d close", _socket); - close(_socket); - _socket = 0; - } -} - -int Site2SitePeer::sendData(uint8_t *buf, int buflen, CRC32 *crc) -{ - int ret = 0, bytes = 0; - - if (_socket <= 0) - { - // this->yield(); - return -1; - } - - while (bytes < buflen) - { - ret = send(_socket, buf+bytes, buflen-bytes, 0); - //check for errors - if (ret == -1) - { - _logger->log_error("Site2Site Peer socket %d send failed %s", _socket, strerror(errno)); - Close(); - // this->yield(); - return ret; - } - bytes+=ret; - } - - if (crc) - crc->update(buf, buflen); - - return bytes; -} - -int Site2SitePeer::Select(int msec) -{ - fd_set fds; - struct timeval tv; - int retval; - int fd = _socket; - - FD_ZERO(&fds); - FD_SET(fd, &fds); - - tv.tv_sec = msec/1000; - tv.tv_usec = (msec % 1000) * 1000; - - if (msec > 0) - retval = select(fd+1, &fds, NULL, NULL, &tv); - else - retval = select(fd+1, &fds, NULL, NULL, NULL); - - if (retval <= 0) - return retval; - if (FD_ISSET(fd, &fds)) - return retval; - else - return 0; -} - -int Site2SitePeer::readData(uint8_t *buf, int buflen, CRC32 *crc) -{ - int sendSize = buflen; - uint8_t *start = buf; - - if (_socket <= 0) - { - // this->yield(); - return -1; - } - - while (buflen) - { - int status; - status = Select((int) _timeOut); - if (status <= 0) - { - Close(); - return status; - } - status = recv(_socket, buf, buflen, 0); - if (status <= 0) - { - Close(); - // this->yield(); - return status; - } - buflen -= status; - buf += status; - } - - if (crc) - crc->update(start, sendSize); - - return sendSize; -} - -int Site2SitePeer::writeUTF(std::string str, bool widen, CRC32 *crc) -{ - int strlen = str.length(); - int utflen = 0; - int c, count = 0; - - /* use charAt instead of copying String to char array */ - for (int i = 0; i < strlen; i++) { - c = str.at(i); - if ((c >= 0x0001) && (c <= 0x007F)) { - utflen++; - } else if (c > 0x07FF) { - utflen += 3; - } else { - utflen += 2; - } - } - - if (utflen > 65535) - return -1; - - uint8_t *bytearr = NULL; - if (!widen) - { - bytearr = new uint8_t[utflen+2]; - bytearr[count++] = (uint8_t) ((utflen >> 8) & 0xFF); - bytearr[count++] = (uint8_t) ((utflen >> 0) & 0xFF); - } - else - { - bytearr = new uint8_t[utflen+4]; - bytearr[count++] = (uint8_t) ((utflen >> 24) & 0xFF); - bytearr[count++] = (uint8_t) ((utflen >> 16) & 0xFF); - bytearr[count++] = (uint8_t) ((utflen >> 8) & 0xFF); - bytearr[count++] = (uint8_t) ((utflen >> 0) & 0xFF); - } - - int i=0; - for (i=0; i<strlen; i++) { - c = str.at(i); - if (!((c >= 0x0001) && (c <= 0x007F))) break; - bytearr[count++] = (uint8_t) c; - } - - for (;i < strlen; i++){ - c = str.at(i); - if ((c >= 0x0001) && (c <= 0x007F)) { - bytearr[count++] = (uint8_t) c; - } else if (c > 0x07FF) { - bytearr[count++] = (uint8_t) (0xE0 | ((c >> 12) & 0x0F)); - bytearr[count++] = (uint8_t) (0x80 | ((c >> 6) & 0x3F)); - bytearr[count++] = (uint8_t) (0x80 | ((c >> 0) & 0x3F)); - } else { - bytearr[count++] = (uint8_t) (0xC0 | ((c >> 6) & 0x1F)); - bytearr[count++] = (uint8_t) (0x80 | ((c >> 0) & 0x3F)); - } - } - int ret; - if (!widen) - { - ret = sendData(bytearr, utflen+2, crc); - } - else - { - ret = sendData(bytearr, utflen+4, crc); - } - delete[] bytearr; - return ret; -} - -int Site2SitePeer::readUTF(std::string &str, bool widen, CRC32 *crc) -{ - uint16_t utflen; - int ret; - - if (!widen) - { - ret = read(utflen, crc); - if (ret <= 0) - return ret; - } - else - { - uint32_t len; - ret = read(len, crc); - if (ret <= 0) - return ret; - utflen = len; - } - - uint8_t *bytearr = NULL; - char *chararr = NULL; - bytearr = new uint8_t[utflen]; - chararr = new char[utflen]; - memset(chararr, 0, utflen); - - int c, char2, char3; - int count = 0; - int chararr_count=0; - - ret = read(bytearr, utflen, crc); - if (ret <= 0) - { - delete[] bytearr; - delete[] chararr; - return ret; - } - - while (count < utflen) { - c = (int) bytearr[count] & 0xff; - if (c > 127) break; - count++; - chararr[chararr_count++]=(char)c; - } - - while (count < utflen) { - c = (int) bytearr[count] & 0xff; - switch (c >> 4) { - case 0: case 1: case 2: case 3: case 4: case 5: case 6: case 7: - /* 0xxxxxxx*/ - count++; - chararr[chararr_count++]=(char)c; - break; - case 12: case 13: - /* 110x xxxx 10xx xxxx*/ - count += 2; - if (count > utflen) - { - delete[] bytearr; - delete[] chararr; - return -1; - } - char2 = (int) bytearr[count-1]; - if ((char2 & 0xC0) != 0x80) - { - delete[] bytearr; - delete[] chararr; - return -1; - } - chararr[chararr_count++]=(char)(((c & 0x1F) << 6) | - (char2 & 0x3F)); - break; - case 14: - /* 1110 xxxx 10xx xxxx 10xx xxxx */ - count += 3; - if (count > utflen) - { - delete[] bytearr; - delete[] chararr; - return -1; - } - char2 = (int) bytearr[count-2]; - char3 = (int) bytearr[count-1]; - if (((char2 & 0xC0) != 0x80) || ((char3 & 0xC0) != 0x80)) - { - delete[] bytearr; - delete[] chararr; - return -1; - } - chararr[chararr_count++]=(char)(((c & 0x0F) << 12) | - ((char2 & 0x3F) << 6) | - ((char3 & 0x3F) << 0)); - break; - default: - delete[] bytearr; - delete[] chararr; - return -1; - } - } - // The number of chars produced may be less than utflen - std::string value(chararr, chararr_count); - str = value; - delete[] bytearr; - delete[] chararr; - if (!widen) - return (2 + utflen); - else - return (4 + utflen); -} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/b02af540/src/TailFile.cpp ---------------------------------------------------------------------- diff --git a/src/TailFile.cpp b/src/TailFile.cpp deleted file mode 100644 index 445255b..0000000 --- a/src/TailFile.cpp +++ /dev/null @@ -1,272 +0,0 @@ -/** - * @file TailFile.cpp - * TailFile 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 <sys/types.h> -#include <sys/stat.h> -#include <time.h> -#include <sstream> -#include <stdio.h> -#include <string> -#include <iostream> -#include <dirent.h> -#include <limits.h> -#include <unistd.h> - -#include "TimeUtil.h" -#include "TailFile.h" -#include "ProcessContext.h" -#include "ProcessSession.h" - -const std::string TailFile::ProcessorName("TailFile"); -Property TailFile::FileName("File to Tail", "Fully-qualified filename of the file that should be tailed", ""); -Property TailFile::StateFile("State File", - "Specifies the file that should be used for storing state about what data has been ingested so that upon restart NiFi can resume from where it left off", ""); -Relationship TailFile::Success("success", "All files are routed to success"); - -void TailFile::initialize() -{ - //! Set the supported properties - std::set<Property> properties; - properties.insert(FileName); - properties.insert(StateFile); - setSupportedProperties(properties); - //! Set the supported relationships - std::set<Relationship> relationships; - relationships.insert(Success); - setSupportedRelationships(relationships); -} - -std::string TailFile::trimLeft(const std::string& s) -{ - const char *WHITESPACE = " \n\r\t"; - size_t startpos = s.find_first_not_of(WHITESPACE); - return (startpos == std::string::npos) ? "" : s.substr(startpos); -} - -std::string TailFile::trimRight(const std::string& s) -{ - const char *WHITESPACE = " \n\r\t"; - size_t endpos = s.find_last_not_of(WHITESPACE); - return (endpos == std::string::npos) ? "" : s.substr(0, endpos+1); -} - -void TailFile::parseStateFileLine(char *buf) -{ - char *line = buf; - - while ((line[0] == ' ') || (line[0] =='\t')) - ++line; - - char first = line[0]; - if ((first == '\0') || (first == '#') || (first == '\r') || (first == '\n') || (first == '=')) - { - return; - } - - char *equal = strchr(line, '='); - if (equal == NULL) - { - return; - } - - equal[0] = '\0'; - std::string key = line; - - equal++; - while ((equal[0] == ' ') || (equal[0] == '\t')) - ++equal; - - first = equal[0]; - if ((first == '\0') || (first == '\r') || (first== '\n')) - { - return; - } - - std::string value = equal; - key = trimRight(key); - value = trimRight(value); - - if (key == "FILENAME") - this->_currentTailFileName = value; - if (key == "POSITION") - this->_currentTailFilePosition = std::stoi(value); - - return; -} - -void TailFile::recoverState() -{ - std::ifstream file(_stateFile.c_str(), std::ifstream::in); - if (!file.good()) - { - _logger->log_error("load state file failed %s", _stateFile.c_str()); - return; - } - const unsigned int bufSize = 512; - char buf[bufSize]; - for (file.getline(buf,bufSize); file.good(); file.getline(buf,bufSize)) - { - parseStateFileLine(buf); - } -} - -void TailFile::storeState() -{ - std::ofstream file(_stateFile.c_str()); - if (!file.is_open()) - { - _logger->log_error("store state file failed %s", _stateFile.c_str()); - return; - } - file << "FILENAME=" << this->_currentTailFileName << "\n"; - file << "POSITION=" << this->_currentTailFilePosition << "\n"; - file.close(); -} - -static bool sortTailMatchedFileItem(TailMatchedFileItem i, TailMatchedFileItem j) -{ - return (i.modifiedTime < j.modifiedTime); -} -void TailFile::checkRollOver() -{ - struct stat statbuf; - std::vector<TailMatchedFileItem> matchedFiles; - std::string fullPath = this->_fileLocation + "/" + _currentTailFileName; - - if (stat(fullPath.c_str(), &statbuf) == 0) - { - if (statbuf.st_size > this->_currentTailFilePosition) - // there are new input for the current tail file - return; - - uint64_t modifiedTimeCurrentTailFile = ((uint64_t) (statbuf.st_mtime) * 1000); - std::string pattern = _fileName; - std::size_t found = _fileName.find_last_of("."); - if (found != std::string::npos) - pattern = _fileName.substr(0,found); - DIR *d; - d = opendir(this->_fileLocation.c_str()); - if (!d) - return; - while (1) - { - struct dirent *entry; - entry = readdir(d); - if (!entry) - break; - std::string d_name = entry->d_name; - if (!(entry->d_type & DT_DIR)) - { - std::string fileName = d_name; - std::string fileFullName = this->_fileLocation + "/" + d_name; - if (fileFullName.find(pattern) != std::string::npos && stat(fileFullName.c_str(), &statbuf) == 0) - { - if (((uint64_t) (statbuf.st_mtime) * 1000) >= modifiedTimeCurrentTailFile) - { - TailMatchedFileItem item; - item.fileName = fileName; - item.modifiedTime = ((uint64_t) (statbuf.st_mtime) * 1000); - matchedFiles.push_back(item); - } - } - } - } - closedir(d); - - // Sort the list based on modified time - std::sort(matchedFiles.begin(), matchedFiles.end(), sortTailMatchedFileItem); - for (std::vector<TailMatchedFileItem>::iterator it = matchedFiles.begin(); it!=matchedFiles.end(); ++it) - { - TailMatchedFileItem item = *it; - if (item.fileName == _currentTailFileName) - { - ++it; - if (it!=matchedFiles.end()) - { - TailMatchedFileItem nextItem = *it; - _logger->log_info("TailFile File Roll Over from %s to %s", _currentTailFileName.c_str(), nextItem.fileName.c_str()); - _currentTailFileName = nextItem.fileName; - _currentTailFilePosition = 0; - storeState(); - } - break; - } - } - } - else - return; -} - - -void TailFile::onTrigger(ProcessContext *context, ProcessSession *session) -{ - std::string value; - if (context->getProperty(FileName.getName(), value)) - { - std::size_t found = value.find_last_of("/\\"); - this->_fileLocation = value.substr(0,found); - this->_fileName = value.substr(found+1); - } - if (context->getProperty(StateFile.getName(), value)) - { - _stateFile = value; - } - if (!this->_stateRecovered) - { - _stateRecovered = true; - this->_currentTailFileName = _fileName; - this->_currentTailFilePosition = 0; - // recover the state if we have not done so - this->recoverState(); - } - checkRollOver(); - std::string fullPath = this->_fileLocation + "/" + _currentTailFileName; - struct stat statbuf; - if (stat(fullPath.c_str(), &statbuf) == 0) - { - if (statbuf.st_size <= this->_currentTailFilePosition) - // there are no new input for the current tail file - { - context->yield(); - return; - } - FlowFileRecord *flowFile = session->create(); - if (!flowFile) - return; - std::size_t found = _currentTailFileName.find_last_of("."); - std::string baseName = _currentTailFileName.substr(0,found); - std::string extension = _currentTailFileName.substr(found+1); - flowFile->updateAttribute(PATH, _fileLocation); - flowFile->addAttribute(ABSOLUTE_PATH, fullPath); - session->import(fullPath, flowFile, true, this->_currentTailFilePosition); - session->transfer(flowFile, Success); - _logger->log_info("TailFile %s for %d bytes", _currentTailFileName.c_str(), flowFile->getSize()); - std::string logName = baseName + "." + std::to_string(_currentTailFilePosition) + "-" + - std::to_string(_currentTailFilePosition + flowFile->getSize()) + "." + extension; - flowFile->updateAttribute(FILENAME, logName); - this->_currentTailFilePosition += flowFile->getSize(); - storeState(); - } -} - http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/b02af540/src/TimerDrivenSchedulingAgent.cpp ---------------------------------------------------------------------- diff --git a/src/TimerDrivenSchedulingAgent.cpp b/src/TimerDrivenSchedulingAgent.cpp deleted file mode 100644 index 3ce57ae..0000000 --- a/src/TimerDrivenSchedulingAgent.cpp +++ /dev/null @@ -1,134 +0,0 @@ -/** - * @file TimerDrivenSchedulingAgent.cpp - * TimerDrivenSchedulingAgent 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 "Property.h" -#include "TimerDrivenSchedulingAgent.h" - -void TimerDrivenSchedulingAgent::schedule(Processor *processor) -{ - std::lock_guard<std::mutex> lock(_mtx); - - _administrativeYieldDuration = 0; - std::string yieldValue; - - if (_configure->get(Configure::nifi_administrative_yield_duration, yieldValue)) - { - TimeUnit unit; - if (Property::StringToTime(yieldValue, _administrativeYieldDuration, unit) && - Property::ConvertTimeUnitToMS(_administrativeYieldDuration, unit, _administrativeYieldDuration)) - { - _logger->log_debug("nifi_administrative_yield_duration: [%d] ms", _administrativeYieldDuration); - } - } - - _boredYieldDuration = 0; - if (_configure->get(Configure::nifi_bored_yield_duration, yieldValue)) - { - TimeUnit unit; - if (Property::StringToTime(yieldValue, _boredYieldDuration, unit) && - Property::ConvertTimeUnitToMS(_boredYieldDuration, unit, _boredYieldDuration)) - { - _logger->log_debug("nifi_bored_yield_duration: [%d] ms", _boredYieldDuration); - } - } - - if (processor->getScheduledState() != RUNNING) - { - _logger->log_info("Can not schedule threads for processor %s because it is not running", processor->getName().c_str()); - return; - } - - std::map<std::string, std::vector<std::thread *>>::iterator it = - _threads.find(processor->getUUIDStr()); - if (it != _threads.end()) - { - _logger->log_info("Can not schedule threads for processor %s because there are existed thread running"); - return; - } - - std::vector<std::thread *> threads; - for (int i = 0; i < processor->getMaxConcurrentTasks(); i++) - { - std::thread *thread = new std::thread(run, this, processor); - thread->detach(); - threads.push_back(thread); - _logger->log_info("Scheduled Time Driven thread %d running for process %s", thread->get_id(), - processor->getName().c_str()); - } - _threads[processor->getUUIDStr().c_str()] = threads; - - return; -} - -void TimerDrivenSchedulingAgent::unschedule(Processor *processor) -{ - std::lock_guard<std::mutex> lock(_mtx); - - if (processor->getScheduledState() != RUNNING) - { - _logger->log_info("Can not unschedule threads for processor %s because it is not running", processor->getName().c_str()); - return; - } - - std::map<std::string, std::vector<std::thread *>>::iterator it = - _threads.find(processor->getUUIDStr()); - - if (it == _threads.end()) - { - _logger->log_info("Can not unschedule threads for processor %s because there are no existed thread running"); - return; - } - for (std::vector<std::thread *>::iterator itThread = it->second.begin(); itThread != it->second.end(); ++itThread) - { - std::thread *thread = *itThread; - _logger->log_info("Scheduled Time Driven thread %d deleted for process %s", thread->get_id(), - processor->getName().c_str()); - delete thread; - } - _threads.erase(processor->getUUIDStr()); - processor->clearActiveTask(); - - return; -} - -void TimerDrivenSchedulingAgent::run(TimerDrivenSchedulingAgent *agent, Processor *processor) -{ - while (agent->_running) - { - bool shouldYield = agent->onTrigger(processor); - - if (processor->isYield()) - { - // Honor the yield - std::this_thread::sleep_for(std::chrono::milliseconds(processor->getYieldTime())); - } - else if (shouldYield && agent->_boredYieldDuration > 0) - { - // No work to do or need to apply back pressure - std::this_thread::sleep_for(std::chrono::milliseconds(agent->_boredYieldDuration)); - } - std::this_thread::sleep_for(std::chrono::nanoseconds(processor->getSchedulingPeriodNano())); - } - return; -} - - http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/b02af540/test/FlowFileRecordTest.cpp ---------------------------------------------------------------------- diff --git a/test/FlowFileRecordTest.cpp b/test/FlowFileRecordTest.cpp deleted file mode 100644 index 09a3d33..0000000 --- a/test/FlowFileRecordTest.cpp +++ /dev/null @@ -1,28 +0,0 @@ -/** - * @file MiNiFiMain.cpp - * MiNiFiMain 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 "FlowFileRecord.h" - -int main(int argc, char **argv) -{ -} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/b02af540/test/Server.cpp ---------------------------------------------------------------------- diff --git a/test/Server.cpp b/test/Server.cpp deleted file mode 100644 index e7b3452..0000000 --- a/test/Server.cpp +++ /dev/null @@ -1,607 +0,0 @@ -/* A simple server in the internet domain using TCP - The port number is passed as an argument */ -#include <stdio.h> -#include <stdlib.h> -#include <string.h> -#include <unistd.h> -#include <sys/types.h> -#include <sys/socket.h> -#include <netinet/in.h> -#include <errno.h> -#include <arpa/inet.h> -#include <fcntl.h> -#include <netdb.h> -#include <string> -#include <errno.h> -#include <chrono> -#include <thread> -#include <iostream> // std::cout -#include <fstream> // std::ifstream -#include <signal.h> - -#define DEFAULT_NIFI_SERVER_PORT 9000 -#define DEFAULT_REPORT_INTERVAL 1000 // 1 sec -#define MAX_READ_TIMEOUT 30000 // 30 seconds - -//! FlowControl Protocol Msg Type -typedef enum { - REGISTER_REQ, // Device Register Request from device to server which contain device serial number, current running flow xml version - REGISTER_RESP, // Device Register Respond from server to device, may contain new flow.xml from server ask device to apply and also device report interval - REPORT_REQ, // Period Device Report from device to server which contain device serial number, current running flow xml name/version and other period report info - REPORT_RESP, // Report Respond from server to device, may ask device to update flow xml or processor property - MAX_FLOW_CONTROL_MSG_TYPE -} FlowControlMsgType; - -//! FlowControl Protocol Msg Type String -static const char *FlowControlMsgTypeStr[MAX_FLOW_CONTROL_MSG_TYPE] = -{ - "REGISTER_REQ", - "REGISTER_RESP", - "REPORT_REQ", - "REPORT_RESP" -}; - -//! Flow Control Msg Type to String -inline const char *FlowControlMsgTypeToStr(FlowControlMsgType type) -{ - if (type < MAX_FLOW_CONTROL_MSG_TYPE) - return FlowControlMsgTypeStr[type]; - else - return NULL; -} - -//! FlowControll Protocol Msg ID (Some Messages are fix length, Some are variable length (TLV) -typedef enum { - //Fix length 8 bytes: client to server in register request, required field - FLOW_SERIAL_NUMBER, - // Flow XML name TLV: client to server in register request and report request, required field - FLOW_XML_NAME, - // Flow XML content, TLV: server to client in register respond, option field in case server want to ask client to load xml from server - FLOW_XML_CONTENT, - // Fix length, 4 bytes Report interval in msec: server to client in register respond, option field - REPORT_INTERVAL, - // Processor Name TLV: server to client in report respond, option field in case server want to ask client to update processor property - PROCESSOR_NAME, - // Processor Property Name TLV: server to client in report respond, option field in case server want to ask client to update processor property - PROPERTY_NAME, - // Processor Property Value TLV: server to client in report respond, option field in case server want to ask client to update processor property - PROPERTY_VALUE, - // Report Blob TLV: client to server in report request, option field in case client want to pickyback the report blob in report request to server - REPORT_BLOB, - MAX_FLOW_MSG_ID -} FlowControlMsgID; - -//! FlowControl Protocol Msg ID String -static const char *FlowControlMsgIDStr[MAX_FLOW_MSG_ID] = -{ - "FLOW_SERIAL_NUMBER", - "FLOW_XML_NAME", - "FLOW_XML_CONTENT", - "REPORT_INTERVAL", - "PROCESSOR_NAME" - "PROPERTY_NAME", - "PROPERTY_VALUE", - "REPORT_BLOB" -}; - -#define TYPE_HDR_LEN 4 // Fix Hdr Type -#define TLV_HDR_LEN 8 // Type 4 bytes and Len 4 bytes - -//! FlowControl Protocol Msg Len -inline int FlowControlMsgIDEncodingLen(FlowControlMsgID id, int payLoadLen) -{ - if (id == FLOW_SERIAL_NUMBER) - return (TYPE_HDR_LEN + 8); - else if (id == REPORT_INTERVAL) - return (TYPE_HDR_LEN + 4); - else if (id < MAX_FLOW_MSG_ID) - return (TLV_HDR_LEN + payLoadLen); - else - return -1; -} - -//! Flow Control Msg Id to String -inline const char *FlowControlMsgIdToStr(FlowControlMsgID id) -{ - if (id < MAX_FLOW_MSG_ID) - return FlowControlMsgIDStr[id]; - else - return NULL; -} - -//! Flow Control Respond status code -typedef enum { - RESP_SUCCESS, - RESP_TRIGGER_REGISTER, // Server respond to client report to re trigger register - RESP_START_FLOW_CONTROLLER, // Server respond to client to start flow controller - RESP_STOP_FLOW_CONTROLLER, // Server respond to client to stop flow controller - RESP_FAILURE, - MAX_RESP_CODE -} FlowControlRespCode; - -//! FlowControl Resp Code str -static const char *FlowControlRespCodeStr[MAX_RESP_CODE] = -{ - "RESP_SUCCESS", - "RESP_TRIGGER_REGISTER", - "RESP_START_FLOW_CONTROLLER", - "RESP_STOP_FLOW_CONTROLLER", - "RESP_FAILURE" -}; - -//! Flow Control Resp Code to String -inline const char *FlowControlRespCodeToStr(FlowControlRespCode code) -{ - if (code < MAX_RESP_CODE) - return FlowControlRespCodeStr[code]; - else - return NULL; -} - -//! Common FlowControlProtocol Header -typedef struct { - uint32_t msgType; //! Msg Type - uint32_t seqNumber; //! Seq Number to match Req with Resp - uint32_t status; //! Resp Code, see FlowControlRespCode - uint32_t payloadLen; //! Msg Payload length -} FlowControlProtocolHeader; - - -//! encode uint32_t -uint8_t *encode(uint8_t *buf, uint32_t value) -{ - *buf++ = (value & 0xFF000000) >> 24; - *buf++ = (value & 0x00FF0000) >> 16; - *buf++ = (value & 0x0000FF00) >> 8; - *buf++ = (value & 0x000000FF); - return buf; -} - -//! encode uint32_t -uint8_t *decode(uint8_t *buf, uint32_t &value) -{ - value = ((buf[0]<<24)|(buf[1]<<16)|(buf[2]<<8)|(buf[3])); - return (buf + 4); -} - -//! encode byte array -uint8_t *encode(uint8_t *buf, uint8_t *bufArray, int size) -{ - memcpy(buf, bufArray, size); - buf += size; - return buf; -} - -//! encode std::string -uint8_t *encode(uint8_t *buf, std::string value) -{ - // add the \0 for size - buf = encode(buf, value.size()+1); - buf = encode(buf, (uint8_t *) value.c_str(), value.size()+1); - return buf; -} - -int sendData(int socket, uint8_t *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; - } - - return bytes; -} - -void error(const char *msg) -{ - perror(msg); - exit(1); -} - -/* readline - read a '\n' terminated line from socket fd - into buffer bufptr of size len. The line in the - buffer is terminated with '\0'. - It returns -1 in case of error or if - the capacity of the buffer is exceeded. - It returns 0 if EOF is encountered before reading '\n'. - */ -int readline( int fd, char *bufptr, size_t len ) -{ - /* Note that this function is very tricky. It uses the - static variables bp, cnt, and b to establish a local buffer. - The recv call requests large chunks of data (the size of the buffer). - Then if the recv call reads more than one line, the overflow - remains in the buffer and it is made available to the next call - to readline. - Notice also that this routine reads up to '\n' and overwrites - it with '\0'. Thus if the line is really terminated with - "\r\n", the '\r' will remain unchanged. - */ - char *bufx = bufptr; - static char *bp; - static int cnt = 0; - static char b[ 4096 ]; - 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 = '\0'; - return bufptr - bufx; - } - } - return -1; -} - -int readData(int socket, uint8_t *buf, int buflen) -{ - int sendSize = buflen; - int status; - - while (buflen) - { -#ifndef __MACH__ - status = read(socket, buf, buflen); -#else - status = recv(socket, buf, buflen, 0); -#endif - if (status <= 0) - { - return status; - } - buflen -= status; - buf += status; - } - - return sendSize; -} - -int readHdr(int socket, FlowControlProtocolHeader *hdr) -{ - uint8_t buffer[sizeof(FlowControlProtocolHeader)]; - - uint8_t *data = buffer; - - int status = readData(socket, buffer, sizeof(FlowControlProtocolHeader)); - if (status <= 0) - return status; - - uint32_t value; - data = decode(data, value); - hdr->msgType = value; - - data = decode(data, value); - hdr->seqNumber = value; - - data = decode(data, value); - hdr->status = value; - - data = decode(data, value); - hdr->payloadLen = value; - - return sizeof(FlowControlProtocolHeader); -} - -int readXML(char **xmlContent) -{ - std::ifstream is ("conf/flowServer.xml", std::ifstream::binary); - if (is) { - // get length of file: - is.seekg (0, is.end); - int length = is.tellg(); - is.seekg (0, is.beg); - - char * buffer = new char [length]; - - printf("Reading %s len %d\n", "conf/flowServer.xml", length); - // read data as a block: - is.read (buffer,length); - - is.close(); - - // ...buffer contains the entire file... - *xmlContent = buffer; - - return length; - } - return 0; -} - -static int sockfd = 0, newsockfd = 0; -void sigHandler(int signal) -{ - if (signal == SIGINT || signal == SIGTERM) - { - close(newsockfd); - close(sockfd); - exit(1); - } -} - -int main(int argc, char *argv[]) -{ - int portno; - socklen_t clilen; - struct sockaddr_in serv_addr, cli_addr; - char buffer[4096]; - int flag = 0; - int number = 0; - - int n; - if (argc < 2) { - fprintf(stderr,"ERROR, no port provided\n"); - exit(1); - } - - if (signal(SIGINT, sigHandler) == SIG_ERR || signal(SIGTERM, sigHandler) == SIG_ERR) - { - - return -1; - } - sockfd = socket(AF_INET, SOCK_STREAM, 0); - if (sockfd < 0) - error("ERROR opening socket"); - bzero((char *) &serv_addr, sizeof(serv_addr)); - portno = atoi(argv[1]); - 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) - error("ERROR on binding"); - listen(sockfd,5); - if (portno == DEFAULT_NIFI_SERVER_PORT) - { - while (true) - { - clilen = sizeof(cli_addr); - newsockfd = accept(sockfd, - (struct sockaddr *) &cli_addr, - &clilen); - if (newsockfd < 0) - { - error("ERROR on accept"); - break; - } - // process request - FlowControlProtocolHeader hdr; - int status = readHdr(newsockfd, &hdr); - if (status > 0) - { - printf("Flow Control Protocol receive MsgType %s\n", FlowControlMsgTypeToStr((FlowControlMsgType) hdr.msgType)); - printf("Flow Control Protocol receive Seq Num %d\n", hdr.seqNumber); - printf("Flow Control Protocol receive Resp Code %s\n", FlowControlRespCodeToStr((FlowControlRespCode) hdr.status)); - printf("Flow Control Protocol receive Payload len %d\n", hdr.payloadLen); - if (((FlowControlMsgType) hdr.msgType) == REGISTER_REQ) - { - printf("Flow Control Protocol Register Req receive\n"); - uint8_t *payload = new uint8_t[hdr.payloadLen]; - uint8_t *payloadPtr = payload; - status = readData(newsockfd, payload, hdr.payloadLen); - while (status > 0 && payloadPtr < (payload + hdr.payloadLen)) - { - uint32_t msgID = 0xFFFFFFFF; - payloadPtr = decode(payloadPtr, msgID); - if (((FlowControlMsgID) msgID) == FLOW_SERIAL_NUMBER) - { - // Fixed 8 bytes - uint8_t seqNum[8]; - memcpy(seqNum, payloadPtr, 8); - printf("Flow Control Protocol Register Req receive serial num\n"); - payloadPtr += 8; - } - else if (((FlowControlMsgID) msgID) == FLOW_XML_NAME) - { - uint32_t len; - payloadPtr = decode(payloadPtr, len); - printf("Flow Control Protocol receive XML name length %d\n", len); - std::string flowName = (const char *) payloadPtr; - payloadPtr += len; - printf("Flow Control Protocol receive XML name %s\n", flowName.c_str()); - } - else - { - break; - } - } - delete[] payload; - // Send Register Respond - // Calculate the total payload msg size - char *xmlContent; - uint32_t xmlLen = readXML(&xmlContent); - uint32_t payloadSize = FlowControlMsgIDEncodingLen(REPORT_INTERVAL, 0); - if (xmlLen > 0) - payloadSize += FlowControlMsgIDEncodingLen(FLOW_XML_CONTENT, xmlLen); - - uint32_t size = sizeof(FlowControlProtocolHeader) + payloadSize; - uint8_t *data = new uint8_t[size]; - uint8_t *start = data; - - // encode the HDR - hdr.msgType = REGISTER_RESP; - hdr.payloadLen = payloadSize; - hdr.status = RESP_SUCCESS; - data = encode(data, hdr.msgType); - data = encode(data, hdr.seqNumber); - data = encode(data, hdr.status); - data = encode(data, hdr.payloadLen); - - // encode the report interval - data = encode(data, REPORT_INTERVAL); - data = encode(data, DEFAULT_REPORT_INTERVAL); - - // encode the XML content - if (xmlLen > 0) - { - data = encode(data, FLOW_XML_CONTENT); - data = encode(data, xmlLen); - data = encode(data, (uint8_t *) xmlContent, xmlLen); - delete[] xmlContent; - } - - // send it - status = sendData(newsockfd, start, size); - delete[] start; - } - else if (((FlowControlMsgType) hdr.msgType) == REPORT_REQ) - { - printf("Flow Control Protocol Report Req receive\n"); - uint8_t *payload = new uint8_t[hdr.payloadLen]; - uint8_t *payloadPtr = payload; - status = readData(newsockfd, payload, hdr.payloadLen); - while (status > 0 && payloadPtr < (payload + hdr.payloadLen)) - { - uint32_t msgID = 0xFFFFFFFF; - payloadPtr = decode(payloadPtr, msgID); - if (((FlowControlMsgID) msgID) == FLOW_XML_NAME) - { - uint32_t len; - payloadPtr = decode(payloadPtr, len); - printf("Flow Control Protocol receive XML name length %d\n", len); - std::string flowName = (const char *) payloadPtr; - payloadPtr += len; - printf("Flow Control Protocol receive XML name %s\n", flowName.c_str()); - } - else - { - break; - } - } - delete[] payload; - // Send Register Respond - // Calculate the total payload msg size - std::string processor = "RealTimeDataCollector"; - std::string propertyName1 = "real Time Message ID"; - std::string propertyValue1 = "41"; - std::string propertyName2 = "Batch Message ID"; - std::string propertyValue2 = "172,30,48"; - if (flag == 0) - { - propertyName1 = "Real Time Message ID"; - propertyValue1 = "41"; - propertyName2 = "Batch Message ID"; - propertyValue2 = "172,48"; - flag = 1; - } - else if (flag == 1) - { - propertyName1 = "Real Time Message ID"; - propertyValue1 = "172,48"; - propertyName2 = "Batch Message ID"; - propertyValue2 = "41"; - flag = 0; - } - uint32_t payloadSize = FlowControlMsgIDEncodingLen(PROCESSOR_NAME, processor.size()+1); - payloadSize += FlowControlMsgIDEncodingLen(PROPERTY_NAME, propertyName1.size()+1); - payloadSize += FlowControlMsgIDEncodingLen(PROPERTY_VALUE, propertyValue1.size()+1); - payloadSize += FlowControlMsgIDEncodingLen(PROPERTY_NAME, propertyName2.size()+1); - payloadSize += FlowControlMsgIDEncodingLen(PROPERTY_VALUE, propertyValue2.size()+1); - - uint32_t size = sizeof(FlowControlProtocolHeader) + payloadSize; - uint8_t *data = new uint8_t[size]; - uint8_t *start = data; - - // encode the HDR - hdr.msgType = REPORT_RESP; - hdr.payloadLen = payloadSize; - hdr.status = RESP_SUCCESS; - - if (number >= 10 && number < 20) - { - // After 10 second report, stop the flow controller for 10 second - hdr.status = RESP_STOP_FLOW_CONTROLLER; - } - else if (number == 20) - { - // restart the flow controller after 10 second - hdr.status = RESP_START_FLOW_CONTROLLER; - } - else if (number == 30) - { - // retrigger register - hdr.status = RESP_TRIGGER_REGISTER; - number = 0; - } - - number++; - - data = encode(data, hdr.msgType); - data = encode(data, hdr.seqNumber); - data = encode(data, hdr.status); - data = encode(data, hdr.payloadLen); - - // encode the processorName - data = encode(data, PROCESSOR_NAME); - data = encode(data, processor); - - // encode the propertyName and value TLV - data = encode(data, PROPERTY_NAME); - data = encode(data, propertyName1); - data = encode(data, PROPERTY_VALUE); - data = encode(data, propertyValue1); - data = encode(data, PROPERTY_NAME); - data = encode(data, propertyName2); - data = encode(data, PROPERTY_VALUE); - data = encode(data, propertyValue2); - // send it - status = sendData(newsockfd, start, size); - delete[] start; - } - } - close(newsockfd); - } - close(sockfd); - } - else - { - clilen = sizeof(cli_addr); - newsockfd = accept(sockfd, - (struct sockaddr *) &cli_addr, - &clilen); - if (newsockfd < 0) - error("ERROR on accept"); - while (1) - { - bzero(buffer,4096); - n = readline(newsockfd,buffer,4095); - if (n <= 0 ) - { - close(newsockfd); - newsockfd = accept(sockfd, - (struct sockaddr *) &cli_addr, - &clilen); - continue; - } - printf("%s",buffer); - } - close(newsockfd); - close(sockfd); - } - return 0; -} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/b02af540/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 deleted file mode 100644 index f23f477..0000000 --- a/thirdparty/yaml-cpp-yaml-cpp-0.5.3/Makefile +++ /dev/null @@ -1,40 +0,0 @@ -# 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
