| 244 | } |
| 245 | |
| 246 | void PutS3Object::onTrigger(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSession> &session) { |
| 247 | logger_->log_debug("PutS3Object onTrigger"); |
| 248 | std::shared_ptr<core::FlowFile> flow_file = session->get(); |
| 249 | if (!flow_file) { |
| 250 | context->yield(); |
| 251 | return; |
| 252 | } |
| 253 | |
| 254 | auto common_properties = getCommonELSupportedProperties(context, flow_file); |
| 255 | if (!common_properties) { |
| 256 | session->transfer(flow_file, Failure); |
| 257 | return; |
| 258 | } |
| 259 | |
| 260 | auto put_s3_request_params = buildPutS3RequestParams(context, flow_file, common_properties.value()); |
| 261 | if (!put_s3_request_params) { |
| 262 | session->transfer(flow_file, Failure); |
| 263 | return; |
| 264 | } |
| 265 | |
| 266 | PutS3Object::ReadCallback callback(flow_file->getSize(), put_s3_request_params.value(), s3_wrapper_.get()); |
| 267 | { |
| 268 | std::lock_guard<std::mutex> lock(s3_wrapper_mutex_); |
| 269 | configureS3Wrapper(common_properties.value()); |
| 270 | session->read(flow_file, &callback); |
| 271 | } |
| 272 | |
| 273 | if (callback.result_ == minifi::utils::nullopt) { |
| 274 | logger_->log_error("Failed to upload S3 object to bucket '%s'", put_s3_request_params->bucket); |
| 275 | session->transfer(flow_file, Failure); |
| 276 | } else { |
| 277 | setAttributes(session, flow_file, put_s3_request_params.value(), callback.result_.value()); |
| 278 | logger_->log_debug("Successfully uploaded S3 object '%s' to bucket '%s'", put_s3_request_params->object_key, put_s3_request_params->bucket); |
| 279 | session->transfer(flow_file, Success); |
| 280 | } |
| 281 | } |
| 282 | |
| 283 | } // namespace processors |
| 284 | } // namespace aws |