RabbitMQ - SQL Server

A week or two ago, I saw a message on the RabbitMQ Users forum about how to get messages sent from SQL Server to RabbitMQ. Since we are working closely with this at Derivco , I left some suggestions there and also said that I was writing a blog post about how this can be done. Part of my message was not entirely correct - at least until this point (sorry, Bro, was very busy).

Awesome stuff, this is your SQL Server . With it, it is very easy to put information into a database. Retrieving data from a database using a query is just as easy. But getting just the updated or inserted data is already a little more difficult. Think about real-time events; A purchase is made - someone needs to be notified at the same moment as soon as this has happened. Perhaps someone will say that such data should not be pushed out of the database, but from somewhere else. Of course, the way it is, but quite often we simply have no choice.

We had a task: to send events from the database to the outside for further processing, and the question was how to do it?

SQL Server and external communications

During the existence of SQL Server, there have been several attempts to organize communications outside the database; SQL Server Notification Services (NS), which appeared in SQL Server 2000, and then, in SQL Server 2005, SQL Server Service Broker (SSB) appeared. I described them in my book A First Look at SQL Server 2005 for Developers , along with Bob Boshman and Dan Sullivan. NS appeared in SQL Server 2000, as I said, and was redesigned in beta version of SQL Server 2005. However, from the ready-for-sale (RTM) version of SQL Server 2005, NS was cut out completely.

Note: If you read the book, you will find a number of features there that were not in the RTM version.

SSB survived, and Microsoft introduced Service Broker External Activator (EA) in the SQL Server 2008 Feature Pack. It allows SSB to interact outside the local database. In theory, it sounds good, but in practice it is cumbersome and confused. We conducted several tests and quickly realized that he was not fulfilling what we needed. In addition, SSB did not give us the performance that was needed, so we had to invent something else.
')

SQLCLR

What we came to as a result was based on the SQLCLR technology. SQLCLR is a .NET platform embedded in the SQL Server kernel and with its help you can execute .NET code inside the kernel. Since we are executing .NET code, we are able to do almost everything in a regular .NET application.

Note: I wrote “almost” above, because in fact there are some limitations. In this context, these restrictions have little effect on what we are going to do.

The principle of operation of SQLCLR is as follows: the code is compiled into a dll library, and then this library is registered by means of SQL Server:

Build build

CREATE ASSEMBLY [RabbitMQ.SqlServer] AUTHORIZATION rmq FROM 'F:\some_path\RabbitMQSqlClr4.dll' WITH PERMISSION_SET = UNSAFE; GO 

Code Snippet 1: Build an absolute path

The code performs the following actions:

• CREATE ASSEMBLY - creates an assembly with the given name (no matter what it should be).
• AUTHORIZATION - Indicates the owner of the assembly. In this case, rmq is a predefined SQL Server role.
• FROM - determines where the original assembly is located. In the FROM , you can also specify the path in binary or UNC formats. The installation files for this project use a binary representation.
• WITH PERMISSION_SET - sets permissions. UNSAFE is the least UNSAFE , and is needed in this case.

Note: regardless of whether the role or the login name was used in the AUTHORIZATION clause, the appdomain class must be created with the same name as when loading the assembly into the domain. It is recommended to separate assemblies with different names of appdomain classes so that when one assembly falls, the others do not collapse. However, if assemblies have dependencies on each other, they cannot be divided into different classes.

When the assembly is created, we make the .NET method wrappers in it:

 CREATE PROCEDURE rmq.pr_clr_PostRabbitMsg @EndpointID int, @Message nvarchar(max) AS EXTERNAL NAME [RabbitMQ.SqlServer].[RabbitMQSqlClr.RabbitMQSqlServer].[pr_clr_PostRabbitMsg]; GO 

Code snippet 2: .NET method wrapper

The code performs the following actions:

• Creates a T-SQL stored procedure named rmq.pr_clr_PostRabbitMsg that takes two parameters; @EndpointID and @Message .
• Instead of the procedure body, an external source is used, which consists of:
  
  • The assembly is named RabbitMQ.SqlServer , that is, the aggregate we created above in the code snippet 1 .
  • Full type (namespace and class): RabbitMQSqlClr.RabbitMQSqlServer
  • Method from the above namespace and class: pr_clr_PostRabbitMsg .

When the rmq.pr_clr_PostRabbitMsg procedure is rmq.pr_clr_PostRabbitMsg , the pr_clr_PostRabbitMsg method will be called.

Note: when creating a procedure, the assembly name is not case-sensitive, unlike the full name of the type and method. It is not necessary that the name of the procedure being created matches the name of the method. However, the final data types for the parameters must match.

As I said earlier, we in Derivco need to send data outside of SQL Server, so we use SQLCLR and RabbitMQ (RMQ).

RabbitMQ

RMQ is an open source message broker that implements the Advanced Message Queuing Protocol (AMQP) and is written in Erlang.

Since RMQ is a message broker, AMQP client libraries are required to connect to it. The application refers to the client libraries and with their help opens the connection and sends messages - such as, for example, a call is made through ADO.NET to SQL Server. But unlike ADO.NET, where, most likely, the connection is opened each time the database is accessed, the connection remains open during the entire period of the application.

Thus, in order to be able to interact from the database with RabbitMQ we need the application and the client .NET library for RabbitMQ.

Note: In the following part of this article, RabbitMQ code snippets will appear, but without detailed explanations of what they do. If you are new to working with RabbitMQ, then I suggest looking at various RabbitMQ lessons to understand the purpose of the code. Hello World C # tutorial is a good start. One of the differences between textbooks and code examples is that the examples do not declare exchangers. It is assumed that they are predetermined.

RabbitMQ.SqlServer

RabbitMQ.SqlServer is an assembly that uses the client .NET library for RabbitMQ and provides the ability to send messages from the database to one or several RabbitMQ endpoints (VHosts and exchangers). The code can be downloaded / posted from my RabbitMQ-SqlServer repository on GitHub. It contains source codes of assembly and installation files (i.e. you do not have to compile them yourself).

Note: this is just an example to show how SQL Server can interact with RabbitMQ. This is NOT a finished product and not even a part of it. If this code breaks your brain you don’t have to blame me, for this is just an example.

Functionality

When the assembly is loaded, or when its initialization is explicitly called, or indirectly, when the wrapper procedure is called, the assembly loads the connection string into the local database into which it was installed, as well as the RabbitMQ endpoints to which it is connected:

Connection

 internal bool InternalConnect() { try { connFactory = new ConnectionFactory(); connFactory.Uri = connString; connFactory.AutomaticRecoveryEnabled = true; connFactory.TopologyRecoveryEnabled = true; RabbitConn = connFactory.CreateConnection(); for (int x = 0; x < channels; x++) { var ch = RabbitConn.CreateModel(); rabbitChannels.Push(ch); } return true; } catch(Exception ex) { return false; } } 

Snippet code 3: connect to the endpoint

At the same time, part of the connection to the endpoint also creates IModels on the connection, and they are used when sending (adding to the queue) messages:

Posting a message

 internal bool Post(string exchange, byte[] msg, string topic) { IModel value = null; int channelTryCount = 0; try { while ((!rabbitChannels.TryPop(out value)) && channelTryCount < 100) { channelTryCount += 1; Thread.Sleep(50); } if (channelTryCount == 100) { var errMsg = $"Channel pool blocked when trying to post message to Exchange: {exchange}."; throw new ApplicationException(errMsg); } value.BasicPublish(exchange, topic, false, null, msg); rabbitChannels.Push(value); return true; } catch (Exception ex) { if (value != null) { _rabbitChannels.Push(value); } throw; } } 

The Post method is called from the pr_clr_PostRabbitMsg(int endPointId, string msgToPost) method pr_clr_PostRabbitMsg(int endPointId, string msgToPost) , which was presented as a procedure using the CREATE PROCEDURE clause in code snippet 2:

Post call method

 public static void pr_clr_PostRabbitMsg(int endPointId, string msgToPost) { try { if(endPointId == 0) { throw new ApplicationException("EndpointId cannot be 0"); } if (!isInitialised) { pr_clr_InitialiseRabbitMq(); } var msg = Encoding.UTF8.GetBytes(msgToPost); if (endPointId == -1) { foreach (var rep in remoteEndpoints) { var exch = rep.Value.Exchange; var topic = rep.Value.RoutingKey; foreach (var pub in rabbitPublishers.Values) { pub.Post(exch, msg, topic); } } } else { RabbitPublisher pub; if (rabbitPublishers.TryGetValue(endPointId, out pub)) { pub.Post(remoteEndpoints[endPointId].Exchange, msg, remoteEndpoints[endPointId].RoutingKey); } else { throw new ApplicationException($"EndpointId: {endPointId}, does not exist"); } } } catch { throw; } } 

Code Snippet 5: Method Representation as a Procedure

When executing the method, it is assumed that the caller sends the identifier of the end point to which the message is to be sent, and, in fact, the message itself. If the value -1 is passed as the identifier of the end point, then we iterate over all the points and send a message to each of them. The message comes in the form of a string from which we get the bytes using Encoding.UTF8.GetBytes . In a production environment, the Encoding.UTF8.GetBytes call should be replaced with serialization.

Installation

To install and run the example, you need all the files in the src\SQL folder. To install, follow these steps:

• Run the script 01.create_database_and_role.sql . He will create:
  
  • RabbitMQTest test database, where the assembly will be created.
  • rmq role to be assigned as the owner of the assembly
  • scheme, which will also be called rmq . In this scheme, various database objects are created.
    
  
  
• Run the 02.create_database_objects.sql file. He will create:
  
  
  • table rmq.tb_RabbitSetting , which will store the connection string to the local database.
  • the rmq.tb_RabbitEndpoint table, in which one or several RabbitMQ endpoints will be stored.
  
  
• In file 03.create_localhost_connstring.sql change the value of the @connString variable to the correct connection string to the RabbitMQTest database created in step 1 and run the script.
  

Before proceeding, you need to have a running instance of the RabbitMQ broker and VHost (by default, VHost is represented as /). As a rule, we have several vhost, just for isolation. This host also needs an exchanger, in the example we use amq.topic . When your RabbitMQ broker is ready, edit the parameters of the rmq.pr_UpsertRabbitEndpoint procedure, which is in the 04.upsert_rabbit_endpoint.sql file:

RabbitMQ endpoint

 EXEC rmq.pr_UpsertRabbitEndpoint @Alias = 'rabbitEp1', @ServerName = 'RabbitServer', @Port = 5672, @VHost = 'testHost', @LoginName = 'rabbitAdmin', @LoginPassword = 'some_secret_password', @Exchange = 'amq.topic', @RoutingKey = '#', @ConnectionChannels = 5, @IsEnabled = 1 

Code Snippet 6: Creating an Endpoint in RabbitMQ

At this stage, it is time to deploy the assembly. There are differences in the deployments for SQL Server versions prior to SQL Server 2014 (2005, 2008, 2008R2, 2012), and for 2014 and higher. The difference is in the supported version of the CLR. Before SQL Server 2014, the .NET platform ran in the CLR version 2, and in SQL Server 2014 and higher, version 4 is used.

SQL Server 2005 - 2012

Let's start with the versions of SQL Server that run on CLR 2, since there are some special features there. We need to deploy the created assembly, and at the same time deploy the Rabbit client .NET library RabbitMQ ( RabbitMQ.Client ). From our assembly we will refer to the RabbitMQ client library. Since we planned to use CLR 2, then our build and RabbitMQ.Client should be compiled based on .NET 3.5. There are problems.

All the latest versions of RabbitMQ.Client compiled for CLR 4, so they cannot be used in our build. The latest version of client libraries for CLR 2 is compiled on .NET 3.4.3. But even if we try to deploy this assembly, we get an error message:


Figure 1: System.ServiceModel assembly missing

This version of RabbitMQ.Client refers to an assembly that is not part of the CLR SQL Server. This is a WCF assembly, and this is one of the limitations in SQLCLR that I mentioned above: this particular assembly is intended for those types of tasks that are not allowed to be performed in SQL Server. The latest versions of RabbitMQ.Client do not have these dependencies, so they can be used without any problems, except for the annoying requirements of the CLR 4 environment. What should I do?

As you know, RabbitMQ is open source, well, and we are developers, right? ;) So let's recompile! In the version to the latest releases (i.e. version <3.5.0) of RabbitMQ.Client I deleted the references to System.ServiceModel and recompiled. I had to change a couple of lines of code using the System.ServiceModel functionality, but these were minor changes.

In this example, I did not use client version 3.4.3, but took the stable release 3.6.6 and recompiled using .NET 3.5 (CLR 2). It almost worked :), except that later RabbitMQ.Client releases use Task 'and which are not originally part of .NET 3.5.

Fortunately, there is a version of System.Threading.dll for .NET 3.5 that includes Task . I downloaded it, set up links and everything went! Here the main thing is that System.Threading.dll should be installed along with the assembly.

Note: The RabbitMQ.Client source, from which I compiled the version on .NET 3.5, is in my repository on the GitHub RabbitMQ Client 3.6.6 .NET 3.5 . The dll binary along with the System.Threading.dll for .NET 3.5 also lies in the lib\NET3.5 repository (RabbitMQ-SqlServer) .

To install the necessary assemblies ( System.Threading , RabbitMQ.Client and RabbitMQ.SqlServer ), run the installation scripts from the src\sql directory in the following order:

1. 05.51.System.Threading.sql2k5-12.sql - System.Threading
2. 05.52.RabbitMQ.Client.sql2k5-12.sql - RabbitMQ.Client
3. 05.53.RabbitMQ.SqlServer.sql2k5-12.sql - RabbitMQ.SqlServer

SQL Server 2014+

In SQL Server 2014 and later, the build is compiled under .NET 4.XX (my example is 4.5.2), and you can refer to any of the latest versions of RabbitMQ.Client , which can be obtained using NuGet . In my example, I use 4.1.1. RabbitMQ.Client , which is also in the lib\NET4 repository (RabbitMQ-SqlServer) .

To install, run the scripts from the src\sql directory in the following order:

1. 05.141.RabbitMQ.Client.sql2k14+.sql - RabbitMQ.Client
2. 05.142.RabbitMQ.SqlServer.sql2k14+.sql - RabbitMQ.SqlServer

SQL method wrappers

To create the procedures that will be used from our assembly (3.5 or 4), run the script 06.create_sqlclr_procedures.sql . It will create T-SQL procedures for the three .NET methods:

• rmq.pr_clr_InitialiseRabbitMq calls pr_clr_InitialiseRabbitMq . Used to load and initialize the RabbitMQ.SqlServer assembly.
• rmq.pr_clr_ReloadRabbitEndpoints calls pr_clr_ReloadRabbitEndpoints . Loads various RabbitMQ endpoints.
• rmq.pr_clr_PostRabbitMsg calls pr_clr_PostRabbitMsg . Used to send a message to RabbitMQ.

The script also creates a simple T-SQL procedure, rmq.pr_PostRabbitMsg , which is applied to rmq.pr_clr_PostRabbitMsg . This is a wrapper procedure that knows what to do with the data, handles exceptions, etc. In a production environment, we have several similar procedures that handle different types of messages. Read more about this below.

Using

From all of the above, we can call rmq.pr_PostRabbitMsg or rmq.pr_clr_PostRabbitMsg to send messages to RabbitMQ, passing the endpoint identifier and the message itself as a string. All this, of course, cool, but I would like to see how it will work in reality.

What we do in production environments — in stored procedures that process data that needs to be sent to RabbitMQ, we collect data to send and in the connection block we call a procedure similar to rmq.pr_PostRabbitMsg . Below is a very simplified example of such a procedure:

Data processing procedure

 ALTER PROCEDURE dbo.pr_SomeProcessingStuff @id int AS BEGIN SET NOCOUNT ON; BEGIN TRY --     DECLARE @endPointId int; --    DECLARE @msg nvarchar(max) = '{' --        SET @msg = @msg + '"Id":' + CAST(@id AS varchar(10)) + ',' --  -  SET @msg = @msg + '"FName":"Hello",'; SET @msg = @msg + '"LName":"World"'; SET @msg = @msg + '}'; -- -  --     -,  -  SELECT @endPointId = 1; --    --     EXEC rmq.pr_PostRabbitMsg @Message = @msg, @EndpointID = @endPointId; END TRY BEGIN CATCH DECLARE @errMsg nvarchar(max); DECLARE @errLine int; SELECT @errMsg = ERROR_MESSAGE(), @errLine = ERROR_LINE(); RAISERROR('Error: %s at line: %d', 16, -1, @errMsg, @errLine); END CATCH END 

In the code snippet 7, we see how the necessary data is captured and processed in the procedure and sent after processing. To use this procedure, run script 07.create_processing_procedure.sql from the src\SQL directory.

Let's run it all

At this point, you should be ready to send several messages. Before testing, make sure you have queues in RabbitMQ that are attached to the endpoint exchanger at rmq.tb_RabbitEndpoint .

So, to start you need to do the following:
Open the file 99.test_send_message.sql .
Execute

 EXEC rmq.pr_clr_InitialiseRabbitMq; 

to initialize the build and load RabbitMQ endpoints. This is not a required step, but it is recommended to preload the assembly after it is created or modified.

Execute

 EXEC dbo.pr_SomeProcessingStuff @id = 101 

(you can use any other identifier you like).

If everything worked out without errors, then a message should appear in the RabbitMQ queue! So you used SQLCLR to send a message to RabbitMQ.

Congrating!