Using Predix Message Queue

<dependency>
  <groupId>com.rabbitmq</groupId>
  <artifactId>amqp-client</artifactId>
  <version>4.2.0</version>
</dependency>

// Parse the RabbitMQ config from the VCAP_SERVICES environment. Assumes there's only one RabbitMQ instance bound.
String vcapServices = System.getenv("VCAP_SERVICES");
JSONObject vcapJson = new JSONObject(vcapServices);
JSONObject amqpCredentials =  vcapJson.getJSONArray("predix-message-queue") //
        .getJSONObject(0) //
        .getJSONObject("credentials")
        .getJSONObject("protocols")
        .getJSONObject("amqp");

/* Set up the RabbitMQ connection. Alternatively you can also use the URI:
*
*   factory.setUri(amqpCredentials.getString("uri"));
*/
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(amqpCredentials.getString("host"));
factory.setUsername(amqpCredentials.getString("username"));
factory.setPassword(amqpCredentials.getString("password"));
factory.setPort(amqpCredentials.getInt("port"));
factory.setVirtualHost(amqpCredentials.getString("vhost"));
Connection connection = factory.newConnection();

// Define Queue name, Exchange name, and routing key as strings.
String QUEUE_NAME = "PredixRabbitMQSample";
String EXCHANGE_NAME = "sample-exchange";
String ROUTING_KEY = "test";

// Create the channel, declare the queue and exchange, and bind them.
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "direct", true);
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);

// Generate a random UUID to use for the message
String message =  UUID.randomUUID().toString();
byte[] messageBodyBytes = message.getBytes();

// Publish the message using basicPublish
channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, null, messageBodyBytes);

Consumer consumer = new DefaultConsumer(channel) {
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties basicProperties,
                               byte[] body) throws IOException {
        String message = new String(body, "UTF-8");
        String output = String.format("Received message from exchange %s: %s", envelope.getExchange(), message);
        System.out.println(output);

    }
};
channel.basicConsume(queueName, true, consumer);

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-dependencies</artifactId>
            <version>Dalston.SR1</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
        <version>1.5.4.RELEASE</version>
    </dependency>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-bus-amqp</artifactId>
        <version>1.3.1.RELEASE</version>
    </dependency>
</dependencies>

@Configuration
public class RabbitMQSpringSampleConsumerConfig {

    @Bean
    public Queue queue() {
        return new Queue("PredixRabbitMQSpringSample");
    }

    @Bean
    public DirectExchange exchange() {
        return new DirectExchange("sample-exchange");
    }

    @Bean
    public Binding binding(DirectExchange exchange, Queue queue) {
        return BindingBuilder.bind(queue).to(exchange).with("test");
    }
}

@Autowired
private DirectExchange exchange;

@Autowired
private RabbitTemplate template;

// Generate a random UUID to use for the message
String message = UUID.randomUUID().toString();

// Publish the message to the configured exchange, using 'test' as the routing key.
template.convertAndSend(exchange.getName(), "test", message);

@RabbitListener(queues="#{queue.name}")
public void receiveMessage(String data) throws InterruptedException {
    String exchangeName = exchange.getName();
    System.out.println(String.format("Received message from exchange %s: %s", exchangeName, data));
}

source 'https://rubygems.org'

ruby '>=2.3.1'
gem 'bunny', '>=2.6.4'

vcap = JSON.parse ENV['VCAP_SERVICES']
amqp_credentials = vcap['predix-message-queue'][0]['credentials']['protocols']['amqp']

credentials = CF::App::Credentials.find_by_service_label('predix-message-queue')
amqp_credentials = credentials['protocols']['amqp']

# Create the RabbitMQ connection. You can also use the URI:
#
#   conn = Bunny.new(amqp_credentials['uri'])
#
conn = Bunny.new(host: amqp_credentials['host'],
                 port: amqp_credentials['port'],
                 username: amqp_credentials['username'],
                 password: amqp_credentials['password'],
                 vhost: amqp_credentials['vhost'] )
conn.start

queue.subscribe(block: true, exclusive: true) do |delivery_info, properties, message|
    STDOUT.puts "Received message from the queue: #{message}"
  end

channel.close
conn.close

• Celery
• Haigha

Install the Library

To ensure that CloudFoundry will install the pika module add a requirements.txt to your app directory that contains the following line:

pika>=0.10.0

Discover the Bound Predix Message Queue Service in Python

You can use os.environ and the json library to parse the VCAP_SERVICES environment variable. If the variable isn't defined, or if the service isn't found, a KeyError will be thrown.

import sys
import json
from os import environ

try:
    vcap_services = json.loads(environ['VCAP_SERVICES'])
    amqp_credentials = vcap_services['predix-message-queue'][0]['credentials']['protocols']['amqp']
except KeyError as error:
    print("Could not parse credentials from the environment. Key not found: {0}".format(error))
    sys.exit(1)

You can also use the cfenv library to parse the app's credentials:

from cfenv import AppEnv

app_env = AppEnv()
rabbit_service = app_env.get_service(label='predix-message-queue')
amqp_credentials = rabbit_service.credentials['protocols']['amqp']

Set Up the Ra bbitMQ Connection

Use the PlainCredentials and ConnectionParameters classes to set up the connection configuration using the service credentials, then use them to open the connection and create a channel.

# Configure the authentication credentials
rabbit_creds = pika.PlainCredentials(amqp_credentials['username'], amqp_credentials['password'])

# Set up connection parameters
params = pika.ConnectionParameters(host=amqp_credentials['host'],
                                   port=amqp_credentials['port'],
                                   virtual_host=amqp_credentials['vhost'],
                                   ssl=amqp_credentials['ssl'],
                                   credentials=rabbit_creds)

# Open the connection and create a channel
connection = pika.BlockingConnection(params)
channel = connection.channel()

Publishing Messages

Use the basic_publish method of the Channel to publish a message. This example uses direct routing with a routing key:

channel.basic_publish(exchange='sample-exchange', routing_key='test', body='Hello World!')

Consuming Messages

To consume messages first declare a queue and an exchange, and then bind them with a routing key:

To consume messages first declare a queue and an exchange, and then bind them with a routing key:

Then define a callback method to handle incoming messages, and pass it to the basic_consume method of the Channel:

# Define a callback method to use for handling messages
def callback(ch, method, properties, body):
    print("Received message with routing key {0}: {1}".format(method.routing_key, body))

# Set up the consumer and begin listening for messages.
channel.basic_consume(callback, queue=queue_name, no_ack=True)
channel.start_consuming()

Full Sample Producer App

Here's the full example publisher which will publish a random UUID to the exchange once every five seconds for five minutes:

requirements.txt

pika>=0.10.0

manifest.yml

applications:
- name: python-sample-producer
  command: python sample-producer.py
  no-route: true
  mem: 256M
  disk: 1024M
  instances: 1

Procfile

web: python sample-producer.py

sample-producer.py

import sys
import json
import pika
from os import environ
from uuid import uuid4
from time import sleep

# Parse the credentials from the VCAP_SERVICES environment variable
try:
    vcap_services = json.loads(environ['VCAP_SERVICES'])
    amqp_credentials = vcap_services['predix-message-queue'][0]['credentials']['protocols']['amqp']

    # Configure the authentication credentials
    rabbit_creds = pika.PlainCredentials(amqp_credentials['username'], amqp_credentials['password'])

    # Set up connection parameters
    params = pika.ConnectionParameters(host=amqp_credentials['host'],
                                       port=amqp_credentials['port'],
                                       virtual_host=amqp_credentials['vhost'],
                                       ssl=amqp_credentials['ssl'],
                                       credentials=rabbit_creds)

except KeyError as error:
    print("Could not parse credentials from the environment. Key not found: {0}".format(error))
    sys.exit(1)

# Open the connection and create a channel
connection = pika.BlockingConnection(params)
channel = connection.channel()

# Publish a message to the queue every 5 seconds for 5 minutes.
max_seconds = 300
total_seconds = 0
while total_seconds < max_seconds:
    # Generate a random UUID string to use as the message.
    message = str(uuid4())
    # Publish the message to the channel using basic_publish with an exchange and routing key
    channel.basic_publish(exchange='sample-exchange', routing_key='test', body=message)
    print("Published message to exchange sample-exchange: {0}".format(message))
    sleep(1)
    total_seconds += 1

connection.close()
sys.exit(0)

Sample log Output:

2017-08-04T10:48:09.28-0700 [DEA/5] OUT Starting app instance (index 0) with guid 0b803338-8c3a-4620-b6c7-73bdc2de9efa
2017-08-04T10:48:11.83-0700 [App/0] OUT Published message to exchange sample-exchange: 5769f6a3-2587-4202-b5c4-050aaf8ea16f
2017-08-04T10:48:12.83-0700 [App/0] OUT Published message to exchange sample-exchange: e83743d8-9d04-4c44-8f88-b3216b890b8d
2017-08-04T10:48:13.83-0700 [App/0] OUT Published message to exchange sample-exchange: 655b684a-8aab-41c3-8370-5dfe9ea8b64a
2017-08-04T10:48:14.83-0700 [App/0] OUT Published message to exchange sample-exchange: 19bbd3b9-dfc9-4c68-b854-ef68f2f82440

Full Sample Consumer App

This sample consumer app will retrieve the messages published by the above producer app and print them to the app's logs. It also demonstrates using the cfenv library to parse the service details from the environment.

requirements.txt

pika>=0.10.0

manifest.yml

applications:
- name: python-sample-consumer
  command: python sample-consumer.py
  no-route: true
  mem: 256M
  disk: 1024M
  instances: 1

Procfile

web: python sample-consumer.py

sample-producer.py

import sys
import json
import pika
from cfenv import AppEnv

app_env = AppEnv()
rabbit_service = app_env.get_service(label='predix-rabbitmq')
amqp_credentials = rabbit_service.credentials['protocols']['amqp']

# Configure the authentication credentials
rabbit_creds = pika.PlainCredentials(amqp_credentials['username'], amqp_credentials['password'])

# Set up connection parameters
params = pika.ConnectionParameters(host=amqp_credentials['host'],
                                   port=amqp_credentials['port'],
                                   virtual_host=amqp_credentials['vhost'],
                                   ssl=amqp_credentials['ssl'],
                                   credentials=rabbit_creds,
                                   heartbeat_interval=30)


# Open the connection and create a channel
connection = pika.BlockingConnection(params)
channel = connection.channel()

# Declare the exchange and the queue. Bind the queue to the exchange with a routing key.
queue_name = 'predix-message-queue-python-sample'
channel.exchange_declare(exchange='sample-exchange', type='direct')
channel.queue_declare(queue=queue_name, exclusive=False)
channel.queue_bind(exchange='sample-exchange', queue=queue_name, routing_key='test')


# Define a callback method to use for handling messages
def callback(ch, method, properties, body):
    print("Received message with routing key {0}: {1}".format(method.routing_key, body))

print("Listening for messages on queue {0}".format(queue_name))
# Set up the consumer and begin listening for messages.
channel.basic_consume(callback, queue=queue_name, no_ack=True)
channel.start_consuming()

Sample Log Output

2017-08-04T10:47:10.97-0700 [DEA/10] OUT Starting app instance (index 0) with guid f37adf39-15f0-45cb-9028-b9ebe9981cd3
2017-08-04T10:47:13.46-0700 [App/0] OUT Listening for messages on queue predix-messagae-queue-python-sample
2017-08-04T10:48:11.58-0700 [App/0] OUT Received message with routing key test: 5769f6a3-2587-4202-b5c4-050aaf8ea16f
2017-08-04T10:48:12.59-0700 [App/0] OUT Received message with routing key test: e83743d8-9d04-4c44-8f88-b3216b890b8d
2017-08-04T10:48:13.59-0700 [App/0] OUT Received message with routing key test: 655b684a-8aab-41c3-8370-5dfe9ea8b64a
2017-08-04T10:48:14.59-0700 [App/0] OUT Received message with routing key test: 19bbd3b9-dfc9-4c68-b854-ef68f2f82440

Install the Library

To ensure that CloudFoundry will install php-amqplib when the app is uploaded add the following entry to the composer.json file:

{
  "require": {
    "php-amqplib/php-amqplib": ">=2.6.1"
  }
}

Make sure to include the autoload.php file from the vendor directory in order to include use the library:

<?php
require_once realpath(__DIR__ . DIRECTORY_SEPARATOR . '../lib/vendor'). '/autoload.php';
?>

Discover the Bound Predix RabbitMQ Service in PHP