Using the WCF SQL Adapter to read messages from SSB queues and submit them to BizTalk
In this post, I’m going to demonstrate how the Polling feature in the WCF SQL Adapter can be used to read messages from SQL Service Broker (SSB) queues and publish them in the BizTalk MessageBox. And to make it really simple, we’re going to do it without an orchestration.
(Note – most of the SSB portion in this post has been based on the SSB tutorial available at http://msdn2.microsoft.com/en-us/library/bb839489.aspx).
- An external source publishes messages into a SSB queue.
- At fixed intervals (the polling interval), the adapter checks the SSB queue to see if a message is available.
- If a message is available, the adapter performs the actual RECEIVE operation to remove the message from the SSB queue and publish it in BizTalk – these operations are performed within the same DTC transaction.
- A filter is used to send the data to a folder on the file system (using the File Adapter)
Creating the database artifacts required for the SSB conversation:
The artifacts which we need to create are:
- A message type, which denotes the format of the messages in the queue
- A contract, which denotes the conversation between a sender and a receiver (the contract also specifies the type of the message which will flow between them)
- The Initiator and Target queues, in which the messages are stored
- The Initiator and Target services, utilizing the above-mentioned queues.
I’m not going to explain the SQL statements below in detail, mainly because this post is not meant to be a tutorial on SSB – you can refer to SQL Server Books Online for more information on SSB.
ALTER DATABASE <your db name here>
USE <your db name here>;
CREATE MESSAGE TYPE
VALIDATION = WELL_FORMED_XML;
CREATE CONTRACT [//SqlAdapterSSBSample/SampleContract]
SENT BY INITIATOR
CREATE QUEUE InitiatorQueue1DB;
ON QUEUE InitiatorQueue1DB;
CREATE QUEUE TargetQueue1DB;
ON QUEUE TargetQueue1DB
At this point, we’ve created the main database objects required for the SSB side of things.
Creating the BizTalk Artifacts
- Start the BizTalk Server 2006 Administration Console.
- Create a new BizTalk application – let’s call it “SSBPollingApplication”.
- Create a new 1-way Receive Port – call it “SqlReceivePort”. Create a new Receive Location – call it “SqlReceiveLocation”.
- Choose the Transport Type (in the Receive Location Configuration dialog) as “WCF-Custom”. Choose “PassThruReceive” for the Receive Pipeline.
- Click configure, the WCF-Custom adapter configuration dialog pops up.
- In the General tab, enter the URI denoting your SQL Server. The format is “mssql://servername/instancename/databasename”. For example, on my machine (since I am using the default instance of SQL Server, the uri I entered is “mssql://localhost//SSBTestDb” (my database is named “SSBTestDb”)).
- In the Binding Tab, choose the binding as “sqlBinding”. In the configuration properties which are displayed below, set these:
pollingIntervalInSeconds = 2 (or whatever you're comfortable with)
polledDataAvailableStatement = SELECT COUNT(*) FROM TargetQueue1DB WITH (NOLOCK)
pollingStatement = (note, multi line statement follows):
DECLARE @DlgHandle UNIQUEIDENTIFIER;
DECLARE @RecvMsg XML;
RECEIVE TOP (1)
@RecvMsg = CAST(message_body as XML)
IF NOT (@DlgHandle IS NULL)
END CONVERSATION @DlgHandle;
SELECT @RecvMsg AS ReceivedMessage;
A brief explanation of the 2 SQL blocks above:
- In the polledDataAvailableStatement, we’re checking if there are any rows in the SSB queue. It is this statement which is going to execute every “pollingIntervalInSeconds” seconds. If a non-zero value is returned, the adapter interprets it to mean that data is available, and only then proceeds to execute the pollingStatement.
- In the pollingStatement, we are selecting (and removing – the RECEIVE syntax removes the message, while SELECT would just peek at it) the first message in the SSB queue. We’re also selecting the conversation handle for that message. Also, if we did indeed successfully pick up the message from the queue (maybe someone got to it first?), we end that specific conversation.
- In the Behavior tab, right click on the ServiceBehavior, and choose “Add Extension”. Add the “sqlAdapterInboundTransactionBehavior” behavior. You can control the transaction isolation level (the default is ReadCommitted). It is this transaction isolation level which will be applied to the DTC transaction spanning the BizTalk Message Box and your SQL Server (from where you’re pulling messages from the SSB queue).
- In the Other tab, choose None for credentials (if you want to use Integrated Security) – or specify a username + password.
- Click OK as many times as required to close the Receive Port configuration dialogs.
- Create a new static 1-way send port named “FileSendPort”. Configure the transport type as FILE, and configure the port to drop messages to a valid folder on your file system. Select PassThruTransmit as the Send pipeline. Also, click on “Filters” in the left pane. Add a filter condition:
- Property = BTS.ReceivePortName
- Value = SqlReceivePort
- Click OK to complete the configuration of the Send Port.
At this point, the configuration of the BizTalk application is complete. Start the application.
We’re now going to send messages to the SSB queue (I’m using SQL Server Management Studio – you could also use osql.exe). I used the following SQL block to send a message to the queue:
DECLARE @RequestMsg XML;
SELECT @RequestMsg = N'<TestMessage>Hello, World</TestMessage>';
DECLARE @DlgHandle UNIQUEIDENTIFIER;
BEGIN DIALOG @DlgHandle
WITH ENCRYPTION = OFF;
SEND ON CONVERSATION @DlgHandle
Once you send the above message to the SSB queue, within a short while, you should see it in the folder specified in the FileSendPort. The message I see is:
<Polling xmlns=" http://schemas.microsoft.com/Sql/2008/05/Polling/" >
<DataSet xmlns=" http://schemas.datacontract.org/2004/07/System.Data" >
<xs:schema id="NewDataSet" xmlns:xs=" http://www.w3.org/2001/XMLSchema" xmlns:msdata="urn:schemas-microsoft-com:xml-msdata">
<xs:element msdata:IsDataSet="true" name="NewDataSet">
<xs:element minOccurs="0" maxOccurs="unbounded" name="NewTable">
<xs:element minOccurs="0" name="ReceivedMessage" type="xs:string"/>
As can be seen above, the adapter has returned the message in System.Data.DataSet format. You could ofcourse use an XPath query to extract only the body of the message. Here’s how -
- Navigate back to the SqlReceivePort configuration. Bring up the WCF-Custom configuration dialog (the dialog which contained the General tab, the Binding tab, the Behavior tab, etc).
- Navigate to the Messages Tab.
- In the “Inbound BizTalk Message Body” section
- Select “Path”. Enter this XPath query: /*[local-name()='Polling']/*[local-name()='PolledData']/*[local-name()='DataSet']/*[local-name()='diffgram']/*[local-name()='NewDataSet']/*[local-name()='NewTable']/*[local-name()='ReceivedMessage']
- For the Node Encoding, select “String”.
The data in the file should now just be: