Simple MassTransit ServiceBus Sample

Following my post on NServiceBus, here I will show you how to get up to speed with MassTransit. Using a similar scenario with NServiceBus.

One thisng I observed withe MassTransit is it is the initialization of the service is dependent on the Dependency Injection framework. when you download the library it provides you with several choice like using Castle, Unity, Ninject, StructureMap. In this sample I will be using the StructureMap.

To begin, we need to create a class that inherits from MassTransitRegistryBase, this is only applicable if you are using StructureMap and your DI. This class contains the necessary initialization of your MassTransitService.

using MassTransit.Configuration;
using MassTransit.Services.Routing.Configuration;
using MassTransit.StructureMapIntegration;
using MassTransit.Transports.Msmq;

namespace Relativity
{
    /// <summary>
    /// 
    /// </summary>
    public class ServiceBusRegistry : MassTransitRegistryBase
    {
        /// <summary>
        /// 
        /// </summary>
        public const string MQPATH = "msmq://localhost/relativity";

        /// <summary>
        /// 
        /// </summary>
        public ServiceBusRegistry()
            : base(typeof(MsmqEndpoint))
        {
            RegisterServiceBus(MQPATH, ConfigHandler);

            MsmqEndpointConfigurator.Defaults(config => { config.CreateMissingQueues = true; });
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="e"></param>
        private void ConfigHandler(IServiceBusConfigurator e)
        {
            //e.ConfigureService<RoutingConfigurator>(rs => rs.Route<RelativityMessage>().To(MQPATH));
            //e.ConfigureService<RoutingConfigurator>(rs => rs.Route<RelativityResponseMessage>().To(MQPATH));
        }
    }
}

After this has been defined, you may now get the instance of the Bus using the MassTransit Container.

using MassTransit;
using StructureMap;
using System;
using System.Linq;
using System.Windows;

namespace Relativity
{
    /// <summary>
    /// Interaction logic for App.xaml
    /// </summary>
    public partial class App : Application
    {
        private Container _container;
        /// <summary>
        /// 
        /// </summary>
        /// <returns></returns>
        public IServiceBus GetServiceBus()
        {
            if (_container == null)
                throw new NullReferenceException("Service has not properly initialized");
            return _container.GetInstance<IServiceBus>();
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="e"></param>
        protected override void OnStartup(StartupEventArgs e)
        {
            base.OnStartup(e);
            _container = new Container(ConfigHandler);

            IServiceBus bus = GetServiceBus();
            bus.Subscribe<RelativityMessage>(RelativityMessageConsumer);
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="e"></param>
        private void ConfigHandler(ConfigurationExpression e)
        {
            e.AddRegistry(new ServiceBusRegistry());
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="message"></param>
        public static void RelativityMessageConsumer(RelativityMessage message)
        {
            //System.Threading.Thread.Sleep(3000);
            App app = Application.Current as App;
            if (app == null)
                throw new NullReferenceException("Service has not properly initialized");

            IServiceBus bus = app.GetServiceBus();
            bus.Publish<RelativityResponseMessage>(new RelativityResponseMessage() { Text = message.Text + " Acknowledged!" });
        }
    }
}

To publish / subscribe messages

using MassTransit;
using System;
using System.Windows;

namespace Relativity
{
    /// <summary>
    /// Interaction logic for MainWindow.xaml
    /// </summary>
    public partial class MainWindow : Window
    {
        /// <summary>
        /// 
        /// </summary>
        public MainWindow()
        {
            InitializeComponent();
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void OnLoaded(object sender, RoutedEventArgs e)
        {
            App app = Application.Current as App;
            if (app == null)
                throw new NullReferenceException("Service has not properly initialized");

            IServiceBus bus = app.GetServiceBus();
            bus.Subscribe<RelativityResponseMessage>(RelativityResponseMessageConsumer);
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void Send(object sender, RoutedEventArgs e)
        {
            App app = Application.Current as App;
            if (app == null)
                throw new NullReferenceException("Service has not properly initialized");

            IServiceBus bus = app.GetServiceBus();
            bus.Publish<RelativityMessage>(new RelativityMessage() { Text = "Hello world!" });
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="message"></param>
        private void RelativityResponseMessageConsumer(RelativityResponseMessage message)
        {
            MessageBox.Show(message.Text);
        }
    }
}

That’s it

Advertisement