Receive Polling-based Data-changed Messages from SQL Server Using the WCF Service Model

You can configure the SQL adapter to receive periodic data-change messages for SQL Server tables or views. You can specify a polling statement that the adapter executes to poll the database. The polling statement can be a SELECT statement or a stored procedure that returns a result set.

For more information on how the adapter supports polling, see Polling in SQL Server using the SQL adapter.

How This Topic Demonstrates Polling

In this topic, to demonstrate how the SQL adapter supports receiving data change messages, create a .NET application and generate the WCF service contract for the Polling operation. If you want to specify the polling related binding properties while generating the WCF service contract, specify the PolledDataAvailableStatement as:

SELECT COUNT(*) FROM Employee

The PolledDataAvailableStatement must return a result set with the first cell containing a positive value. If the first cell does not contain a positive value, the adapter does not execute the polling statement.

As part of the polling statement, perform the following operations:

1. Select all the rows from the Employee table.

2. Execute a stored procedure (MOVE_EMP_DATA) to move all the records from the Employee table to an EmployeeHistory table.

3. Execute a stored procedure (ADD_EMP_DETAILS) to add a new record to the Employee table. This procedure takes the employee name, designation, and salary as parameters.

   To perform these operations, you must specify the following for the PollingStatement binding property:

SELECT * FROM Employee;EXEC MOVE_EMP_DATA;EXEC ADD_EMP_DETAILS John, Tester, 100000

After the polling statement is executed, all the records from the Employee table are selected and the message from SQL Server is received. After the MOVE_EMP_DATA stored procedure is executed by the adapter, all the records are moved to the EmployeeHistory table. Then, the ADD_EMP_DETAILS stored procedure is executed to add a new record to the Employee table. The next polling execution will only return a single record. This cycle continues until you close the service host.

Configuring a Polling Query with the SQL Adapter Binding Properties

The following table summarizes the SQL adapter binding properties that you use to configure the adapter to receive data-change messages. You must specify these binding properties as part of the .NET application for polling.

For a more complete description of these properties, see Read about the BizTalk Adapter for SQL Server adapter binding properties. For a complete description of how to use the SQL adapter to poll SQL Server, read further.

Configuring Polling in the WCF Service Model

To receive the Polling operation when you use the WCF service model, you must:

1. Generate a WCF service contract (interface) for the Polling operation from the metadata exposed by the adapter. To do this, you could use the Add Adapter Service Reference Visual Studio Plug-in.

2. Implement a WCF service from this interface.

3. Host this WCF service using a service host (System.ServiceModel.ServiceHost).

About the Examples Used in this Topic

The examples in this topic poll the Employee table. The example also uses the MOVE_EMP_DATA and ADD_EMP_DETAILS stored procedure. A script to generate these artifacts is supplied with the samples. For more information about the samples, see Samples for the SQL adapter. A sample, Polling_ServiceModel, which is based on this topic, is also provided with the SQL adapter samples.

The WCF Service Contract and Class

You can use the Add Adapter Service Reference Plug-in to create a WCF service contract (interface) and supporting classes for the Polling operation. For more information about generating a WCF service contract, see Generate a WCF Client or WCF Service Contract for SQL Server Artifacts.

The WCF Service Contract (Interface)

The following code shows the WCF service contract (interface) generated for the Polling operation.

[System.CodeDom.Compiler.GeneratedCodeAttribute("System.ServiceModel", "3.0.0.0")]
[System.ServiceModel.ServiceContractAttribute(Namespace="http://schemas.microsoft.com/Sql/2008/05/", ConfigurationName="PollingOperation")]
public interface PollingOperation {

    // CODEGEN: Generating message contract since the wrapper namespace (https://schemas.microsoft.com/Sql/2008/05/Polling/) of message Polling
    // does not match the default value (https://schemas.microsoft.com/Sql/2008/05/)
    [System.ServiceModel.OperationContractAttribute(IsOneWay=true, Action="Polling")]
    [System.ServiceModel.XmlSerializerFormatAttribute()]
    void Polling(Polling request);
}

The Message Contracts

The message contract namespace is modified by the InboundID parameter in the connection URI, if specified. In this example, you did not specify an inbound ID in the connection URI. The request message returns a DataSet.

[System.Diagnostics.DebuggerStepThroughAttribute()]
[System.CodeDom.Compiler.GeneratedCodeAttribute("System.ServiceModel", "3.0.0.0")]
[System.ServiceModel.MessageContractAttribute(WrapperName="Polling", WrapperNamespace="http://schemas.microsoft.com/Sql/2008/05/Polling/", IsWrapped=true)]
public partial class Polling {

[System.ServiceModel.MessageBodyMemberAttribute(Namespace="http://schemas.microsoft.com/Sql/2008/05/Polling/", Order=0)]
    [System.Xml.Serialization.XmlArrayAttribute(IsNullable=true)]
    [System.Xml.Serialization.XmlArrayItemAttribute("DataSet", Namespace="http://schemas.datacontract.org/2004/07/System.Data", IsNullable=false)]
    public System.Data.DataSet[] PolledData;

    public Polling() {
    }

    public Polling(System.Data.DataSet[] PolledData) {
        this.PolledData = PolledData;
    }
}

WCF Service Class

The Add Adapter Service Reference Plug-in also generates a file that has a stub for the WCF service class implemented from the service contract (interface). The name of the file is SqlAdapterBindingService.cs. You can insert the logic to process the Polling operation directly into this class. The following code shows the WCF service class generated by the Add Adapter Service Reference Plug-in.

namespace SqlAdapterBindingNamespace {

    public class SqlAdapterBindingService : PollingOperation {

        // CODEGEN: Generating message contract since the wrapper namespace (https://schemas.microsoft.com/Sql/2008/05/Polling/) of message Polling
        // does not match the default value (https://schemas.microsoft.com/Sql/2008/05/)
        public virtual void Polling(Polling request) {
            throw new System.NotImplementedException("The method or operation is not implemented.");
        }
    }
}

Receiving Inbound Messages for Polling Operation

This section provides instructions on how to write a .NET application to receive inbound polling messages using the SQL adapter.

To receive polling messages from the SQL adapter

1. Use the Add Adapter Service Reference Plug-in to generate a WCF service contract (interface) and helper classes for the Polling operation. For more information, see Generate a WCF Client or WCF Service Contract for SQL Server Artifacts. You can optionally specify the binding properties while generating the service contract and helper classes. This guarantees that they are properly set in the generated configuration file.

2. Implement a WCF service from the interface and helper classes generated in step 1. The Polling method of this class can throw an exception to abort the polling transaction, if an error is encountered processing the data received from the Polling operation; otherwise the method does not return anything. You must attribute the WCF service class as follows:

   [ServiceBehavior(InstanceContextMode = InstanceContextMode.Single)]
   

   Within the Polling method, you can implement your application logic directly. This class can be found in SqlAdapterBindingService.cs. This code in this example sub-classes the SqlAdapterBindingService class. In this code, the polling message received as a DataSet is written to the console.

   [ServiceBehavior(InstanceContextMode = InstanceContextMode.Single)]
   
   public class PollingService : SqlAdapterBindingNamespace.SqlAdapterBindingService
   {
   
   public override void Polling(Polling request)
   {
       Console.WriteLine("\nNew Polling Records Received");
       Console.WriteLine("*************************************************");
       DataSet[] dataArray = request.PolledData;
       foreach (DataTable tab in dataArray[0].Tables)
       {
           foreach (DataRow row in tab.Rows)
           {
               for (int i = 0; i < tab.Columns.Count; i++)
               {
                   Console.WriteLine(row[i]);
               }
           }
       }
       Console.WriteLine("*************************************************");
       Console.WriteLine("\nHit <RETURN> to stop polling");
       }
   }
   

3. Because the SQL adapter does not accept credentials as part of the connection URI, you must implement the following class to pass credentials for the SQL Server database. In the latter part of the application, you will instantiate this class to pass on the SQL Server credentials.

   class PollingCredentials : ClientCredentials, IServiceBehavior
   {
       public void AddBindingParameters(ServiceDescription serviceDescription, ServiceHostBase serviceHostBase, Collection<ServiceEndpoint> endpoints, BindingParameterCollection bindingParameters)
       {
           bindingParameters.Add(this);
       }
   
       public void ApplyDispatchBehavior(ServiceDescription serviceDescription, ServiceHostBase serviceHostBase)
       { }
   
       public void Validate(ServiceDescription serviceDescription, ServiceHostBase serviceHostBase)
       { }
   
       protected override ClientCredentials CloneCore()
       {
           ClientCredentials clone = new PollingCredentials();
           clone.UserName.UserName = this.UserName.UserName;
           clone.UserName.Password = this.UserName.Password;
           return clone;
       }
   }
   

4. Create a SqlAdapterBinding and configure the polling operation by specifying the binding properties. You can do this either explicitly in code or declaratively in configuration. At a minimum, you must specify the InboundOperationType, PolledDataAvailableStatement, and PollingStatement.

   SqlAdapterBinding binding = new SqlAdapterBinding();
   binding.InboundOperationType = InboundOperation.Polling;
   binding.PolledDataAvailableStatement = "SELECT COUNT (*) FROM EMPLOYEE";
   binding.PollingStatement = "SELECT * FROM Employee;EXEC MOVE_EMP_DATA;EXEC ADD_EMP_DETAILS John, Tester, 100000";
   

5. Specify SQL Server database credentials by instantiating the PollingCredentials class you created in Step 3.

   PollingCredentials credentials = new PollingCredentials();
   credentials.UserName.UserName = "<Enter user name here>";
   credentials.UserName.Password = "<Enter password here>";
   

6. Create an instance of the WCF service created in step 2.

   // create service instance
   PollingService service = new PollingService();
   

7. Create an instance of System.ServiceModel.ServiceHost by using the WCF service and a base connection URI. The base connection URI cannot contain the inbound ID, if specified. You should also specify the credentials here.

   // Enable service host
   Uri[] baseUri = new Uri[] { new Uri("mssql://mysqlserver//mydatabase") };
   ServiceHost serviceHost = new ServiceHost(service, baseUri);
   serviceHost.Description.Behaviors.Add(credentials);
   
   

8. Add a service endpoint to the service host. To do this:

   • Use the binding created in step 4.

   • Specify a connection URI that contains credentials and, if required, an inbound ID.

   • Specify the contract as "PollingOperation".

   // Add service endpoint: be sure to specify PollingOperation as the contract
   Uri ConnectionUri = new Uri("mssql://mysqlserver//mydatabase?");
   serviceHost.AddServiceEndpoint("PollingOperation", binding, ConnectionUri);
   

9. To receive polling data, open the service host. The adapter will return data whenever the query returns a result set.

   // Open the service host to begin polling
   serviceHost.Open();
   

10. To terminate polling, close the service host.

    Important

    The adapter will continue to poll until the service host is closed.

    serviceHost.Close();
    

Example

The following example shows a polling query that executes the Employee table. The polling statement performs the following tasks:

1. Selects all the records from the Employee table.

2. Executes the MOVE_EMP_DATA stored procedure to move all records from Employee table to EmployeeHistory table.

3. Executes the ADD_EMP_DETAILS stored procedure to add a single record to the Employee table.

   The first polling message will contain all the records from the Employee table. The subsequent polling messages will contain only the last record inserted by the ADD_EMP_DETAILS stored procedure. The adapter will continue to poll until you close the service host by pressing <RETURN>.

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;

using Microsoft.Adapters.Sql;
using Microsoft.ServiceModel.Channels;
using System.ServiceModel;
using System.ServiceModel.Description;
using System.ServiceModel.Channels;
using System.Collections.ObjectModel;
using System.Data;

namespace Polling_ServiceModel
{
    [ServiceBehavior(InstanceContextMode = InstanceContextMode.Single)]

    public class PollingService : SqlAdapterBindingNamespace.SqlAdapterBindingService
    {

        public override void Polling(Polling request)
        {
            Console.WriteLine("\nNew Polling Records Received");
            Console.WriteLine("*************************************************");
            DataSet[] dataArray = request.PolledData;
            foreach (DataTable tab in dataArray[0].Tables)
            {
                foreach (DataRow row in tab.Rows)
                {
                    for (int i = 0; i < tab.Columns.Count; i++)
                    {
                        Console.WriteLine(row[i]);
                    }
                }
            }
            Console.WriteLine("*************************************************");
            Console.WriteLine("\nHit <RETURN> to stop polling");
        }
    }

    class PollingCredentials : ClientCredentials, IServiceBehavior
    {
        public void AddBindingParameters(ServiceDescription serviceDescription, ServiceHostBase serviceHostBase, Collection<ServiceEndpoint> endpoints, BindingParameterCollection bindingParameters)
        {
            bindingParameters.Add(this);
        }

        public void ApplyDispatchBehavior(ServiceDescription serviceDescription, ServiceHostBase serviceHostBase)
        { }

        public void Validate(ServiceDescription serviceDescription, ServiceHostBase serviceHostBase)
        { }

        protected override ClientCredentials CloneCore()
        {
            ClientCredentials clone = new PollingCredentials();
            clone.UserName.UserName = this.UserName.UserName;
            clone.UserName.Password = this.UserName.Password;
            return clone;
        }
    }

    class Program
    {
        static void Main(string[] args)
        {
            ServiceHost serviceHost = null;
            try
            {
                Console.WriteLine("Sample started...");
                Console.WriteLine("Press any key to start polling...");
                Console.ReadLine();

                SqlAdapterBinding binding = new SqlAdapterBinding();
                binding.InboundOperationType = InboundOperation.Polling;
                binding.PolledDataAvailableStatement = "SELECT COUNT (*) FROM EMPLOYEE";
                binding.PollingStatement = "SELECT * FROM Employee;EXEC MOVE_EMP_DATA;EXEC ADD_EMP_DETAILS John, Tester, 100000";
                Console.WriteLine("Binding properties assigned...");

                // This URI is used to specify the address for the ServiceEndpoint
                // It must contain the InboundId (if any) that was used to generate
                // the WCF service callback interface
                Uri ConnectionUri = new Uri("mssql://mysqlserver//mydatabase?");

                // This URI is used to initialize the ServiceHost. It cannot contain
                // a query_string (InboundID); otherwise,an exception is thrown when
                // the ServiceHost is initialized.
                Uri[] baseUri = new Uri[] { new Uri("mssql://mysqlserver//mydatabase") };

                PollingCredentials credentials = new PollingCredentials();
                credentials.UserName.UserName = "<Enter user name here>";
                credentials.UserName.Password = "<Enter password here>";

                Console.WriteLine("Opening service host...");
                PollingService service = new PollingService();
                serviceHost = new ServiceHost(service, baseUri);
                serviceHost.Description.Behaviors.Add(credentials);
                serviceHost.AddServiceEndpoint("PollingOperation", binding, ConnectionUri);
                serviceHost.Open();
                Console.WriteLine("Service host opened...");
                Console.WriteLine("Polling started...");
                Console.ReadLine();
            }
            catch (Exception e)
            {
                Console.WriteLine("Exception :" + e.Message);
                Console.ReadLine();

                /* If there is an error it will be specified in the inner exception */
                if (e.InnerException != null)
                {
                    Console.WriteLine("InnerException: " + e.InnerException.Message);
                    Console.ReadLine();
                }
            }
            finally
            {
                // IMPORTANT: you must close the ServiceHost to stop polling
                if (serviceHost.State == CommunicationState.Opened)
                    serviceHost.Close();
                else
                    serviceHost.Abort();
            }
        }
    }
}

See Also

Poll SQL Server using the SQL Adapter with WCF Service Model