In this section, I’ll present a simple Service Broker application that demonstrates how all this fits together. I’ll start with an application that runs in a single database and, in subsequent sections, show how to move this application into a distributed environment.
The application to be built is an inventory application that accepts items from a manufacturing application and adds them to inventory. To simplify the code, it will be completely implemented as T-SQL stored procedures. In reality, one or more of the services would probably be either an external application connecting to the database or a CLR stored procedure.
A stored procedure called AddItemProc will send a message to the InventoryService with an XML body that contains an item to be added to the inventory. The procedure InventoryProc receives the message from the InventoryQueue, inserts a row in the Inventory table, and sends a response back to the ManufacturingService. The procedure ManufacturingProc receives the response message from the ManufacturingQueue and updates the State table with the response information. Figure 11-1 shows how this works.
First, run the code in Example 11-1 to create the Service Broker objects required to implement the Inventory service that receives and processes AddItem messages.
Example 11-1. Inventory service metadata
CREATE DATABASE Inventory;
GO
USE Inventory;
GO
---------------------------------------------------------------------
-- Create the message types we will need for the conversation
---------------------------------------------------------------------
CREATE MESSAGE TYPE [//microsoft.com/Inventory/AddItem];
CREATE MESSAGE TYPE [//microsoft.com/Inventory/ItemAdded];
/*-------------------------------------------------------------------
-- Create a contract for the AddItem conversation
-------------------------------------------------------------------*/
CREATE CONTRACT [//microsoft.com/Inventory/AddItemContract]
([//microsoft.com/Inventory/AddItem] SENT BY INITIATOR,
[//microsoft.com/Inventory/ItemAdded] SENT BY TARGET);
GO
/*--------------------------------------------------------------------
-- Create the procedure to service the Inventory target queue
-- Make it an empty procedure for now. We will fill it in later
--------------------------------------------------------------------*/
CREATE PROCEDURE dbo.InventoryProc AS
RETURN 0;
GO
/*--------------------------------------------------------------------
-- Create the Inventory Queue which will be the target of
-- the conversations. This is created with activation off.
--------------------------------------------------------------------*/
CREATE QUEUE dbo.InventoryQueue
WITH ACTIVATION (
STATUS = ON,
PROCEDURE_NAME = dbo.InventoryProc ,
MAX_QUEUE_READERS = 2,
EXECUTE AS SELF
) ;
/*--------------------------------------------------------------------
-- Create the Inventory Service. Because this is the Target
-- service, the contract must be specified
--------------------------------------------------------------------*/
CREATE SERVICE [//microsoft.com/InventoryService] ON QUEUE dbo.InventoryQueue
([//microsoft.com/Inventory/AddItemContract]);
/*--------------------------------------------------------------------
-- Create a table to hold the inventory we're adding
-- This isn't meant to be realistic - just a way to show that the
-- service did something
--------------------------------------------------------------------*/
CREATE TABLE dbo.Inventory
(
PartNumber NVARCHAR(50) Primary Key Clustered NOT NULL,
Description NVARCHAR (2000) NULL,
Quantity INT NULL,
ReorderLevel INT NULL,
Location NVARCHAR(50) NULL
);
GOThere’s nothing too exciting here. One thing that might seem odd is that the InventoryProc stored procedure is empty. This is done so that the name can be used to specify the activation parameters for the queue without creating the whole procedure. The other way to do this is to create the queue without activation and then use ALTER QUEUE to add activation after the stored procedure exists. This approach works well for me because I have a tendency to forget to do activation if I don’t do it when I create the queue. The Inventory table has a few columns to illustrate how to transfer data from a message to a table, but it’s obviously not a realistic inventory table. Example 11-2 has the real logic to implement the Inventory service that processes the AddItem messages.
Example 11-2. Inventory service program
ALTER PROCEDURE dbo.InventoryProc
AS
Declare @message_body AS xml;
Declare @response AS xml;
Declare @message_type AS sysname;
Declare @dialog AS uniqueidentifier ;
Declare @hDoc AS int;
-- This procedure will just sit in a loop processing event messages in
-- the queue until the queue is empty
WHILE (1 = 1)
BEGIN
BEGIN TRANSACTION
-- Receive the next available message
WAITFOR (
RECEIVE top(1) -- just handle one message at a time
@message_type = message_type_name,
@message_body = message_body,
@dialog = conversation_handle
FROM dbo.InventoryQueue
), TIMEOUT 2000
-- If we didn't get anything, bail out
IF (@@ROWCOUNT = 0)
BEGIN
ROLLBACK TRANSACTION
BREAK;
END
/*--------------------------------------------------------------------
-- Message handling logic based on the message type received
--------------------------------------------------------------------*/
-- Handle End Conversation messages by ending our conversation also
IF (@message_type = 'http://schemas.microsoft.com/SQL/ServiceBroker/EndDialog')
BEGIN
PRINT 'End Dialog for dialog # ' + cast(@dialog as nvarchar(40));
END CONVERSATION @dialog;
END
-- For error messages, just end the conversation. In a real app, we
-- would log the error and do any required cleanup.
ELSE IF (@message_type = 'http://schemas.microsoft.com/SQL/ServiceBroker/Error')
BEGIN
PRINT 'Dialog ERROR dialog # ' + cast(@dialog as nvarchar(40));
END CONVERSATION @dialog;
END
-- Handle an AddItem message
ELSE IF (@message_type = '//microsoft.com/Inventory/AddItem')
BEGIN
SET @response = N'Item added successfully'
-- Parse the message body and add to the inventory
BEGIN TRY
INSERT INTO dbo.Inventory
(PartNumber, Description, Quantity, ReorderLevel, Location)
select itm.itm. value ('(PartNumber/text())[1]', 'nvarchar(50)')
as PartNumber,
itm.itm.value('(Description/text())[1]', 'nvarchar(2000)')
as Description,
itm.itm.value('(Quantity/text())[1]', 'int')
as Quantity,
itm.itm.value('(ReorderLevel/text())[1]', 'int')
as ReorderLevel,
itm.itm.value('(Location/text())[1]', 'nvarchar(50)')
as Location
from @message_body.nodes('/Item[1]') as itm(itm);
END TRY
BEGIN CATCH
ROLLBACK TRANSACTION
-- Create a new transaction to send the response
BEGIN TRANSACTION
SET @response = ERROR_MESSAGE();
-- ToDo - log the error
-- ToDo - poison message handling
END CATCH;
-- Send a response message confirming the add was done
SEND ON CONVERSATION @dialog
MESSAGE TYPE [//microsoft.com/Inventory/ItemAdded] (@response);
-- We handle one message at a time so we're done with this dialog
END CONVERSATION @dialog;
END -- If message type
COMMIT TRANSACTION
END -- while
GOThe process flow of this procedure is the basis of almost all Service Broker services. The procedure is a continuous loop that receives a message at the top, processes the message with conditional logic based on the message type received, and then goes back to the top and receives another message. The loop continues until the @@ROWCOUNT after RECEIVE is 0, which indicates the queue is empty. Each loop is a separate transaction, which means there is only one message per transaction. This is not optimal, but it makes the logic easier to follow, so I usually do this for samples. A high-performance Service Broker application would read all the messages available for a conversation group in a single RECEIVE statement and then process them all at once. If the message volume is high, you might also consider going through the loop a few times before committing the transaction. This approach makes the logic more efficient, but it also makes the processing logic much more complex.
The message processing logic is the T-SQL equivalent of a C# "switch" statement with three different sections for the three message types that might be found on the queue. It’s important to remember that because a RECEIVE statement specifies a queue, any messages type allowed on the queue might be received. This means that if there are two or more services associated with the queue, the logic in the queue reader must be able to process any message allowed on any of the services.
The first two message types are error and end dialog. These two message types can be received by any service, so all service logic must have logic to handle them. At a minimum, the message-handling logic must contain an END CONVERSATION to close out this end of the dialog. If some cleanup is required before ending the conversation, handling these messages will be more complex.
The last message type is the AddItem message. The message body is an XML document that contains the data required to add a row to the Inventory table. The sample code processes the message with a Nodes query that inserts the XML data in the Inventory table. The error-handling logic is not robust, and it’s there only to illustrate where real error-handling would go. If there is an error in the insert statement–a primary key violation, perhaps–the CATCH block will roll back the transaction and put the error message into the response message. The problem with doing that is that if the error is unrecoverable, as it would be if the message contained a duplicate key, the message would be put back in the queue. When the same message is read again, it would fail again and be rolled back again. This behavior would put the stored procedure into a tight loop. A message such as this that can never be processed correctly is called a poison message. Transactional messaging applications must deal with poison messages. (See the "Poison Messages" section later in the chapter for details about how to deal with them.) To avoid getting into a loop, the sample unconditionally ends the conversation whether the message was processed successfully or not. This is probably not what you would want to do in a real application.
After inserting the message contents into the Inventory table, the response message is sent back to the ManufacturingService and the dialog is ended. The response message contains either a message indicating success or the error text if the insert failed. The @dialog variable used for the SEND and END DIALOG was obtained from the received message, so the message will be routed back to the calling dialog.
Now that a service is ready to process messages, let’s implement a stored procedure to send the AddItem message to this service. This stored procedure will begin a dialog to the InventoryService and send a message on the dialog. The FROM SERVICE for this dialog will be the ManufacturingService. Because the BEGIN DIALOG command uses the ManufacturingService as the FROM SERVICE, the ManufacturingService is the initiator of the dialog and the InventoryService is the target. Example 11-3 creates the queue and service for the ManufacturingService, and it creates the stored procedure to send the AddItem message.
Example 11-3. Dialog initiator procedure
/*-------------------------------------------------------------------
-- Create an empty procedure for the initiator so we can use it
-- in the activation parameters when we create the queue
-------------------------------------------------------------------*/
CREATE PROCEDURE dbo.ManufacturingProc AS
RETURN 0;
GO
/*-------------------------------------------------------------------
-- Create the initiator queue. Activation is configured
-- but turned off
-------------------------------------------------------------------*/
CREATE QUEUE dbo.ManufacturingQueue
WITH ACTIVATION (
STATUS = ON,
PROCEDURE_NAME = dbo.ManufacturingProc ,
MAX_QUEUE_READERS = 2,
EXECUTE AS SELF
);
/*-------------------------------------------------------------------
-- Create the Manufacturing service. Because it is the initiator, it
-- doesn't require contracts.
-------------------------------------------------------------------*/
CREATE SERVICE [//microsoft.com/ManufacturingService]
ON QUEUE dbo.ManufacturingQueue;
/*--------------------------------------------------------------------
-- Create a table to hold the state for our conversation
-- We use the conversation handle as a key instead of the
-- conversation group ID because we just have one conversation
-- in our group.
--------------------------------------------------------------------*/
CREATE TABLE dbo.InventoryState
(
PartNumber UNIQUEIDENTIFIER Primary Key Clustered NOT NULL,
Dialog UNIQUEIDENTIFIER NULL,
State NVARCHAR(50) NULL
);
GO
/*-------------------------------------------------------------------
-- Create the initiator stored procedure
-------------------------------------------------------------------*/
CREATE PROCEDURE AddItemProc
AS
DECLARE @message_body AS xml;
DECLARE @Dialog AS uniqueidentifier;
DECLARE @partno AS uniqueidentifier;
--Set the part number to a new GUID so we can run
--this an unlimited number of times
SET @partno = NEWID();
-- Populate the message body
SET @message_body = '<Item>
<PartNumber>' + CAST (@partno as NVARCHAR(50)) + '</PartNumber>
<Description>2 cm Left Threaded machine screw</Description>
<Quantity>5883</Quantity>
<ReorderLevel>1000</ReorderLevel>
<Location>Aisle 32, Rack 7, Bin 29</Location>
</Item>';
BEGIN TRANSACTION
-- Begin a dialog to the Hello World Service
BEGIN DIALOG @Dialog
FROM SERVICE [//microsoft.com/ManufacturingService]
TO SERVICE '//microsoft.com/InventoryService'
ON CONTRACT [//microsoft.com/Inventory/AddItemContract]
WITH ENCRYPTION = OFF, LIFETIME = 3600;
-- Send message
SEND ON CONVERSATION @Dialog
MESSAGE TYPE [//microsoft.com/Inventory/AddItem] (@message_body);
-- Put a row into the state table to track this conversation
INSERT INTO dbo.InventoryState
VALUES (@partno, @Dialog, 'Add Item Sent'),
COMMIT TRANSACTION
GOThis service uses the message types and contract created for the InventoryService, so they don’t have to be created again. If you create this service in another database, you must first create the message types and contract.
AddItemProc is a simple procedure. It populates the message body with an XML document, begins a dialog to the InventoryService, and sends the message. For each message sent, a row is inserted into the InventoryState table. The InventoryState table tracks the progress of the AddItem messages, so a user can see the progress of the dialog.
To see whether this works, run the following:
EXEC AddItemProc
If everything is working correctly, there should be a row in the Inventory table and a row in the InventoryState table. Because you haven’t written the logic to handle the response yet, there should be a response message in the ManufacturingQueue.
You have now sent a message on a dialog, processed the message, and sent a response message. All that’s left to do is process the response message. To do this, run the code in Example 11-4 to create another simple Service Broker service with a message processing loop. The process flow will be the same as that for the inventory procedure.
Example 11-4. Manufacturing procedure
ALTER PROCEDURE dbo.ManufacturingProc
AS
DECLARE @message_body AS xml;
DECLARE @message_type AS sysname;
DECLARE @dialog AS uniqueidentifier ;
-- This procedure will just sit in a loop processing event messages in
-- the queue until the queue is empty
WHILE (1 = 1)
BEGIN
BEGIN TRANSACTION
-- Receive the next available message
WAITFOR (
RECEIVE top(1)
@message_type=message_type_name,
@message_body=message_body,
@dialog = conversation_handle
FROM dbo.ManufacturingQueue
), TIMEOUT 2000;
-- If we didn't get anything, bail out
IF (@@ROWCOUNT = 0)
BEGIN
ROLLBACK TRANSACTION
BREAK;
END
IF (@message_type =
'http://schemas.microsoft.com/SQL/ServiceBroker/EndDialog')
BEGIN
PRINT 'End Dialog for dialog # ' + CAST(@dialog as nvarchar(40));
END CONVERSATION @dialog;
END
ELSE IF (@message_type =
'http://schemas.microsoft.com/SQL/ServiceBroker/Error')
BEGIN
PRINT 'Dialog ERROR dialog # ' + CAST(@dialog as nvarchar(40));
END CONVERSATION @dialog;
END
ELSE IF (@message_type = '//microsoft.com/Inventory/ItemAdded')
BEGIN
UPDATE dbo.InventoryState SET State = CAST(@message_body
AS NVARCHAR(1000)) WHERE Dialog = @dialog;
END
COMMIT TRANSACTION
END -- while
GOThis procedure is almost identical to InventoryProc, with the exception of the ItemAdded message type handling. The logic just updates the state table with the status returned in the message body. Notice that the initiator side of the dialog is actually two different stored procedures–one that begins the dialog and sends the message, and another that handles the responses. This is a normal pattern for asynchronous services. The dialog initiator sends a message to a background service and then either goes on to do something else or goes away. When the response is received, it is handled in the background. In some cases, the response is just recorded in the database; in other cases, it might be communicated back to the user through an e-mail message or something similar.
You’ve now seen a simple Service Broker conversation in action. The logic might seem complex at first, but keep in mind that most Service Broker applications follow the same pattern. So, once you understand it, you can apply it to a number of applications. In fact, this processing pattern is so common that it has been incorporated into a sample object model that ships with SQL Server 2005. Install the SQL Server samples, and look for the ServiceBrokerInterface sample. The sample service incorporates the message loop handling, and you just have to supply the logic to handle the messages received. In the next two sections of this chapter, I’ll move the ManufacturingService first to another database in the same instance to demonstrate dialog security and then to another instance to demonstrate remote communications with Service Broker.
One of the advantages of transactional messaging is that if the transaction that is processing a message fails and rolls back, the message is still on the queue because the RECEIVE command is also rolled back. The application can then receive the message again, and if it processes successfully this time, the transaction will commit and the message will disappear from the queue. This is why writing transactional message processing applications is much easier than writing nontransactional message applications.
The down side of transactional messaging occurs when processing a message produces an error that retrying won’t fix. For example, in the sample application that inserts an item into the inventory table, if there is already an item with the same part number and the part number column has a unique constraint, then no matter how many times the message is processed it will fail. The application will then go into a tight loop processing the message, rolling back the transaction, and then receiving the same message again. A message that can’t be processed is known as a poison message. If left unchecked, this kind of message will cause the application to hang on the same message and have a major negative impact on the performance of the entire database. To prevent this, if Service Broker detects five rollbacks in a row on a queue, it will disable the queue to stop the poison message. This action is a last resort, however, because the application that uses the queue will stop until the administrator ends the offending dialog and re-enables the queue.
The right way to deal with poison messages is to prevent them. If the application fails to process a message, it should roll back the transaction and then decide whether processing the same message again will be likely to succeed. If the error was something like a deadlock or a lock timeout, it is worth trying again. However, if the error is a constraint violation that won’t be resolved the next time the message is processed, the conversation that contains the message should be ended and an application error should be logged. If it makes sense in the application to continue the dialog without the poison message, you can commit the receive to get rid of the poison message and go on with the dialog. The approach to handling poison messages depends on the application, the dialog, and even which message failed–so there’s no single right way to handle them. There is one wrong way, which is to always roll back errors and rely on Service Broker to handle them by disabling the queue.