| 411 | } |
| 412 | |
| 413 | int16_t SiteToSiteClient::send(const utils::Identifier& transactionID, DataPacket *packet, const std::shared_ptr<core::FlowFile> &flowFile, const std::shared_ptr<core::ProcessSession> &session) { |
| 414 | int ret; |
| 415 | |
| 416 | if (peer_state_ != READY) { |
| 417 | bootstrap(); |
| 418 | } |
| 419 | |
| 420 | if (peer_state_ != READY) { |
| 421 | return -1; |
| 422 | } |
| 423 | |
| 424 | auto it = this->known_transactions_.find(transactionID); |
| 425 | |
| 426 | if (it == known_transactions_.end()) { |
| 427 | return -1; |
| 428 | } |
| 429 | std::shared_ptr<Transaction> transaction = it->second; |
| 430 | |
| 431 | if (transaction->getState() != TRANSACTION_STARTED && transaction->getState() != DATA_EXCHANGED) { |
| 432 | logger_->log_warn("Site2Site transaction %s is not at started or exchanged state", transactionID.to_string()); |
| 433 | return -1; |
| 434 | } |
| 435 | |
| 436 | if (transaction->getDirection() != SEND) { |
| 437 | logger_->log_warn("Site2Site transaction %s direction is wrong", transactionID.to_string()); |
| 438 | return -1; |
| 439 | } |
| 440 | |
| 441 | if (transaction->current_transfers_ > 0) { |
| 442 | ret = writeResponse(transaction, CONTINUE_TRANSACTION, "CONTINUE_TRANSACTION"); |
| 443 | if (ret <= 0) { |
| 444 | return -1; |
| 445 | } |
| 446 | } |
| 447 | // start to read the packet |
| 448 | uint32_t numAttributes = gsl::narrow<uint32_t>(packet->_attributes.size()); |
| 449 | ret = transaction->getStream().write(numAttributes); |
| 450 | if (ret != 4) { |
| 451 | return -1; |
| 452 | } |
| 453 | |
| 454 | std::map<std::string, std::string>::iterator itAttribute; |
| 455 | for (itAttribute = packet->_attributes.begin(); itAttribute != packet->_attributes.end(); itAttribute++) { |
| 456 | ret = transaction->getStream().write(itAttribute->first, true); |
| 457 | |
| 458 | if (ret <= 0) { |
| 459 | return -1; |
| 460 | } |
| 461 | ret = transaction->getStream().write(itAttribute->second, true); |
| 462 | if (ret <= 0) { |
| 463 | return -1; |
| 464 | } |
| 465 | logger_->log_debug("Site2Site transaction %s send attribute key %s value %s", transactionID.to_string(), itAttribute->first, itAttribute->second); |
| 466 | } |
| 467 | |
| 468 | bool flowfile_has_content = (flowFile != nullptr); |
| 469 | |
| 470 | if (flowFile && (flowFile->getResourceClaim() == nullptr || !flowFile->getResourceClaim()->exists())) { |
no test coverage detected