A few days ago, during the BizTalk Innovation Day held in Norway, I had the opportunity to meet with several BizTalk developers and speak about the integration with database systems.

Some of them have asked me to send the examples I have shown during the labs.

This article refers to integration with Microsoft SQL Server, but the concept is the same when you use a different platform. I used the same approach integrating Oracle and DB2.

The scenario illustrated in this example is extremely simple; it is an export from a database table to a generic target system.

Database entities

When you integrate a database system, you need to use additional information to maintain the transaction status and other attributes discussed later in this article.

There are two different approaches you can use during database design to add these information. You can extend current entities or implement a separated entity that references the data table.

The second one, allow you to delete old transaction maintaining the table small and faster to query. In case you require a transaction history both approach are similar.

For simplicity, we will implement the first scenario.

Following is described the table structure.

In the following table are detailed all the transaction information fields.

Database Procedures

The figure below shows the approach in general.

Table Source Code

CREATE TABLE [dbo].[SourceData](
    [sourceDataId] [int] IDENTITY(1,1) NOT NULL,
    [attribute1] [varchar](10) NULL,
    [attribute2] [varchar](10) NULL,
    [attribute3] [int] NULL,
    [transactionId] [uniqueidentifier] NULL,
    [transactionDate] [datetime] NULL,
    [transactionStatus] [int] NULL,
    [transactionRetryCount] [int] NULL,
    CONSTRAINT [PK_SourceData] PRIMARY KEY CLUSTERED 
    (
        [sourceDataId] ASC
    )
    WITH (
            PAD_INDEX = OFF, 
            STATISTICS_NORECOMPUTE = OFF, 
            IGNORE_DUP_KEY = OFF, 
            ALLOW_ROW_LOCKS = ON, 
            ALLOW_PAGE_LOCKS = ON
        ) ON [PRIMARY]
) ON [PRIMARY]
GO

Constants Definition Source Code

Transaction Status Constants:

CREATE FUNCTION [dbo].[StatusToProcess]()
RETURNS INT
AS
BEGIN
    RETURN 0;
END
GO

CREATE FUNCTION [dbo].[StatusInProcess]()
RETURNS INT
AS
BEGIN
    RETURN 1;
END
GO

CREATE FUNCTION [dbo].[StatusSucceded]()
RETURNS INT
AS
BEGIN
    RETURN 2;
END
GO

CREATE FUNCTION [dbo].[StatusSuccededWithWarnings]()
RETURNS INT
AS
BEGIN
    RETURN 3;
END
GO

CREATE FUNCTION [dbo].[StatusFailed]()
RETURNS INT
AS
BEGIN
    RETURN 4;
END
GO

CREATE FUNCTION [dbo].[StatusFailedRetryNotAllowed]()
RETURNS INT
AS
BEGIN
    RETURN 5;
END
GO

Retry Logic Constants:

CREATE FUNCTION [dbo].[MaxRetryCount]()
RETURNS INT
AS
BEGIN
    RETURN 3;
END
GO

CREATE FUNCTION [dbo].[RetryInterval]()
RETURNS INT
AS
BEGIN
    RETURN 1;
END
GO

Views Source Code

The available statement and the polling statement, have in common the logic that determines which rows are to be processed.

Sometimes it can be quite complex so it is advisable define it into a view just to centralize.

CREATE VIEW [dbo].[SourceDataToProcess] AS

SELECT
    SourceData.sourceDataId

FROM
    SourceData

WHERE
    transactionStatus = dbo.StatusToProcess()
    OR
    (
        transactionStatus = dbo.StatusFailed()
        AND ISNULL(transactionRetryCount, 0) <= dbo.MaxRetryCount()
        AND GETDATE() > DATEADD(MINUTE, dbo.RetryInterval(), ISNULL(transactionDate, '01/01/2000'))
    );

Stored Procedures Source Code

From my point of view, the database is a service, for that reason, the entry point must be implemented exclusively through stored procedures.

Available Statement Procedure:

CREATE PROCEDURE [dbo].[SourceDataAvailableStmt]
AS
BEGIN

    SET NOCOUNT ON;
    SET XACT_ABORT ON;
    SET TRANSACTION ISOLATION LEVEL READ COMMITTED;

    SELECT
        COUNT(*)
    FROM
        SourceDataToProcess;

END

Polling Statement Procedure:

CREATE PROCEDURE [dbo].[SourceDataPollingStmt]
AS
BEGIN

    SET NOCOUNT ON;
    SET XACT_ABORT ON;
    SET TRANSACTION ISOLATION LEVEL READ COMMITTED;

    DECLARE @transactionId UNIQUEIDENTIFIER;

    SET @transactionId = NEWID();

    -- Lock processing records
    UPDATE SourceData
    SET
        transactionId = @transactionId,
        transactionDate = GETDATE(),
        transactionStatus = dbo.StatusInProcess()
    FROM
        SourceData

        INNER JOIN SourceDataToProcess
        ON SourceData.sourceDataId = SourceDataToProcess.sourceDataId;

    -- Return data
    SELECT     
        transactionId = CONVERT(CHAR(36), @transactionId),
        sourceDataId,
        attribute1,
        attribute2,
        attribute3

    FROM         
        SourceData WITH (NOLOCK)

    WHERE
        transactionId = @transactionId;

END

Update Status Procedure:

This procedure is used to set the final state of the transaction at the end of workflow integration.

CREATE PROCEDURE [dbo].[SourceDataUpdateStatus]
    @sourceDataId INT,
    @transactionStatus INT
AS
BEGIN

    SET NOCOUNT ON;
    SET XACT_ABORT OFF;
    SET TRANSACTION ISOLATION LEVEL READ COMMITTED;

    BEGIN TRY

        UPDATE SourceData
        SET
            transactionStatus = @transactionStatus,
            transactionRetryCount = CASE
                            WHEN @transactionStatus IN (dbo.StatusSucceded(), dbo.StatusSuccededWithWarnings()) THEN transactionRetryCount
                            WHEN @transactionStatus = dbo.StatusFailed() AND ISNULL(transactionRetryCount, 0) < dbo.MaxRetryCount() THEN ISNULL(transactionRetryCount, 0) + 1
                            ELSE 100
                            END,
            transactionDate = GETDATE()
        WHERE
            sourceDataId = @sourceDataId;

    END TRY
    BEGIN CATCH

        DECLARE @ERROR_NUMBER INT;
        DECLARE @ERROR_MESSAGE NVARCHAR(4000);
        DECLARE @ERROR_SEVERITY INT;
        DECLARE @ERROR_STATE INT;
        DECLARE @procedureName SYSNAME;

        SELECT
            @ERROR_NUMBER = ERROR_NUMBER(),
            @ERROR_MESSAGE = ERROR_MESSAGE(),
            @ERROR_SEVERITY = ERROR_SEVERITY(),
            @ERROR_STATE = ERROR_STATE(),
            @procedureName = ISNULL('Sql Procedure: "' + DB_NAME() + '.' + OBJECT_SCHEMA_NAME(@@PROCID) + '.' + OBJECT_NAME(@@PROCID) + '". ', '');

        SET @ERROR_MESSAGE = @procedureName + ' Error Detail: ' + @ERROR_MESSAGE;
        RAISERROR(@ERROR_MESSAGE, @ERROR_SEVERITY, @ERROR_STATE);

    END CATCH

END

BizTalk Process

The following figure shows the integration process.

The transaction result status is initialized to “Success”. In the case in which an error occurs it will set to “Failed”.

When you query databases that contain data structures in the form header-detail using statement of the type “select * from header inner join detail on …”, typically you obtain datasets that is difficult to map to hierarchical xml structures if not using custom XSLT.

Using a SQL Server statement as shown below:

You get a schema similar to the following:

But often it must be mapped to a form of the type shown below:

The only way I know of to do this, is to use a map specifying a custom XSLT. This is an excellent solution if the mapping rules for individual fields are simple. If you need to deploy complex maps the XSLT could become difficult to implement and maintain.

Some time ago, Nino Crudele and I have implemented a custom pipeline component that brings together the functionality of the old SQL Adapter with the advantages of WCF-SQL Adapter.

Summarizing what allows you to make this custom pipeline component is:

1. You call a stored procedure that performs the query;

2. The query contains FOR XML AUTO or in any case it returns the data in XML form;

3. The pipeline extracts the XML string from the return message and engages in a specified schema;

4. The new scheme can be used as the source of the map.

Creating the stored procedure

Generating the schemas

You have to generate the schema for the WCF-SQL port and the hierarchical XML schema. To generate the first schema use the consume adapter service wizard.

This wizard generates the following schema:

“XML_F52E2B61-18A1-11d1-B105-00805F49916B” is the field that will contains the XML returned by the stored procedure.

Now you have to generate the schema for the XML message. To do that you can use the old SQL Adapter.

So, temporary add “, xmldata” at the end of the stored procedure (the statement became “for xml auto, elements, xmldata”). You have to remove it at the end of the wizard.

In your project add generated item and select “Add Adapter Metadata”, then select “SQL”

Specify a connection string and click Next. In the SQL Transport Schema Generation Wizard page, specify “Receive Port”, a target namespace and a root name.

Note that the FOR XML AUTO clause does not return the root node element. This is why the wizard is asking you to specify it.

Click Next and select “Stored Procedure”

Select you stored procedure and click on Generate.

Then click Next to complete the wizard. What you get is the following schema:

Now you have to clear the Target Namespace property for this schema.

Regarding the schema of the stored procedure is all that is needed. Remember that you have to use the schema generated by the WCF-SQL adapter for the send operation to the database, and the schema generated by the SQL Adapter for receiving the response.

Creating the Pipeline

To complete development activities, it is need to create a receive pipeline that includes the custom pipeline component as shown in the following picture.

Port configuration

From the administration console, import the binding file generated from the WCF-SQL adapter wizard during the schema generation.

Change the receive pipeline, specifying what developed in the previous chapter. In the property of the receive pipeline, specify the namespace, the root element of the SQL adapter generated schema and the XPath query of the element of the schema WCF-SQL that contains the XML result as shown in the follow.

You can find the XPath query in the WCF-SQL schema.

Source code

Following you can find the custom pipeline component source code.

using System;
using System.Text;
using System.Xml;
using System.IO;
using Microsoft.BizTalk.Message.Interop;
using Microsoft.BizTalk.Component.Interop;
using System.Xml.XPath;
using System.ComponentModel;
using System.Diagnostics;

namespace Sample.CustomPipeline
{
    [ComponentCategory(CategoryTypes.CATID_PipelineComponent)]
    [ComponentCategory(CategoryTypes.CATID_Decoder)]
    [System.Runtime.InteropServices.Guid("9d0e4103-4cce-4536-83fa-4a5040674ad6")]
    public class WcfSqlXmlPipelineComponent : IBaseComponent, IComponentUI, Microsoft.BizTalk.Component.Interop.IComponent, IPersistPropertyBag
    {

        #region properties

        private string wcfXmlXPath;

        [System.ComponentModel.Description("Specify the XPath query of the WCF XML node data.")]
        public string WcfXmlXPath
        {
            get { return wcfXmlXPath; }
            set { wcfXmlXPath = value; }
        }

        private string newNameSpace;

        [System.ComponentModel.Description("Namespace to be assigned.")]
        public string NewNameSpace
        {
            get { return newNameSpace; }
            set { newNameSpace = value; }
        }

        private string rootName;

        [System.ComponentModel.Description("XML Root name to be assigned.")]
        public string RootName
        {
            get { return rootName; }
            set { rootName = value; }
        }

        #endregion

        #region IBaseComponent Members

        [Browsable(false)]
        public string Description
        {
            get
            {
                return "Pipeline component used to return a specific XML SQL statement";
            }
        }

        [Browsable(false)]
        public string Name
        {
            get { return "WcfSqlXmlPipelineComponent"; }
        }

        [Browsable(false)]
        public string Version
        {
            get { return "1.0.0.0"; }
        }

        #endregion

        #region IPersistPropertyBag Members

        public void GetClassID(out Guid classID)
        {
            classID = new Guid("0C86A664-3D33-4E70-99AE-E130D9D8620C");
        }

        public void InitNew()
        {

        }

        public void Load(IPropertyBag pb, int errorLog)
        {

            object val;
            val = ReadPropertyBag(pb, "WcfXmlXPath");
            if ((val != null))
            {
                wcfXmlXPath = ((string)(val));
            }

            val = ReadPropertyBag(pb, "NewNameSpace");
            if ((val != null))
            {
                newNameSpace = ((string)(val));
            }

            val = ReadPropertyBag(pb, "RootName");
            if ((val != null))
            {
                rootName = ((string)(val));
            }
        }

        public void Save(IPropertyBag pb, bool clearDirty, bool saveAllProperties)
        {
            WritePropertyBag(pb, "WcfXmlXPath", WcfXmlXPath);
            WritePropertyBag(pb, "NewNameSpace", NewNameSpace);
            WritePropertyBag(pb, "RootName", RootName);
        }

        #region utility functionality

        /// <summary>
        /// Reads property value from property bag
        /// </summary>
        /// <param name="pb">Property bag</param>
        /// <param name="propName">Name of property</param>
        /// <returns>Value of the property</returns>
        private object ReadPropertyBag(IPropertyBag pb, string propName)
        {
            object val = null;
            try
            {
                pb.Read(propName, out val, 0);
            }
            catch (ArgumentException)
            {
                return val;
            }
            catch (Exception e)
            {
                throw new ApplicationException(e.Message);
            }
            return val;
        }

        /// <summary>
        /// Writes property values into a property bag.
        /// </summary>
        /// <param name="pb">Property bag.</param>
        /// <param name="propName">Name of property.</param>
        /// <param name="val">Value of property.</param>
        private void WritePropertyBag(IPropertyBag pb, string propName, object val)
        {
            try
            {
                pb.Write(propName, ref val);
            }
            catch (Exception e)
            {
                throw new ApplicationException(e.Message);
            }
        }
        #endregion

        #endregion

        #region IComponentUI Members

        /// <summary>
        /// Component icon to use in BizTalk Editor
        /// </summary>
        [Browsable(false)]
        public IntPtr Icon
        {
            get { return new IntPtr(); }
        }

        System.Collections.IEnumerator IComponentUI.Validate(object projectSystem)
        {
            return null;
        }

        #endregion

        #region IComponent Members

        public IBaseMessage Execute(IPipelineContext pContext, IBaseMessage pInMsg)
        {

            string systemPropertiesNamespace = @"http://schemas.microsoft.com/BizTalk/2003/system-properties";
            string messageType = string.Concat(newNameSpace, "#", rootName);
            StringBuilder outMessage = new StringBuilder();

            try
            {

                Debug.Write("Execute started");

                IBaseMessagePart bodyPart = pInMsg.BodyPart;

                if (bodyPart != null)
                {

                    Stream messageInputStream = bodyPart.GetOriginalDataStream();
                    string xmlMessage = ExtractDataValueXPath(messageInputStream, wcfXmlXPath);

                    outMessage.Append(string.Format("<{0}>", rootName));
                    outMessage.Append(xmlMessage);
                    outMessage.Append(string.Concat("</", rootName, ">"));

                    Debug.Write(String.Concat("messageType: ", messageType));
                    Debug.Write(String.Concat("rootName: ", rootName));
                    Debug.Write(String.Concat("wcfXmlXPath: ", wcfXmlXPath));
                    Debug.Write(String.Concat("xmlMessage: ", xmlMessage));

                    byte[] outBytes = System.Text.Encoding.ASCII.GetBytes(outMessage.ToString());

                    MemoryStream memStream = new MemoryStream();
                    memStream.Write(outBytes, 0, outBytes.Length);
                    memStream.Position = 0;
                    bodyPart.Data = memStream;
                    pContext.ResourceTracker.AddResource(memStream);

                }

                pInMsg.Context.Promote("MessageType", systemPropertiesNamespace, messageType);

            }
            catch (Exception exc)
            {
                Debug.Write(exc.Message);
                throw;
            }
            finally
            {
                Debug.Write("Execute finished");
            }

            return pInMsg;

        }

        #endregion

        private string ExtractDataValueXPath(Stream MsgStream, string MsgXPath)
        {

            XmlReaderSettings settings = new XmlReaderSettings()
            {

                ConformanceLevel = ConformanceLevel.Document,
                IgnoreWhitespace = true,
                ValidationType = ValidationType.None,
                IgnoreProcessingInstructions = true,
                IgnoreComments = true,
                CloseInput = false
            };

            MsgStream.Seek(0, SeekOrigin.Begin);
            XmlReader reader = XmlReader.Create(MsgStream, settings);
            StringBuilder strValue = new StringBuilder();

            if (!string.IsNullOrEmpty(MsgXPath))
            {
                if (reader.Read())
                {
                    XPathDocument xPathDoc = new XPathDocument(reader);
                    XPathNavigator xNavigator = xPathDoc.CreateNavigator();
                    XPathNodeIterator xNodes = xNavigator.Select(MsgXPath);
                    while (xNodes.MoveNext())
                    {
                        strValue.Append(xNodes.Current.Value);
                    }
                    MsgStream.Seek(0, SeekOrigin.Begin);
                }
            }

            return strValue.ToString();
        }

    }
}