Read SQL Server Broker messages and publish them u

I am very new to NServiceBus, and in one of our project, we want to accomplish following -

Whenever table data is modified in Sql server, construct a message and insert in sql server broker queue
Read the broker queue message using NServiceBus
Publish the message again as another event so that other subscribers can handle it.

Now it is point 2, that I do not have much clue, how to get it done.

I have referred the following posts, after which I was able to enter the message in broker queue, but unable to integrate with NServiceBus in our project, as the NServiceBus libraries are of older version and also many methods used are deprecated. So using them with current versions is getting very troublesome, or if I was doing it in improper way.

http://www.nullreference.se/2010/12/06/using-nservicebus-and-servicebroker-net-part-2 https://github.com/jdaigle/servicebroker.net

Any help on the correct way of doing this would be invaluable.

Thanks.

回答1:

I'm using the current version of nServiceBus (5), VS2013 and SQL Server 2008. I created a Database Change Listener using this tutorial, which uses SQL Server object broker and SQLDependency to monitor the changes to a specific table. (NB This may be deprecated in later versions of SQL Server).

SQL Dependency allows you to use a broad selection of all the basic SQL functionality, although there are some restrictions that you need to be aware of. I modified the code from the tutorial slightly to provide better error information:

    void NotifyOnChange(object sender, SqlNotificationEventArgs e)
    {
        // Check for any errors
        if (@"Subscribe|Unknown".Contains(e.Type.ToString())) { throw _DisplayErrorDetails(e); }

        var dependency = sender as SqlDependency;
        if (dependency != null) dependency.OnChange -= NotifyOnChange;
        if (OnChange != null) { OnChange(); }
    }

    private Exception _DisplayErrorDetails(SqlNotificationEventArgs e)
    {
        var message = "useful error info";

        var messageInner = string.Format("Type:{0}, Source:{1}, Info:{2}", e.Type.ToString(), e.Source.ToString(), e.Info.ToString());

        if (@"Subscribe".Contains(e.Type.ToString()) && @"Invalid".Contains(e.Info.ToString()))
            messageInner += "\r\n\nThe subscriber says that the statement is invalid - check your SQL statement conforms to specified requirements (http://stackoverflow.com/questions/7588572/what-are-the-limitations-of-sqldependency/7588660#7588660).\n\n";

        return new Exception(messageMain, new Exception(messageInner));

    }

I also created a project with a "database first" Entity Framework data model to allow me do something with the changed data.

[The relevant part of] My nServiceBus project comprises two "Run as Host" endpoints, one of which publishes event messages. The second endpoint handles the messages. The publisher has been setup to IWantToRunAtStartup, which instantiates the DBListener and passes it the SQL statement I want to run as my change monitor. The onChange() function is passed an anonymous function to read the changed data and publish a message:

using statements

namespace Sample4.TestItemRequest
{
    public partial class MyExampleSender : IWantToRunWhenBusStartsAndStops
    {
        private string NOTIFY_SQL = @"SELECT [id] FROM [dbo].[Test] WITH(NOLOCK) WHERE ISNULL([Status], 'N') = 'N'";
    public void Start() { _StartListening(); }
    public void Stop() { throw new NotImplementedException(); }

    private void _StartListening()
    {
        var db = new Models.TestEntities();

        // Instantiate a new DBListener with the specified connection string            
        var changeListener = new DatabaseChangeListener(ConfigurationManager.ConnectionStrings["TestConnection"].ConnectionString);

        // Assign the code within the braces to the DBListener's onChange event
        changeListener.OnChange += () =>
        {
            /* START OF EVENT HANDLING CODE  */

            //This uses LINQ against the EF data model to get the changed records
            IEnumerable<Models.TestItems> _NewTestItems = DataAccessLibrary.GetInitialDataSet(db);

            while (_NewTestItems.Count() > 0)
            {
                foreach (var qq in _NewTestItems)
                {
                    // Do some processing, if required

                    var newTestItem = new NewTestStarted() { ... set properties from qq object ... };
                    Bus.Publish(newTestItem);
                }

                // Because there might be a number of new rows added, I grab them in small batches until finished.
                // Probably better to use RX to do this, but this will do for proof of concept
                _NewTestItems = DataAccessLibrary.GetNextDataChunk(db);

            }

            changeListener.Start(string.Format(NOTIFY_SQL));

            /* END OF EVENT HANDLING CODE  */

        };

        // Now everything has been set up.... start it running.
        changeListener.Start(string.Format(NOTIFY_SQL));

        }
    }
}

Important The OnChange event firing causes the listener to stop monitoring. It basically is a single event notifier. After you have handled the event, the last thing to do is restart the DBListener. (You can see this in the line preceding the END OF EVENT HANDLING comment).

You need to add a reference to System.Data and possibly System.Data.DataSetExtensions.

The project at the moment is still proof of concept, so I'm well aware that the above can be somewhat improved. Also bear in mind I had to strip out company specific code, so there may be bugs. Treat it as a template, rather than a working example.

I also don't know if this is the right place to put the code - that's partly why I'm on StackOverflow today; to look for better examples of ServiceBus host code. Whatever the failings of my code, the solution works pretty effectively - so far - and meets your goals, too.

Don't worry too much about the ServiceBroker side of things. Once you have set it up, per the tutorial, SQLDependency takes care of the details for you.

回答2:

The ServiceBroker Transport is very old and not supported anymore, as far as I can remember. A possible solution would be to "monitor" the interesting tables from the endpoint code using something like a SqlDependency (http://msdn.microsoft.com/en-us/library/62xk7953(v=vs.110).aspx) and then push messages into the relevant queues.