| 941 | } |
| 942 | |
| 943 | void ListSFTP::onTrigger(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSession> &session) { |
| 944 | /* Parse common properties */ |
| 945 | SFTPProcessorBase::CommonProperties common_properties; |
| 946 | if (!parseCommonPropertiesOnTrigger(context, nullptr /*flow_file*/, common_properties)) { |
| 947 | context->yield(); |
| 948 | return; |
| 949 | } |
| 950 | |
| 951 | /* Parse processor-specific properties */ |
| 952 | std::string remote_path; |
| 953 | uint64_t entity_tracking_time_window = 0U; |
| 954 | |
| 955 | std::string value; |
| 956 | context->getProperty(RemotePath.getName(), remote_path); |
| 957 | /* Remove trailing slashes */ |
| 958 | while (remote_path.size() > 1U && remote_path.back() == '/') { |
| 959 | remote_path.resize(remote_path.size() - 1); |
| 960 | } |
| 961 | if (context->getProperty(EntityTrackingTimeWindow.getName(), value)) { |
| 962 | core::TimeUnit unit; |
| 963 | if (!core::Property::StringToTime(value, entity_tracking_time_window, unit) || |
| 964 | !core::Property::ConvertTimeUnitToMS(entity_tracking_time_window, unit, entity_tracking_time_window)) { |
| 965 | /* The default is 3 hours */ |
| 966 | entity_tracking_time_window = 3 * 3600 * 1000; |
| 967 | logger_->log_error("Entity Tracking Time Window attribute is invalid"); |
| 968 | } |
| 969 | } else { |
| 970 | /* The default is 3 hours */ |
| 971 | entity_tracking_time_window = 3 * 3600 * 1000; |
| 972 | } |
| 973 | |
| 974 | /* Check whether we need to invalidate the cache based on the new properties */ |
| 975 | if ((!last_hostname_.empty() && last_hostname_ != common_properties.hostname) || |
| 976 | (!last_username_.empty() && last_username_ != common_properties.username) || |
| 977 | (!last_remote_path_.empty() && last_remote_path_ != remote_path)) { |
| 978 | invalidateCache(); |
| 979 | } |
| 980 | last_hostname_ = common_properties.hostname; |
| 981 | last_username_ = common_properties.username; |
| 982 | last_remote_path_ = remote_path; |
| 983 | |
| 984 | /* Get SFTPClient from cache or create it */ |
| 985 | const SFTPProcessorBase::ConnectionCacheKey connection_cache_key = {common_properties.hostname, |
| 986 | common_properties.port, |
| 987 | common_properties.username, |
| 988 | proxy_type_, |
| 989 | common_properties.proxy_host, |
| 990 | common_properties.proxy_port, |
| 991 | common_properties.proxy_username}; |
| 992 | auto client = getOrCreateConnection(connection_cache_key, |
| 993 | common_properties.password, |
| 994 | common_properties.private_key_path, |
| 995 | common_properties.private_key_passphrase, |
| 996 | common_properties.proxy_password); |
| 997 | if (client == nullptr) { |
| 998 | context->yield(); |
| 999 | return; |
| 1000 | } |
nothing calls this directly
no test coverage detected