This is the final post in my series about building a service relay for Dynamics 365 CE with RabbitMQ and Python. In my previous post in this series, I showed the Python code to make the service relay work. In today's post, I will show how you can use Azure Functions to make a consumer service proxy using C# so client applications don't have to access to your RabbitMQ broker directly, and I will also discuss some general thoughts on security and scalability for this service relay architecture.

Although this simple service relay allows external consumers to get data from Dynamics 365 CE without needing to connect directly, the examples I've shown so far require that they can connect to a RabbitMQ broker. This may be problematic for a variety of reasons, so you would probably want external consumers to connect to a web service proxy that would write requests to and read responses from the RabbitMQ broker.

Building a service proxy function

You can build an Azure Functions service proxy with Python, but I don't recommend it for three reasons:

1. Azure Functions Python support is still considered experimental.
2. Python scripts that use external libraries can run exceedingly slow.
3. Getting the environment set up is a bit of a hassle.

On the other hand, building a service proxy function with C# was so much easier, and it performed much better than a comparable Python function (~.5 seconds for C# compared to 5+ seconds for Python).

Here are the steps I took to build my C# service proxy function:

1. Create a C# HTTP trigger function.
2. Create and upload a project.json file with a dependency on the RabbitMQ client (see below).
3. Take the "RpcClient" class from the RabbitMQ .Net RPC tutorial and call it from within my function.

Here's my project.json file:

{
  "frameworks": {
    "net46":{
      "dependencies": {
        "RabbitMQ.Client": "5.0.1"
      }
    }
   }
}

And here's my run.csx file:

using System.Net;
using System;
using System.Collections.Concurrent;
using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

public static async Task<HttpResponseMessage> Run(HttpRequestMessage req, TraceWriter log)
{
    log.Info("Processing request");

    // parse query parameter
    string query = req.GetQueryNameValuePairs()
        .FirstOrDefault(q => string.Compare(q.Key, "query", true) == 0)
        .Value;

    // Get request body
    dynamic data = await req.Content.ReadAsAsync<object>();

    // Set name to query string or body data
    query = query ?? data?.query;

    var rpcClient = new RpcClient();
    
    log.Info(string.Format(" [.] query start time {0}", DateTime.Now.ToString("MM/dd/yyyy hh:mm:ss.fff tt")));
    var response = rpcClient.Call(query);

    log.Info(string.Format(" [.] query end time {0}", DateTime.Now.ToString("MM/dd/yyyy hh:mm:ss.fff tt")));
    rpcClient.Close();

    return req.CreateResponse(HttpStatusCode.OK, response);
}

public class RpcClient
{
    private readonly IConnection connection;
    private readonly IModel channel;
    private readonly string replyQueueName;
    private readonly EventingBasicConsumer consumer;
    private readonly BlockingCollection<string> respQueue = new BlockingCollection<string>();
    private readonly IBasicProperties props;

    public RpcClient()
    {
        var factory = new ConnectionFactory() { HostName = "RABBITHOST", UserName="RABBITUSER", Password="RABBITUSERPASS"  };

        connection = factory.CreateConnection();
        channel = connection.CreateModel();
        replyQueueName = channel.QueueDeclare().QueueName;
        consumer = new EventingBasicConsumer(channel);

        props = channel.CreateBasicProperties();
        var correlationId = Guid.NewGuid().ToString();
        props.CorrelationId = correlationId;
        props.ReplyTo = replyQueueName;

        consumer.Received += (model, ea) =>
        {
            var body = ea.Body;
            var response = Encoding.UTF8.GetString(body);
            if (ea.BasicProperties.CorrelationId == correlationId)
            {
                respQueue.Add(response);
            }
        };
    }

    public string Call(string message)
    {
        var messageBytes = Encoding.UTF8.GetBytes(message);
        channel.BasicPublish(
            exchange: "",
            routingKey: "rpc_queue",
            basicProperties: props,
            body: messageBytes);

        channel.BasicConsume(
            consumer: consumer,
            queue: replyQueueName,
            autoAck: true);

        return respQueue.Take(); ;
    }

    public void Close()
    {
        connection.Close();
    }
}

Here's a screenshot showing me calling the C# function with Postman.

Because I did actually build a Python function, I will go ahead and share how I did it if you're interested. Here are the steps I took:

1. Create a Python HTTP trigger function.
2. Install Python 3.6 via site extensions (see steps 2.1-2.4 here).
3. Install the necessary libraries using pip via KUDU.

Here's the Python function code:

import os
import sys
import json
import pika
import uuid
import datetime

class CrmRpcClient(object):
    def __init__(self):
        #RabbitMQ connection details
        self.rabbituser = 'RABBITUSERNAME'
        self.rabbitpass = 'RABBITUSERPASS'
        self.rabbithost = 'RABBITHOST' 
        self.rabbitport = 5672
        self.rabbitqueue = 'rpc_queue'
        rabbitcredentials = pika.PlainCredentials(self.rabbituser, self.rabbitpass)
        rabbitparameters = pika.ConnectionParameters(host=self.rabbithost,
                                    port=self.rabbitport,
                                    virtual_host='/',
                                    credentials=rabbitcredentials)

        self.rabbitconn = pika.BlockingConnection(rabbitparameters)

        self.channel = self.rabbitconn.channel()

        #create an anonymous exclusive callback queue
        result = self.channel.queue_declare(exclusive=True)
        self.callback_queue = result.method.queue

        self.channel.basic_consume(self.on_response, no_ack=True,
                                   queue=self.callback_queue)

    #callback method for when a response is received - note the check for correlation id
    def on_response(self, ch, method, props, body):
        if self.corr_id == props.correlation_id:
            self.response = body

    #method to make the initial request
    def call(self, n):
        self.response = None
        #generate a new correlation id
        self.corr_id = str(uuid.uuid4())

        #publish the message to the rpc_queue - note the reply_to property is set to the callback queue from above
        self.channel.basic_publish(exchange='',
                                   routing_key=self.rabbitqueue,
                                   properties=pika.BasicProperties(
                                         reply_to = self.callback_queue,
                                         correlation_id = self.corr_id,
                                         ),
                                   body=n)
        while self.response is None:
            self.rabbitconn.process_data_events()
        return self.response

print(" [.] query start time %r" % str(datetime.datetime.now()))
#instantiate an rpc client
crm_rpc = CrmRpcClient()

postreqdata = json.loads(open(os.environ['req']).read())
query = postreqdata['query']

crm_rpc = CrmRpcClient()
print(" [.] query start time %r" % str(datetime.datetime.now()))
queryresponse = crm_rpc.call(query)
print(" [.] query end time %r" % str(datetime.datetime.now()))
response = open(os.environ['res'], 'w')
response.write(queryresponse.decode())
response.close()

Here's a screenshot showing me calling the Python function with Postman.

Note the difference in time between the two functions - 5.62 seconds for Python and .46 seconds for C#!

Security and scalability

If you decide to use this approach in production, I'd suggest you carefully consider both security and scalability. Obviously the overall solution will only be as secure as your RabbitMQ broker and communications between the broker and its clients, so you'll want to look at best practices for access control and securing the communications with TLS. Here are some links for further reading on those subjects:

• TLS - https://www.rabbitmq.com/ssl.html
• Access control - https://www.rabbitmq.com/access-control.html

As for scalability, the approach I've shown creates a separate response queue for each consumer, but it can have problems scaling, especially if you are using a RabbitMQ cluster. You may want to look at the "direct reply-to" approach instead. For an interesting real-world overview of using direct reply-to, take a look at this blog post..

Wrapping up

I hope you've enjoyed this series and that it has given you some ideas about how to implement service relays in your Dynamics 365 CE projects. As I worked through the examples, I certainly learned a few new things, especially when I created my Python service proxy in Azure Functions.

Here are links to all the previous posts in this series.

1. Part 1 - Series introduction
2. Part 2 - Solution prerequisites
3. Part 3 - Python code for the consumer and listener processes

What do you think about this approach? Is it something you think you'd use in production? Let us know in the comments!

In my last post in this series, I walked through the prerequisites for building a simple service relay for Dynamics 365 CE with RabbitMQ and Python. In today's post I will show the Python code to make the service relay work.

As I described in the first post in this series, this approach relies on a consumer process and a queue listener process that can both access a RabbitMQ message broker.

A consumer writes a request to a cloud-hosted RabbitMQ request queue (either directly or through a proxy service) and starts waiting for a response. On the other end, a Python script monitors the request queue for inbound requests. When it sees a new one, it executes the appropriate request through the Dynamics 365 Web API and writes the response back to a client-specific RabbitMQ response queue. The consumer then picks up the response from the queue.

This solution is based on the remote procedure call (RPC) approach shown here. The main difference is that I have added logic to the queue monitoring script to query the Dynamics 365 Web API based on the inbound request from the consumer.

Consumer sample

The consumer does the following:

1. Read the text of the request to write to the queue from a command-line argument.
2. Establish a connection to the RabbitMQ broker.
3. Create a new anonymous, exclusive callback queue.
4. Write a request message a queue called "rpc_queue." This message will include the callback queue as its "reply_to" property.
5. Monitor the callback queue for a response.

There's no validation in this sample, so if you run it without a command-line argument, it will just throw an error and exit.

import sys
import pika
import uuid
import datetime

class CrmRpcClient(object):
    def __init__(self):
        #RabbitMQ connection details
        self.rabbituser = 'crmuser'
        self.rabbitpass = 'crmpass'
        self.rabbithost = '127.0.0.1' 
        self.rabbitport = 5672
        self.rabbitqueue = 'rpc_queue'
        rabbitcredentials = pika.PlainCredentials(self.rabbituser, self.rabbitpass)
        rabbitparameters = pika.ConnectionParameters(host=self.rabbithost,
                                    port=self.rabbitport,
                                    virtual_host='/',
                                    credentials=rabbitcredentials)

                self.rabbitconn = pika.BlockingConnection(rabbitparameters)

        self.channel = self.rabbitconn.channel()

        #create an anonymous exclusive callback queue
        result = self.channel.queue_declare(exclusive=True)
        self.callback_queue = result.method.queue

        self.channel.basic_consume(self.on_response, no_ack=True,
                                   queue=self.callback_queue)

    #callback method for when a response is received - note the check for correlation id
    def on_response(self, ch, method, props, body):
        if self.corr_id == props.correlation_id:
            self.response = body

    #method to make the initial request
    def call(self, n):
        self.response = None
        #generate a new correlation id
        self.corr_id = str(uuid.uuid4())

        #publish the message to the rpc_queue - note the reply_to property is set to the callback queue from above
        self.channel.basic_publish(exchange='',
                                   routing_key=self.rabbitqueue,
                                   properties=pika.BasicProperties(
                                         reply_to = self.callback_queue,
                                         correlation_id = self.corr_id,
                                         ),
                                   body=n)
        while self.response is None:
            self.rabbitconn.process_data_events()
        return self.response

#instantiate an rpc client
crm_rpc = CrmRpcClient()

#read the request from the command line
request = sys.argv[1]

#make the request and get the response
print(" [x] Requesting crm data("+request+")")
print(" [.] Start time %s" % str(datetime.datetime.now()))
response = crm_rpc.call(request)

#convert the response message body from the queue to a string 
decoderesponse = response.decode()

#print the output
print(" [.] Received response: %s" % decoderesponse)
print(" [.] End time %s" % str(datetime.datetime.now()))

Queue listener sample

The queue listener does the following:

1. Establish a connection to the RabbitMQ broker
2. Monitor "rpc_queue" queue.
3. When a new message from the "rpc_queue" queue is delivered, decode the message body as a string, and determine what Web API query to execute. Note: This sample can return a list of contacts or accounts from Dynamics 365 CE based on the request the consumer sends ("getcontacts" or "getaccounts"). If any other request is received, the listener will return an error message to the consumer callback queue.
4. Execute the appropriate query against the Dynamics 365 Web API and write the response to the callback queue the client established originally.

import pika
import requests
from requests_ntlm import HttpNtlmAuth
import json

#NTLM credentials to access on-prem Dynamics 365 Web API
username = 'DOMAIN\\USERNAME'
userpassword = 'PASSWORD'

#full path to Web API
crmwebapi = 'http://33.0.0.16/lucastest02/api/data/v8.1'

#RabbitMQ connection details
rabbituser = 'crmuser'
rabbitpass = 'crmpass'
rabbithost = '127.0.0.1' 
rabbitport = 5672

#method to execute a Web API query based on the client request
def processquery(query):
    #set the Web API request headers
    crmrequestheaders = {
        'OData-MaxVersion': '4.0',
        'OData-Version': '4.0',
        'Accept': 'application/json',
        'Content-Type': 'application/json; charset=utf-8',
        'Prefer': 'odata.maxpagesize=500',
        'Prefer': 'odata.include-annotations=OData.Community.Display.V1.FormattedValue'
    }

    #determine which Web API query to execute
    if query == 'getcontacts':
        crmwebapiquery = '/contacts?$select=fullname,contactid'
    elif query == 'getaccounts':
        crmwebapiquery = '/accounts?$select=name,accountid'
    else:
        #only handle 'getcontacts' or 'getaccounts' requests
        return 'Operation not supported'

    #execute the query
    crmres = requests.get(crmwebapi+crmwebapiquery, headers=crmrequestheaders,auth=HttpNtlmAuth(username,userpassword))
    
    #get the results json
    crmjson = crmres.json()

    #return the json
    return crmjson

#method to handle new inbound requests
def on_request(ch, method, props, body):
    #convert the message body from the queue to a string
    decodebody = body.decode('utf-8')

    #print the request
    print(" [.] Received request: '%s'" % decodebody)

    #process the request query
    response = processquery(decodebody)

    #publish the response back to 'reply-to' queue from the request message and set the correlation id
    ch.basic_publish(exchange='',
                     routing_key=props.reply_to,
                     properties=pika.BasicProperties(correlation_id = \
                                                         props.correlation_id),
                     body=str(response).encode(encoding="utf-8", errors="strict"))
    ch.basic_ack(delivery_tag = method.delivery_tag)

print(" [x] Awaiting RPC requests")

#connect to RabbitMQ broker
rabbitcredentials = pika.PlainCredentials(rabbituser, rabbitpass)
rabbitparameters = pika.ConnectionParameters(host=rabbithost,
                               port=rabbitport,
                               virtual_host='/',
                               credentials=rabbitcredentials)
rabbitconn = pika.BlockingConnection(rabbitparameters)
channel = rabbitconn.channel()

#declare the 'rpc_queue' queue
channel.queue_declare(queue='rpc_queue')

#set qos settings for the channel
channel.basic_qos(prefetch_count=1)

#assign the 'on_request' method as a callback for when new messages delivered from the 'rpc_queue' queue
channel.basic_consume(on_request, queue='rpc_queue')

#start listening for requests
channel.start_consuming()

Trying it out

As I mentioned in my last post, I initially wrote my code to use a RabbitMQ broker running on my local PC, so that's why the connections in the samples show 127.0.0.1 as the host. For a demo, I've spun up a copy of RabbitMQ in a Docker container in the cloud and updated my connection parameters accordingly, but I am still running my queue listener and consumer processes on my local PC.

When the listener first starts, it displays a simple status message.

Then I execute a "getcontacts" request from the consumer in a separate window.

From the timestamps before and after the request, you can see the round-trip time is less than .2 seconds, which includes two round trips between my local PC and the cloud-based RabbitMQ broker plus the actual query processing time in my local Dynamics 365 CE org.

Then I execute a "getaccounts" request.

This request was also fulfilled in less than .2 seconds.

Finally I execute an invalid request to show what the error response looks like.

You'll note the total time from request to response is only about .05 seconds less than the total time for the valid queries. That indicates most of the time used in these samples is being spent on the round trips between my local PC and the RabbitMQ broker, which is not surprising.

Meanwhile, the queue listener wrote a simple status update for every request it received. If I were using this in production, I would use more sophisticated logging.

Wrapping up

That's it for now. In my next (and final) post in this series, I will show how you can use Azure Functions to make a consumer service proxy so consuming applications don't have to access to your RabbitMQ broker directly, and I will also discuss some general thoughts on security and scalability for the service .

In my last post in this series, I outlined an approach for building a simple service relay with RabbitMQ and Python to easily expose an on-premises Dynamics 365 Customer Engagement organization to external consumers. In this post I will walk through the prerequisites for building this out. I'm assuming you have access to a Dynamics 365 CE organization, so I'm going to skip the setup for that and focus on just RabbitMQ and Python today.

Setting up RabbitMQ

Back in 2015 When I first blogged about RabbitMQ and Dynamics 365, I wrote a detailed post showing how to install and configure RabbitMQ. Since then I have discovered the joys of Docker, which makes the process significantly easier. If you have access to Docker, I highly recommend using it. Once you have Docker running, you can use one of the official RabbitMQ images. For this project, I initially used the rabbitmq:3-management image in Docker for Windows running on my local PC. After I got the basic functionality working, I then moved to an instance of Docker running in the cloud on a Digital Ocean VPS.

If don't want to use Docker, you can use a full RabbitMQ install like I showed previously. The main thing to remember is that no matter how you set up your RabbitMQ server, if it is not accessible from the public internet, you will not be able to use it as a service relay between an on-premises Dynamics 365 org and external consumers.

Setting up Python

I'm assuming if you've gotten this far, you have a functional Python development environment (if not, give Visual Studio Code a try), and the code I have written works in Python versions 2.7 or 3.x. In order to connect to both RabbitMQ and Dynamics 365, you will need a few additional packages. To connect to RabbitMQ, Pika is the RabbitMQ team's recommended Python client, and you can get it using pip.

To communicate with Dynamics 365, you'll need to use the Web API, but authentication will be handled differently depending on whether you connect to an on-premises org or an online / IFD org. For online or IFD orgs, you can either use ADAL or this alternate approach I described back in 2016. If you have an on-premises org, you can authenticate using the requests_ntlm package like I showed here. As with the Pika client, all the packages you need to connect to Dynamics 365 are also available via pip.

Wrapping up

That's it for today. In my next post in this series I will show the Python code you need to make this service relay work.

Integrating with external systems is a common requirement in Dynamics 365 Customer Engagement projects, but when the project involves an on-premises instance of Dynamics 365, routing requests from external systems through your firewall can present an additional challenge. Over the course of the next few posts, I will show you can easily build a simple service relay with RabbitMQ and Python to handle inbound requests from external data interface consumers.

Here's how my approach works. A consumer writes a request to a cloud-hosted RabbitMQ request queue (either directly or through a proxy service) and starts waiting for a response. On the other end, a Python script monitors the request queue for inbound requests. When it sees a new one, it executes the appropriate request through the Dynamics 365 Web API and writes the response back to a client-specific RabbitMQ response queue. The consumer then picks up the response from the queue. This way the consumer doesn't need to know anything other than how to write the initial request, and no extra inbound firewall ports need to be opened.

This diagram shows an overview of the process.

Although my original goal was to accelerate the deployment of data interfaces for on-premises Dynamics 365 CE instances, a simple service relay like this could also be useful for IFD or Dynamics 365 online deployments if you don't want to allow direct access to your organization. Because the queue monitoring process is single-threaded, it's an easy way to throttle requests, but you can run multiple instances of the queue monitor script if you want to increase the number of concurrent requests the relay can process.

Why use this approach?

There are lots message brokers and service bus offerings (Azure Service Bus, IBM MQ, Amazon SQS, etc.) you could use to build a service relay. In fact there's even an Azure offering called Azure Relay that aims to solve exactly the same problem that my approach does, but not just for Dynamics 365, so "why use this?" is a great question.

First, I think RabbitMQ is just a great tool, and I previously wrote a five-part series about using RabbitMQ with Dynamics 365 (back when it was still called Dynamics CRM). Second, using RabbitMQ instead of a cloud-specific service bus offering gives you maximum flexibility in where you host your request and response queues and how you chose to scale. For example, my RabbitMQ broker runs in a Docker container on a Digital Ocean VPS. If I ever decide to move off of Digital Ocean, I can easily switch to any IaaS or VPS provider. I can also configure a RabbitMQ cluster to achieve significantly faster throughput.

As for why I'm using Python instead of C#, which is probably more familiar to most Dynamics 365 developers, Python also makes this approach more flexible. Using Python means I'm not tied to the Dynamics 365 SDK client libraries or a Windows host for running my queue monitoring process, and I can easily package my monitoring process in a Docker image. (Although I highly recommend Python, there are RabbitMQ clients for .Net, and you can also find RabbitMQ tutorials for other languages including Java, Ruby and JavaScript here.)

Wrapping up

That's it for now. In my next post in this series I will walk through the prerequisites for building the simple service relay.

How have you handled inbound data interfaces for on-premises Dynamics 365 CE organizations? Let us know in the comments!

This the final post in my five-part series on creating loosely coupled data interfaces for Dynamics CRM using RabbitMQ. In part 3 and part 4 I showed two approaches for building a Dynamics CRM plug-in that publishes notification messages to a RabbitMQ exchange. In today’s post I will show how to create a Windows console application that reads messages from a queue and writes the data to Dynamics CRM. The code for this application is available on GitHub in the LeadWriterSample project under the LucasCrmMessageQueueTools solution.

The approach

This application is extraordinarily simple. On startup it prompts the user to supply connection information for the RabbitMQ queue that it will monitor as well as a Dynamics CRM connection string. It then monitors the queue for new JSON-formatted messages. When new messages arrive, it attempts to deserialize them into a lightweight "leadtype" object, and then it creates new lead records in CRM. Once a message is successfully processed and a lead is created, the application then sends a confirmation back to RabbitMQ so that the message can be removed from the queue.

The following code shows what happens after a connection to the RabbitMQ is established:

//wait for some messages

var consumer = new QueueingBasicConsumer(channel);

channel.BasicConsume(_queue, false, consumer);

 

Console.WriteLine(" [*] Waiting for messages. To exit press CTRL+C");

 

//instantiate crm org service

using (OrganizationService service = new OrganizationService(_targetConn))

{

   while (true)

   {

     //get the message from the queue

     var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();

 

     var body = ea.Body;

     var message = Encoding.UTF8.GetString(body);

 

     try

     {

       //deserialize message json to object

       LeadType lead = JsonConvert.DeserializeObject<LeadType>(message);

 

       try

       {

         //create record in crm

         Entity entity = new Entity("lead");

         entity["firstname"] = lead.FirstName;

         entity["lastname"] = lead.LastName;

         entity["subject"] = lead.Topic;

         entity["companyname"] = lead.Company;

         service.Create(entity);

 

         //write success message to cli

         Console.WriteLine("Created lead: {0} {1}", lead.FirstName, lead.LastName);

 

         //IMPORTANT - tell the queue the message was processed successfully so it doesn't get requeued

         channel.BasicAck(ea.DeliveryTag, false);

       }

       catch (FaultException<Microsoft.Xrm.Sdk.OrganizationServiceFault> ex)

       {

         //return error - note no confirmation is sent to the queue, so the message will be requeued

         Console.WriteLine("Could not create lead: {0} {1}", lead.FirstName, lead.LastName);

         Console.WriteLine("Error: {0}", ex.Message);

       }

     }

     catch(Exception ex)

     {

       //return error - note no confirmation is sent to the queue, so the message will be requeued

       Console.WriteLine("Could not process message from queue");

       Console.WriteLine("Error: {0}", ex.Message);

     }

   }

}

If this were to be used in production, I would have created a Windows service instead of a console application, but I wanted to make it easy to try out different connection parameters.

Verifying the application

The queuewriter.js application in the node-app directory in the GitHub repository contains a sample web page that can be used to publish lead data to the CRM-Leads queue. If the application is running, you can access the web page at http://<YOUR_SERVER_NAME>:3000/leadform. When the form’s submit button is clicked, an AJAX call posts a JSON object to the Node.js POST endpoint I showed in my previous post. If the LeadWriterSample console application is running, it will take the message from the queue and you will see a new lead record created in CRM. The screenshots below show each piece working.

The lead has been submitted via the web form, and a success message has been received from the Node.js endpoint.

The lead has landed in the CRM-Leads queue and is ready to be retrieved.

The console application has retrieved and processed the submitted lead message.

The lead record has been created in CRM.

One caveat about the demo lead form is that it has the RabbitMQ credentials embedded in the HTML source, so this code should not be used in production. My approach was originally formulated with the thought that a server-side process would build the JSON message to post to Node.js, so sensitive information would not be exposed. If you decide to use an AJAX post operation like is shown here, you would want to modify the queuewriter.js application to contain the credentials so they do not need to be passed from the end user’s web browser.

Wrapping up

That does it for this series, but I’ve just barely explored the capabilities of RabbitMQ. There’s so much more you can do with it than what I’ve shown here, and I hope I’ve piqued your interest about how you can use RabbitMQ or any other message broker in your Dynamics CRM projects. If you have any questions or want to continue the discussion, please share your thoughts in the comments.

A version of this post was originally published on the HP Enterprise Services Application Services blog.

Welcome back to my five-part series on creating loosely coupled data interfaces for Dynamics CRM using RabbitMQ. In my last post I showed how to build a Dynamics CRM plug-in that publishes notification messages to a RabbitMQ exchange using the official RabbitMQ .Net client library. Unfortunately, that plug-in can’t successfully communicate with a RabbitMQ server if it’s executed inside the Dynamics CRM sandbox, so in today’s post I will show how to achieve the same results with a sandboxed plug-in. The code for this plug-in is available on GitHub in the MessageQueueSandboxPlugin project under the LucasCrmMessageQueueTools solution.

The approach

As I mentioned in my previous post, last month I wrote a series of blog posts about how to create a near real-time streaming API using plug-ins and Node.js. That plug-in worked fine in the Dynamics CRM sandbox, and Node.js can easily publish messages to a RabbitMQ exchange, so today’s plug-in will post a JSON-formatted message to a Node.js application, and then that Node.js application will do the actual publishing to RabbitMQ. As a result, I only need to make a couple of minor modifications to my earlier Node.js message-posting plug-in so that it can pass the RabbitMQ connection parameters to my Node.js application. Additionally, the Node.js application that I described in my earlier series only needs a few changes to publish the message to a RabbitMQ exchange instead of sending it to Socket.IO clients.

The plug-in

The plug-in is registered for an operation (create, update, delete, etc.) with a FetchXML query in its unsecure configuration. When the plug-in step is triggered, its associated FetchXML query is executed, and then the resulting fields are serialized into a JSON object, which is then sent to a Node.js application called queuewriter.js via an HTTP POST request. The JSON object also needs to contain RabbitMQ connection details, so I pass them as part of the plug-in step’s unsecure configuration. Here’s the configuration XML fragment to enable case notifications:

<nodeendpoint>http://lucas-ajax.cloudapp.net:3000/rabbit_post_endpoint</nodeendpoint>
<endpoint>lucas-ajax.cloudapp.net</endpoint>
<exchange>CRM</exchange>
<routingkey>Case</routingkey>
<user>rabbituser</user>
<password>PASSWORDHERE</password>
<query><![CDATA[
<fetch mapping='logical'>
<entity name='incident'>
 <attribute name='ownerid'/>
 <attribute name='modifiedby'/>
 <attribute name='createdby'/>
 <attribute name='title'/>
 <attribute name='incidentid'/>
 <attribute name='ticketnumber'/>
 <attribute name='createdon'/>
 <attribute name='modifiedon'/>
 <filter type='and'>
  <condition attribute='incidentid' operator='eq' value='{0}' />
 </filter>
</entity>
</fetch>
]]>
</query>
</config>

Just like in my earlier Node.js plug-in, the FetchXML is extracted from the configuration XML, and the query is executed against Dynamics CRM. The results are then serialized to JSON using Json.NET just like before, except the serialized CRM data is included as a "message" object that is part of a parent JSON object that includes the RabbitMQ connection parameters. Here’s an example of the structure:

{

   "endpoint":"lucas-ajax.cloudapp.net",

   "username":"rabbituser",

   "password":"XXXXXXXX",

   "exchange":"CRM",

   "routingkey":"Lead",

   "message":{

     "property1":"value 1",

     "property2":"value 2",

     "property3":"value 3"

   }

}

Because this plug-in uses the Json.NET client library, it has to be merged with the plug-in assembly before registering it in Dynamics CRM. I’ve included a batch script called ilmerge.bat in the project directory on GitHub.

The Node.js application

The Node.js application (queuewriter.js) waits to receive JSON messages via HTTP POST from a client. When it receives a POST request, it checks whether the message is valid JSON. If it is, the RabbitMQ connection parameters are extracted and then the notification "message" object is published to the RabbitMQ exchange. If everything is successful, it sends "success" back as a response to the client. If any errors are encountered, it sends back a descriptive error message. I am using the node-amqp library for communicating with the RabbitMQ server, but the behavior isn’t that different from a .Net client. Here’s an extract with the relevant code:

if (request.method == 'POST') {

   request.on('data', function(chunk) {

     //check if received data is valid json

     if(IsJsonString(chunk.toString())){

       //convert message to json object

       var requestobject = JSON.parse(chunk.toString());

      

       //connect to rabbitmq

       var connection = amqp.createConnection({ host: requestobject.endpoint

       , port: 5672 //assumes default port

       , login: requestobject.username

       , password: requestobject.password

       , connectionTimeout: 0

       , authMechanism: 'AMQPLAIN'

       , vhost: '/' //assumes default vhost

       });

      

       //when connection is ready

       connection.on('ready', function () {

          //get the "message" property of the supplied request

          var message = JSON.stringify(requestobject.message);

         

          //post it to the exchange with the supplied routing key

          connection.exchange = connection.exchange(requestobject.exchange, {passive: true, confirm: true }, function(exchange) {

            exchange.publish(requestobject.routingkey, message, {mandatory: true, deliveryMode: 2}, function () {

              //if successful, write message to console

              console.log('Message published: ' + message);

             

              //send "success" back in response

              response.write('success');

             

              //close the rabbitmq connection and end the response

              connection.end();

              response.end();

            });

          });

       });

      

       //if an error occurs with rabbitmq

       connection.on('error', function () {

          //send error message back in response and end it

          response.write('failure writing message to exchange');

          response.end();

       });

     }

     else {

       //if request contains invalid json

       //send error message back in response and end it

       response.write("invalid JSON");

       response.end();

     }

   });

}

The complete queuewriter.js application is contained in the node-app directory in the GitHub repository.

Wrapping up

In addition to registering the plugin and registering a step to publish a notification message to RabbitMQ, you need to deploy and start the queuewriter.js application to publish messages. Once that’s done, you can verify everything is working as expected either by looking at the Queues tab in the RabbitMQ management web UI or running the CliConsumer sample application I showed in part 2.

Obviously using queuewriter.js as message proxy adds an extra layer of complexity, and you have to make sure that the application is up and running in order to process message, but it also offers a couple of advantages. First, by using queuewriter.js instead of a direct connection, you can easily use this same plug-in with different message brokers like Apache ActiveMQ and Microsoft’s Azure Service Bus. Second, the queuewriter.js application isn’t limited to just handling messages outbound from Dynamics CRM. You can also use it to process inbound messages without any changes. You just have to configure a client application to read messages from the queue and process them accordingly. A good example of this would be writing data submitted through a web form to Dynamics CRM via a RabbitMQ queue, and I will show that exact scenario in my next post!

A version of this post was originally published on the HP Enterprise Services Application Services blog.

This is the third post of a five-part series on creating loosely coupled data interfaces for Dynamics CRM using RabbitMQ.
Last time I showed how to install and configure a RabbitMQ server to support passing messages to and from Dynamics CRM. Today I will show how to build a Dynamics CRM plug-in that publishes notification messages to a RabbitMQ exchange using the official RabbitMQ .Net client library. The code for this plug-in is available on GitHub in the MessageQueuePlugin project under the LucasCrmMessageQueueTools solution.

Before going any further, let’s get some bad news out of the way. Plug-ins that execute in the Dynamics CRM sandbox cannot use RabbitMQ .Net client library to publish messages to a RabbitMQ server, so you can’t use today’s plug-in approach from a CRM Online organization. In my next post, I will be showing an alternate mechanism for publishing messages that you can use from a sandboxed plug-in, but today I want to focus on the most direct integration method. Now that we’re clear on the limitations of this approach, let’s get started!

The approach

Last month I wrote a series of blog posts about how to create a near real-time streaming API using plug-ins and Node.js. For this plug-in I’m going to basically copy the logic I used for the plug-in in that series.

This post outlines the approach in detail, but if you don’t want to read the entire thing, the basic idea was to create a plug-in that is registered for an operation (create, update, delete, etc.) with a FetchXML query in its unsecure configuration. When the plug-in step is triggered, its associated FetchXML query is executed, and then the resulting fields are serialized into a JSON object, which is then sent to the Node.js application via an HTTP POST request. Today’s plug-in operates in the exact same way, except instead of sending the JSON object to a Node.js endpoint, the JSON object will be published as a message to a RabbitMQ exchange.

Configuring the plug-in

To make the plug-in easily useable in any organization without needing to be recompiled, all the RabbitMQ connection parameters are stored in the unsecure configuration along with the FetchXML query for the data to retrieve. Here’s the configuration XML fragment to enable case notifications:

<config>
<endpoint>lucas-ajax.cloudapp.net</endpoint>
<exchange>CRM</exchange>
<routingkey>Case</routingkey>
<user>rabbituser</user>
<password>PASSWORDHERE</password>
<query><![CDATA[
<fetch mapping='logical'>
<entity name='incident'>
 <attribute name='ownerid'/>
 <attribute name='modifiedby'/>
 <attribute name='createdby'/>
 <attribute name='title'/>
 <attribute name='incidentid'/>
 <attribute name='ticketnumber'/>
 <attribute name='createdon'/>
 <attribute name='modifiedon'/>
 <filter type='and'>
  <condition attribute='incidentid' operator='eq' value='{0}' />
 </filter>
</entity>
</fetch>
]]>
</query>
</config>

Generating the notification message

Just like in my Node.js plug-in, the FetchXML is extracted from the configuration XML, and the query is executed against Dynamics CRM. The results are then serialized to JSON using Json.NET.

Publishing the message

The endpoint, exchange name, RabbitMQ user, RabbitMQ password and routing key values from the configuration XML are then used to establish a connection to RabbitMQ and publish the notification message to the exchange like so:

try
{
     //connect to rabbitmq
     var factory = new ConnectionFactory();
     factory.UserName = \_brokerUser;
     factory.Password = \_brokerPassword;
     factory.VirtualHost = "/";
     factory.Protocol = Protocols.DefaultProtocol;
     factory.HostName = \_brokerEndpoint;
     factory.Port = AmqpTcpEndpoint.UseDefaultPort;
     IConnection conn = factory.CreateConnection();
     using (var connection = factory.CreateConnection())
     {
         using (var channel = connection.CreateModel())
         {
             //tell rabbitmq to send confirmation when messages are successfully published
             channel.ConfirmSelect();
             channel.WaitForConfirmsOrDie();
            
             //prepare message to write to queue
             var body = Encoding.UTF8.GetBytes(jsonMsg);
 
             var properties = channel.CreateBasicProperties();
             properties.SetPersistent(true);
            
             //publish the message to the exchange with the supplied routing key
             channel.BasicPublish(_exchange, _routingKey, properties, body);
         }
     }
}
catch (Exception e)
{
     tracingService.Trace("Exception: {0}", e.ToString());
     throw;
}

If any errors are encountered, the message is captured via the tracing service, and then an exception is thrown.

Because this plug-in uses both the RabbitMQ .Net and Json.NET client libraries, they have to be merged with the plug-in assembly before registering it in Dynamics CRM. I’ve included a batch script called ilmerge.bat in the project directory on GitHub.

Wrapping up

After you register the plugin and register a step to publish a notification message to RabbitMQ, you can verify everything is working as expected either by looking at the Queues tab in the RabbitMQ management web UI or running the CliConsumer sample application I showed in
part 2.

A version of this post was originally published on the HP Enterprise Services Application Services blog.

Welcome back to this five-part series on creating loosely coupled data interfaces for Dynamics CRM using RabbitMQ. In my last post I discussed why you would want to incorporate a message broker into your Dynamics CRM data interfaces, and today I will show how to install and configure RabbitMQ to support the examples I’ll be presenting in the rest of the series.

Installation

First, you’ll need to download the installation files from here: http://www.rabbitmq.com/download.html. The RabbitMQ server runs on Windows, Linux, UNIX and Mac OS X, and there are installation guides for each supported platform. Because RabbitMQ is written in Erlang, you will need to install an Erlang VM before you can install RabbitMQ, but there is a download link provided in the installation guide. I set up my RabbitMQ server on a Windows 2012 server, and I was up and running in less than 10 minutes.

Once you’ve installed RabbitMQ and started the server, the easiest way to manage it is via the web-based management interface that’s included with the server distribution. You can enable the management interface with the rabbitmq-plugins tool. Run the following command to enable it: rabbitmq-plugins enable rabbitmq_management.

After the management plugin is enabled, you can access the web management UI from your server at http://localhost:15672. The default username is "guest" with "guest" as the password.

You’ll also need to configure any firewall rules necessary to allow access to your RabbitMQ server if it’s running on a server separate from your Dynamics CRM server. The default port is 5672, but that can be changed if you like. This page discusses RabbitMQ configuration in great detail.

Setting up users, queues and exchanges

The first thing you should do after the install is complete is change your default guest user password via the management UI. Then you can add additional users as necessary. For the examples in the rest of this series, you’ll need a user with full permissions on the default "/" virtual host. Here is what my "rabbituser" user account looks like:

Next you need to create the entities required to broker the messages between publishers and consumers. Before continuing, I recommend you take a moment to skim this Advanced Message Queuing Protocol (AMQP) overview document. If nothing else, at least read through the "hello, world" example section because it’s a great introduction to concepts that will be important throughout the rest of this series.

Queues
In the management UI, navigate to the Queues tab, and create two new durable queues named CRM-Cases and CRM-Leads. (You can create any queues you want, but my examples in the rest of this series use queues with those names.) The screenshot below shows the queues in my system.

Exchanges
After your queues are created, you can create an exchange and bindings to your queues so messages get routed correctly. Navigate to the Exchanges tab and create a new, durable exchange named CRM. After your CRM exchange is created, you should see something like the screenshot below.

Next, click on the name of the CRM exchange to open its edit screen. Scroll to the "add binding" section toward the bottom of the page and add a binding to the CRM-Cases queue for a routing key value of "Case" live in the following picture and click "bind."

Do the same for the CRM-Leads queue with "Lead" as the routing key. You should then see the two queues bound to the exchange.

Checking the configuration

At this point you should have everything in place to start publishing and consuming messages. You can verify your configuration works with the CliProvider and CliConsumer sample applications included in my GitHub repository as part of the LucasCrmMessageQueueTools solution.

First, build and run the CliProvider application. You will be prompted to supply basic connection details, and then you can type a message to publish to your RabbitMQ server.

Once the message has been published, you can verify there’s a message waiting in the correct queue on the Queues tab of the management UI.

Next, build and run the CliConsumer application. Once it connects to the CRM-Cases queue, the message will be retrieved and displayed.

When the CliConsumer application processes a message, it sends a confirmation back to queue that triggers removal of the message from the queue. You can check the Queues tab in the management UI to verify that the CRM-Cases queue is empty.

Wrapping up

That’s it for today. Your RabbitMQ server is now fully configured and ready for use with the examples in the rest of this series. Next time I will show how to send messages to a RabbitMQ exchange from a plug-in using the RabbitMQ .Net client library. See you then!

A version of this post was originally published on the HP Enterprise Services Application Services blog.

One of the things I love about Dynamics CRM is how easy it is to create data interfaces to enable integration with other systems. If you’ve worked with Dynamics CRM for any length of time, you’ve probably seen multiple web service integrations that enable interoperability with other line-of-business and legacy systems. A typical pair of inbound and outbound integrations might look like the picture below.

Using a tightly coupled connection between the source and target systems is usually the easiest (thus the quickest and cheapest) way to establish an integration, but this is often a bad idea. Consider the inbound scenario in which an external application is sending data to Dynamics CRM. What happens if the calling application misbehaves and starts sending thousands of requests per second? This has the potential to overload your CRM server and make it completely unusable. Now consider the outbound scenario in which a CRM plug-in calls an external web service. If the destination application’s web service is offline for a few minutes, the update from the CRM plug-in will not get received unless there’s some sort of error handling and retry logic built into the plug-in

An alternate approach

For these reasons, and lots of others (logging, security, scalability, just to name a few), it’s considered a best practice to create loosely coupled integrations that rely on a message broker that sits between the source and destination systems. Though the formal definition is more complicated, for our purposes a message broker can be thought of as a collection of queues that hold messages. Publishers write messages to queues, and then consumers pick up the messages and process them appropriately. Additionally, the message broker can be configured to keep messages in their queues until the consumers provide confirmation of successful processing.

Here’s an example of what the integrations I showed earlier would look like with a message broker.

For the outbound call from the CRM plug-in, the plug-in writes the message to a broker. The message is routed to a queue where it waits to be processed. A separate processing service application retrieves the message from the queue and sends it to the destination application. For the inbound call to CRM, the process works exactly the same, except the source and destination applications are reversed.

Why is a message broker better?

In the inbound call scenario, an effective message broker would typically be expected to handle a larger volume of inbound messages than Dynamics CRM because all it’s doing is receiving and routing the data without any additional processing. The processing service can then process the messages in the queue at a speed that doesn’t overload the Dynamics CRM server. In the case of the outbound call, the combination of a message broker and processing service can enable complex retry logic and custom logging without having to store it in the plugin layer. As an added bonus to either scenario, a message broker can provide a guarantee that messages don’t get lost between the source and destination systems as long as the message is successfully published to the broker.

Where do we go from here?

Over the course of my next four blog posts, I will show how to use RabbitMQ as a message broker in your Dynamics CRM data interfaces. I chose RabbitMQ for this series for several reasons:

1. It’s open source.
2. It runs on multiple platforms.
3. It’s easy to install and configure.
4. It’s fast at processing messages.

If you already have a different message broker in place in your organization or you would like to try a different message broker like Apache ActiveMQ or Microsoft’s Azure Service Bus, most of the approaches and a lot of the code I’m going to show in this series will still be applicable, with the notable exception of the post that discusses how to install and configure RabbitMQ.

Here’s the roadmap for the rest of the series:

• Part 2 – basic installation and configuration of a RabbitMQ
• Part 3 – creating a Dynamics CRM plug-in that publishes messages using the RabbitMQ .Net client library
• Part 4 – creating a sandboxed Dynamics CRM plug-in that publishes messages to RabbitMQ via Node.js
• Part 5 – reading messages from a queue and writing them to Dynamics CRM

If you just can’t wait to dig into the code, I’ve already posted everything to my repository on GitHub, so you can go ahead and take a look.

See you next time!

A version of this post was originally published on the HP Enterprise Services Application Services blog.