RabbitMQ

Springs RabbitMQ Camel Producer (and Consumer)

Note

Connection details are presented via a named spring bean which must be registered before a route is executed

For more information about Spring RabbitMQ: https://camel.apache.org/components/4.22.x/spring-rabbitmq-component.html.

Note

Since P6 Core 6.10.26 a queue created by autoDeclare is durable, and the request/reply replyTimeout defaults to 30 seconds instead of 5. Add durable=false or replyTimeout=5000 to the endpoint URI to keep the previous behaviour.

Example

Template: Dev_BaseRoute.groovy

${addRoutes} :=

def cf = new org.springframework.amqp.rabbit.connection.CachingConnectionFactory()
cf.setHost('localhost')
cf.setPort(5672)
cf.setUsername('guest')
cf.setPassword('notapassword')

p6.camel.registerBean('rabbitConnectionFactory', cf)

p6.camel.getCtx().addRoutes(new RouteBuilder() {
    void configure() {
        from('direct:mqevents')
                .to('spring-rabbitmq:platform6?connectionFactory=#rabbitConnectionFactory&routingKey=platform6.events.out')
                .routeId('RabbitMQEventsOut')
    }
})

${destroyRoutes} :=

p6.camel.unregisterBean('rabbitConnectionFactory')
p6.camel.destroyRoute('RabbitMQEventsOut')

The above example will send the byte[] body of any Exchange sent to direct:mqevents to the platform6 exchange with a routingKey of platform6.events.out

To test the route simply send a test Exchange via the p6.camel DSL

def exchange = [
  body: '''
  {  
   "EventType": "statusUpdate",  
   "EventId": "97347AD890E8766",
   "EventDetail" {
      "id": "INVOICE_876856_9785ADE87644E987A",
      "newStatus": "Acknowledged",
      "oldStatus": "Sent"   
   }
  }
  '''
]
p6.camel.endpointExchangeWaitInput('direct:mqevents', exchange)

Secure connections (TLS)

Since P6 Core 6.10.26 the bundled RabbitMQ Java client (5.35) validates the broker certificate and checks that it matches the broker host name whenever TLS is enabled. Earlier versions trusted any certificate and did not verify the host name, so a route that used to connect to a broker with a self-signed certificate may now fail with an SSLHandshakeException after the upgrade.

The recommended setup is a broker certificate that is either issued by a CA known to the JVM or added to a trust store, and that carries the broker host name in its subject alternative names. With the JVM trust store nothing else is needed:

def cf = new org.springframework.amqp.rabbit.connection.CachingConnectionFactory()
cf.setHost('rabbit.example.internal')
cf.setPort(5671)
cf.setUsername('guest')
cf.setPassword('notapassword')
cf.getRabbitConnectionFactory().useSslProtocol()   // JVM default trust store, host name verified

To trust a private CA, build an SSLContext on a trust store that contains it and pass it to the client:

def trustStore = java.security.KeyStore.getInstance('PKCS12')
new File('/opt/p6core.data/resources/certificates/rabbit-truststore.p12').withInputStream { trustStore.load(it, 'changeit'.toCharArray()) }
def tmf = javax.net.ssl.TrustManagerFactory.getInstance(javax.net.ssl.TrustManagerFactory.getDefaultAlgorithm())
tmf.init(trustStore)
def sslContext = javax.net.ssl.SSLContext.getInstance('TLS')
sslContext.init(null, tmf.getTrustManagers(), null)
cf.getRabbitConnectionFactory().useSslProtocol(sslContext)

Self-signed certificates

For non-production brokers, cf.getRabbitConnectionFactory().useTlsWithNoVerification() accepts any certificate chain. It does not switch off the host-name check: the certificate must still name the broker host in its subject alternative names, otherwise the handshake fails with No subject alternative DNS name matching ... found. Skipping both checks requires an SSLContext built on a custom javax.net.ssl.X509ExtendedTrustManager, which should never reach production.

Note

Spring AMQP’s RabbitConnectionFactoryBean.setSkipServerCertificateValidation(true) relies on the client’s former trust-all behaviour and no longer skips the validation with this client version.