| 531 | } |
| 532 | |
| 533 | bool 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(); |
no test coverage detected