MCPcopy Create free account
hub / github.com/apache/nifi-minifi-cpp / transmitPayload

Method transmitPayload

libminifi/src/sitetosite/RawSocketProtocol.cpp:533–599  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

531}
532
533bool RawSiteToSiteClient::transmitPayload(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSession> &session, const std::string &payload,
534 std::map<std::string, std::string> attributes) {
535 std::shared_ptr<Transaction> transaction = NULL;
536
537 if (payload.length() <= 0)
538 return false;
539
540 if (peer_state_ != READY) {
541 if (!bootstrap()) {
542 return false;
543 }
544 }
545
546 if (peer_state_ != READY) {
547 context->yield();
548 tearDown();
549 throw Exception(SITE2SITE_EXCEPTION, "Can not establish handshake with peer");
550 }
551
552 // Create the transaction
553 transaction = createTransaction(SEND);
554
555 if (transaction == NULL) {
556 context->yield();
557 tearDown();
558 throw Exception(SITE2SITE_EXCEPTION, "Can not create transaction");
559 }
560
561 utils::Identifier transactionID = transaction->getUUID();
562
563 try {
564 DataPacket packet(getLogger(), transaction, attributes, payload);
565
566 int16_t resp = send(transactionID, &packet, nullptr, session);
567 if (resp == -1) {
568 throw Exception(SITE2SITE_EXCEPTION, "Send Failed in transaction " + transactionID.to_string());
569 }
570 logging::LOG_INFO(logger_) << "Site2Site transaction " << transactionID.to_string() << " sent bytes length" << payload.length();
571
572 if (!confirm(transactionID)) {
573 throw Exception(SITE2SITE_EXCEPTION, "Confirm Failed in transaction " + transactionID.to_string());
574 }
575 if (!complete(transactionID)) {
576 throw Exception(SITE2SITE_EXCEPTION, "Complete Failed in transaction " + transactionID.to_string());
577 }
578 logging::LOG_INFO(logger_) << "Site2Site transaction " << transactionID.to_string()
579 << " successfully send flow record " << transaction->current_transfers_ << " content bytes " << transaction->_bytes;
580 } catch (std::exception &exception) {
581 if (transaction)
582 deleteTransaction(transactionID);
583 context->yield();
584 tearDown();
585 logger_->log_debug("Caught Exception %s", exception.what());
586 throw;
587 } catch (...) {
588 if (transaction)
589 deleteTransaction(transactionID);
590 context->yield();

Callers 1

onTriggerMethod · 0.45

Calls 13

bootstrapFunction · 0.85
tearDownFunction · 0.85
ExceptionClass · 0.85
createTransactionFunction · 0.85
confirmFunction · 0.85
completeFunction · 0.85
deleteTransactionFunction · 0.85
lengthMethod · 0.80
log_debugMethod · 0.80
yieldMethod · 0.45
getUUIDMethod · 0.45
to_stringMethod · 0.45

Tested by

no test coverage detected