ssbsqlcommandexec.cs

SQL Server 2005 Service Broker (SSB) is an asynchronous messaging technology built into SQL Server.

using System;
using System.Collections.Generic;
using System.Text;
using System.Data.SqlClient;
using System.Data.SqlTypes;
using System.Data;
using System.Transactions;

namespace SSBExternalActivationRouter
{
    class SSBSQLCommandExec
    {
        string _queueName = "TestSSB_TimeBillingQueue";
        SqlConnection _connection = null;
        string _connectString = "";
        string _currentConversation;

        public void Open(string connString)
        {
            _connectString = connString;

            _connection = new SqlConnection(_connectString);

            _connection.Open();
        }

        public SqlDataReader PeekAtQueue()
        {
            SqlDataReader dr = null;
            SqlCommand cmd;

            cmd = BuildPeekCommand();

            dr = cmd.ExecuteReader();

            dr.Read();

            return (dr);
        }

        public SqlDataReader GetNextMessage()
        {
            SqlDataReader dr = null;
            SqlCommand cmd;

            _connection.EnlistTransaction(Transaction.Current);

            cmd = BuildReceiveCommand();

            dr = cmd.ExecuteReader();

            dr.Read();

            _currentConversation = dr["conversation_handle"].ToString();

            return (dr);
        }

        public void EndCurrentConversation()
        {
            SqlCommand cmd;

            cmd = BuildEndCommand(_currentConversation);
            cmd.ExecuteNonQuery();
        }

        //This code was borrowed from the ServiceBrokerInterface sample
        private SqlCommand BuildReceiveCommand(  )//,Guid convHandle)
        {
            SqlCommand cmd = _connection.CreateCommand();
            StringBuilder query = new StringBuilder();

                query.Append("WAITFOR(");
            query.Append("RECEIVE ");

                query.Append("TOP(" + "1" + ") ");

            query.Append("conversation_group_id, conversation_handle, " +
                         "message_sequence_number, service_name, service_contract_name, " +
                         "message_type_name, validation, message_body " +
                         "FROM ");
            query.Append(_queueName);
                //query.Append(" WHERE conversation_handle = @ch");
                //param = cmd.Parameters.Add("@ch", SqlDbType.UniqueIdentifier);
                //param.Value = convHandle;

                query.Append("), TIMEOUT 2000");
                cmd.CommandTimeout = 0;

            cmd.CommandText = query.ToString();
            return cmd;
        }

        private SqlCommand BuildEndCommand (  string convHandle )
        {
            SqlParameter param;
			SqlCommand cmd = _connection.CreateCommand();
			cmd.CommandText = "END CONVERSATION @ch";
			param = cmd.Parameters.Add("@ch", SqlDbType.UniqueIdentifier);
            param.Value = convHandle;

            return (cmd);

        }

        private SqlCommand BuildPeekCommand()
        {
            SqlCommand cmd = _connection.CreateCommand();
            cmd.CommandText = "select COUNT(*) MsgCount from " + _queueName;

            return (cmd);
        }

    }
}