@Test public void validateUpdateFlowFileAttributesWithAmqpProperties() { PublishAMQP processor = new PublishAMQP(); ProcessSession processSession = new MockProcessSession(new SharedSessionState(processor, new AtomicLong()), processor); FlowFile sourceFlowFile = processSession.create(); BasicProperties amqpProperties = new AMQP.BasicProperties.Builder() .contentType("text/plain").deliveryMode(2) .priority(1).userId("joe") .build(); FlowFile f2 = AMQPUtils.updateFlowFileAttributesWithAmqpProperties(amqpProperties, sourceFlowFile, processSession); assertEquals("text/plain", f2.getAttributes().get(AMQPUtils.AMQP_PROP_PREFIX + "contentType")); assertEquals("joe", f2.getAttributes().get(AMQPUtils.AMQP_PROP_PREFIX + "userId")); assertEquals("2", f2.getAttributes().get(AMQPUtils.AMQP_PROP_PREFIX + "deliveryMode")); }
public static FlowFile updateFlowFileAttributesWithAmqpProperties(BasicProperties amqpProperties, FlowFile flowFile, ProcessSession processSession) { if (amqpProperties != null) { try { Method[] methods = BasicProperties.class.getDeclaredMethods(); Map<String, String> attributes = new HashMap<>(); for (Method method : methods) { if (Modifier.isPublic(method.getModifiers()) && method.getName().startsWith("get")) { Object amqpPropertyValue = method.invoke(amqpProperties); if (amqpPropertyValue != null) { String propertyName = extractPropertyNameFromMethod(method); if (isValidAmqpPropertyName(propertyName)) { if (propertyName.equals(PropertyNames.CONTENT_TYPE.getValue())) { attributes.put(CoreAttributes.MIME_TYPE.key(), amqpPropertyValue.toString()); } attributes.put(propertyName, amqpPropertyValue.toString()); } } } } flowFile = processSession.putAllAttributes(flowFile, attributes); } catch (Exception e) { logger.warn("Failed to update FlowFile with AMQP attributes", e); } } return flowFile; }
AMQPUtils { public static FlowFile updateFlowFileAttributesWithAmqpProperties(BasicProperties amqpProperties, FlowFile flowFile, ProcessSession processSession) { if (amqpProperties != null) { try { Method[] methods = BasicProperties.class.getDeclaredMethods(); Map<String, String> attributes = new HashMap<>(); for (Method method : methods) { if (Modifier.isPublic(method.getModifiers()) && method.getName().startsWith("get")) { Object amqpPropertyValue = method.invoke(amqpProperties); if (amqpPropertyValue != null) { String propertyName = extractPropertyNameFromMethod(method); if (isValidAmqpPropertyName(propertyName)) { if (propertyName.equals(PropertyNames.CONTENT_TYPE.getValue())) { attributes.put(CoreAttributes.MIME_TYPE.key(), amqpPropertyValue.toString()); } attributes.put(propertyName, amqpPropertyValue.toString()); } } } } flowFile = processSession.putAllAttributes(flowFile, attributes); } catch (Exception e) { logger.warn("Failed to update FlowFile with AMQP attributes", e); } } return flowFile; } }
AMQPUtils { public static FlowFile updateFlowFileAttributesWithAmqpProperties(BasicProperties amqpProperties, FlowFile flowFile, ProcessSession processSession) { if (amqpProperties != null) { try { Method[] methods = BasicProperties.class.getDeclaredMethods(); Map<String, String> attributes = new HashMap<>(); for (Method method : methods) { if (Modifier.isPublic(method.getModifiers()) && method.getName().startsWith("get")) { Object amqpPropertyValue = method.invoke(amqpProperties); if (amqpPropertyValue != null) { String propertyName = extractPropertyNameFromMethod(method); if (isValidAmqpPropertyName(propertyName)) { if (propertyName.equals(PropertyNames.CONTENT_TYPE.getValue())) { attributes.put(CoreAttributes.MIME_TYPE.key(), amqpPropertyValue.toString()); } attributes.put(propertyName, amqpPropertyValue.toString()); } } } } flowFile = processSession.putAllAttributes(flowFile, attributes); } catch (Exception e) { logger.warn("Failed to update FlowFile with AMQP attributes", e); } } return flowFile; }  }
AMQPUtils { public static FlowFile updateFlowFileAttributesWithAmqpProperties(BasicProperties amqpProperties, FlowFile flowFile, ProcessSession processSession) { if (amqpProperties != null) { try { Method[] methods = BasicProperties.class.getDeclaredMethods(); Map<String, String> attributes = new HashMap<>(); for (Method method : methods) { if (Modifier.isPublic(method.getModifiers()) && method.getName().startsWith("get")) { Object amqpPropertyValue = method.invoke(amqpProperties); if (amqpPropertyValue != null) { String propertyName = extractPropertyNameFromMethod(method); if (isValidAmqpPropertyName(propertyName)) { if (propertyName.equals(PropertyNames.CONTENT_TYPE.getValue())) { attributes.put(CoreAttributes.MIME_TYPE.key(), amqpPropertyValue.toString()); } attributes.put(propertyName, amqpPropertyValue.toString()); } } } } flowFile = processSession.putAllAttributes(flowFile, attributes); } catch (Exception e) { logger.warn("Failed to update FlowFile with AMQP attributes", e); } } return flowFile; }  static FlowFile updateFlowFileAttributesWithAmqpProperties(BasicProperties amqpProperties, FlowFile flowFile, ProcessSession processSession); static boolean isValidAmqpPropertyName(String name); static Map<String, Object> validateAMQPHeaderProperty(String amqpPropValue); static Integer validateAMQPDeliveryModeProperty(String amqpPropValue); static Integer validateAMQPPriorityProperty(String amqpPropValue); static Date validateAMQPTimestampProperty(String amqpPropValue); }
AMQPUtils { public static FlowFile updateFlowFileAttributesWithAmqpProperties(BasicProperties amqpProperties, FlowFile flowFile, ProcessSession processSession) { if (amqpProperties != null) { try { Method[] methods = BasicProperties.class.getDeclaredMethods(); Map<String, String> attributes = new HashMap<>(); for (Method method : methods) { if (Modifier.isPublic(method.getModifiers()) && method.getName().startsWith("get")) { Object amqpPropertyValue = method.invoke(amqpProperties); if (amqpPropertyValue != null) { String propertyName = extractPropertyNameFromMethod(method); if (isValidAmqpPropertyName(propertyName)) { if (propertyName.equals(PropertyNames.CONTENT_TYPE.getValue())) { attributes.put(CoreAttributes.MIME_TYPE.key(), amqpPropertyValue.toString()); } attributes.put(propertyName, amqpPropertyValue.toString()); } } } } flowFile = processSession.putAllAttributes(flowFile, attributes); } catch (Exception e) { logger.warn("Failed to update FlowFile with AMQP attributes", e); } } return flowFile; }  static FlowFile updateFlowFileAttributesWithAmqpProperties(BasicProperties amqpProperties, FlowFile flowFile, ProcessSession processSession); static boolean isValidAmqpPropertyName(String name); static Map<String, Object> validateAMQPHeaderProperty(String amqpPropValue); static Integer validateAMQPDeliveryModeProperty(String amqpPropValue); static Integer validateAMQPPriorityProperty(String amqpPropValue); static Date validateAMQPTimestampProperty(String amqpPropValue); final static String AMQP_PROP_DELIMITER; final static String AMQP_PROP_PREFIX; }