| 676 | } |
| 677 | |
| 678 | void ProcessSession::commit() { |
| 679 | try { |
| 680 | // First we clone the flow record based on the transferred relationship for updated flow record |
| 681 | for (auto && it : _updatedFlowFiles) { |
| 682 | auto record = it.second.modified; |
| 683 | if (routeFlowFile(record) == RouteResult::Error_NoRelationship) { |
| 684 | // Can not find relationship for the flow |
| 685 | throw Exception(PROCESS_SESSION_EXCEPTION, "Can not find the transfer relationship for the updated flow " + record->getUUIDStr()); |
| 686 | } |
| 687 | } |
| 688 | |
| 689 | // Do the same thing for added flow file |
| 690 | for (const auto& it : _addedFlowFiles) { |
| 691 | auto record = it.second; |
| 692 | if (routeFlowFile(record) == RouteResult::Error_NoRelationship) { |
| 693 | // Can not find relationship for the flow |
| 694 | throw Exception(PROCESS_SESSION_EXCEPTION, "Can not find the transfer relationship for the added flow " + record->getUUIDStr()); |
| 695 | } |
| 696 | } |
| 697 | |
| 698 | std::map<std::shared_ptr<Connectable>, std::vector<std::shared_ptr<FlowFile>>> connectionQueues; |
| 699 | |
| 700 | std::shared_ptr<Connectable> connection = nullptr; |
| 701 | // Complete process the added and update flow files for the session, send the flow file to its queue |
| 702 | for (const auto &it : _updatedFlowFiles) { |
| 703 | auto record = it.second.modified; |
| 704 | logger_->log_trace("See %s in %s", record->getUUIDStr(), "_updatedFlowFiles"); |
| 705 | if (record->isDeleted()) { |
| 706 | continue; |
| 707 | } |
| 708 | |
| 709 | connection = record->getConnection(); |
| 710 | if ((connection) != nullptr) { |
| 711 | connectionQueues[connection].push_back(record); |
| 712 | } |
| 713 | } |
| 714 | for (const auto &it : _addedFlowFiles) { |
| 715 | auto record = it.second; |
| 716 | logger_->log_trace("See %s in %s", record->getUUIDStr(), "_addedFlowFiles"); |
| 717 | if (record->isDeleted()) { |
| 718 | continue; |
| 719 | } |
| 720 | connection = record->getConnection(); |
| 721 | if ((connection) != nullptr) { |
| 722 | connectionQueues[connection].push_back(record); |
| 723 | } |
| 724 | } |
| 725 | // Process the clone flow files |
| 726 | for (const auto &record : _clonedFlowFiles) { |
| 727 | logger_->log_trace("See %s in %s", record->getUUIDStr(), "_clonedFlowFiles"); |
| 728 | if (record->isDeleted()) { |
| 729 | continue; |
| 730 | } |
| 731 | connection = record->getConnection(); |
| 732 | if ((connection) != nullptr) { |
| 733 | connectionQueues[connection].push_back(record); |
| 734 | } |
| 735 | } |
no test coverage detected