PublishAMQP 2.0.0

Bundle
org.apache.nifi | nifi-amqp-nar
Description
Creates an AMQP Message from the contents of a FlowFile and sends the message to an AMQP Exchange. In a typical AMQP exchange model, the message that is sent to the AMQP Exchange will be routed based on the 'Routing Key' to its final destination in the queue (the binding). If due to some misconfiguration the binding between the Exchange, Routing Key and Queue is not set up, the message will have no final destination and will return (i.e., the data will not make it to the queue). If that happens you will see a log in both app-log and bulletin stating to that effect, and the FlowFile will be routed to the 'failure' relationship.
Tags
amqp, message, publish, put, rabbit, send
Input Requirement
REQUIRED
Supports Sensitive Dynamic Properties
false
  • Additional Details for PublishAMQP 2.0.0

    PublishAMQP

    Summary

    This processor publishes the contents of the incoming FlowFile to an AMQP-based messaging system. At the time of writing this document the supported AMQP protocol version is v0.9.1.

    The component is based on RabbitMQ Client API The following guide and tutorial may also help you to brush up on some of the AMQP basics.

    This processor does two things. It constructs AMQP Message by extracting FlowFile contents (both body and attributes). Once message is constructed it is sent to an AMQP Exchange. AMQP Properties will be extracted from the FlowFile and converted to com.rabbitmq.client.AMQP.BasicProperties to be sent along with the message. Upon success the incoming FlowFile is transferred to success Relationship and upon failure FlowFile is penalized and transferred to the failure Relationship.

    Where did my message go?

    In a typical AMQP exchange model, the message that is sent to an AMQP Exchange will be routed based on the Routing Key to its final destination in the Queue. It’s called Binding. If due to some misconfiguration the binding between the Exchange, Routing Key and the Queue is not set up, the message will have no final destination and will return ( i.e., the data will not make it to the queue). If that happens you will see a log in both app-log and bulletin stating to that effect. Fixing the binding (normally done by AMQP administrator) will resolve the issue.

    AMQP Properties

    Attributes extracted from the FlowFile are considered candidates for AMQP properties if their names are prefixed with amqp$ (e.g., amqp$contentType=text/xml). To enrich message with additional AMQP properties you may use * UpdateAttribute* processor between the source processor and PublishAMQP processor. The following is the list of available standard AMQP properties: (“amqp$contentType”, “amqp$contentEncoding”, " amqp$headers" (if ‘Headers Source’ is set to ‘Attribute “amqp$headers” Value’) , “amqp$deliveryMode”, “amqp$priority”, “amqp$correlationId”, “amqp$replyTo”, “amqp$expiration”, “amqp$messageId”, “amqp$timestamp”, “amqp$type”, “amqp$userId”, “amqp$appId”, “amqp$clusterId”)

    AMQP Message Headers Source

    The headers attached to AMQP message by the processor depends on the “Headers Source” property value.

    1. Attribute “amqp$headers” Value - The processor will read single attribute “amqp$headers” and split it based on " Header Separator" and then read headers in key=value format.
    2. Attributes Matching Regex - The processor will pick flow file attributes by matching the regex provided in " Attributes To Headers Regular Expression". The name of the attribute is used as key of header

    Configuration Details

    At the time of writing this document it only defines the essential configuration properties which are suitable for most cases. Other properties will be defined later as this component progresses. Configuring PublishAMQP:

    1. Exchange Name - [OPTIONAL] the name of AMQP exchange the messages will be sent to. Usually provided by the administrator (e.g., ‘amq.direct’) It is an optional property. If kept empty the messages will be sent to a default AMQP exchange. More on AMQP Exchanges could be found here.
    2. Routing Key - [REQUIRED] the name of the routing key that will be used by AMQP to route messages from the exchange to destination queue(s). Usually provided by administrator (e.g., ‘myKey’) In the event when messages are sent to a default exchange this property corresponds to a destination queue name, otherwise a binding from the Exchange to a Queue via Routing Key must be set (usually by the AMQP administrator). More on AMQP Exchanges and Bindings could be found here.
    3. Host Name - [REQUIRED] the name of the host where AMQP broker is running. Usually provided by administrator ( e.g., ‘myhost.com’). Defaults to ’localhost’.
    4. Port - [REQUIRED] the port number where AMQP broker is running. Usually provided by the administrator (e.g., ' 2453’). Defaults to ‘5672’.
    5. User Name - [REQUIRED] user name to connect to AMQP broker. Usually provided by the administrator (e.g., ‘me’). Defaults to ‘guest’.
    6. Password - [REQUIRED] password to use with user name to connect to AMQP broker. Usually provided by the administrator. Defaults to ‘guest’.
    7. Use Certificate Authentication - [OPTIONAL] Use the SSL certificate common name for authentication rather than user name/password. This can only be used in conjunction with SSL. Defaults to ‘false’.
    8. Virtual Host - [OPTIONAL] Virtual Host name which segregates AMQP system for enhanced security. Please refer to this blog for more details on Virtual Host.
Properties
System Resource Considerations
Resource Description
MEMORY An instance of this component can cause high usage of this system resource. Multiple instances or high concurrency settings may result a degradation of performance.
Relationships
Name Description
failure All FlowFiles that cannot be routed to the AMQP destination are routed to this relationship
success All FlowFiles that are sent to the AMQP destination are routed to this relationship
Reads Attributes
Name Description
amqp$appId The App ID field to set on the AMQP Message
amqp$contentEncoding The Content Encoding to set on the AMQP Message
amqp$contentType The Content Type to set on the AMQP Message
amqp$headers The headers to set on the AMQP Message, if 'Header Source' is set to use it. See additional details of the processor.
amqp$deliveryMode The numeric indicator for the Message's Delivery Mode
amqp$priority The Message priority
amqp$correlationId The Message's Correlation ID
amqp$replyTo The value of the Message's Reply-To field
amqp$expiration The Message Expiration
amqp$messageId The unique ID of the Message
amqp$timestamp The timestamp of the Message, as the number of milliseconds since epoch
amqp$type The type of message
amqp$userId The ID of the user
amqp$clusterId The ID of the AMQP Cluster