Bus Adapters

.gitignore

@@ -181,3 +181,5 @@ UpgradeLog*.htm
 
 # Microsoft Fakes
 FakesAssemblies/
+
+.vs/

SachaBarber.CQRS.Demo/BusAdapters/SachaBarber.CQRS.Demo.MSQMAdapter/BusSubscriber.cs

@@ -0,0 +1,96 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Reflection;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using System.Reactive.Subjects;
+using System.Configuration;
+using System.Reactive.Linq;
+using System.Messaging;
+using NLog;
+
+namespace SachaBarber.CQRS.Demo.MSMQAdapter
+{
+    public class BusSubscriber : SharedCore.Services.IInterProcessBusSubscriber, IDisposable
+    {
+        private MessageQueue _messageQueue;
+        private readonly string queueName;
+        private CancellationTokenSource cancellationToken;
+        private Task workerTask;
+        private ISubject<string> eventsSubject = new Subject<string>();
+        private IObservable<Message> _obs;
+        private IDisposable _subscription;
+        private Logger logger = LogManager.GetLogger("MSMQAdapter.BusSubscriber");
+
+        public BusSubscriber()
+        {
+            queueName = ConfigurationManager.AppSettings["MSMQueueName"];
+            StartMessageListener();
+        }
+
+        private void StartMessageListener()
+        {
+            cancellationToken = new CancellationTokenSource();
+            workerTask = Task.Factory.StartNew(() => ListenForMessage(), cancellationToken.Token);
+        }
+
+        public void Dispose()
+        {
+            CancelWorkerTask();
+        }
+
+        private void CancelWorkerTask()
+        {
+            if (workerTask == null) return;
+            logger.Info("BusSubscriber task cancelled.", queueName);
+            cancellationToken.Cancel();
+            workerTask.Wait();
+            workerTask.Dispose();
+            _subscription.Dispose();
+            _messageQueue.Dispose();
+            _obs = null;
+        }
+
+        private void ListenForMessage()
+        {
+            logger.Info("Listening on queue {0} for messages.", queueName);
+            _messageQueue = new MessageQueue(queueName);
+            _messageQueue.ReceiveCompleted += ReadQueue;
+            _messageQueue.BeginReceive();
+        }
+
+        private void ReadQueue(object sender, ReceiveCompletedEventArgs args)
+        {
+            var msg = _messageQueue.EndReceive(args.AsyncResult);
+            try
+            {
+                msg.Formatter = new XmlMessageFormatter(new string[] { "System.String,mscorlib" });
+                logger.Info("Message received: {0}", msg.Body.ToString());
+            }
+            catch (Exception e)
+            {
+                logger.Error("Error occurred when message received", e);
+                return;
+            }
+
+            var message = msg.Body.ToString();
+            Task.Run(async () =>
+            {
+                await Task.Run(() =>
+                {
+                    logger.Info("Publishing method to Rx stream");
+                    eventsSubject.OnNext(message);
+                });
+            });
+
+            _messageQueue.BeginReceive();
+        }
+
+        public IObservable<string> GetEventStream()
+        {
+            return eventsSubject.AsObservable();
+        }
+    }
+}

SachaBarber.CQRS.Demo/BusAdapters/SachaBarber.CQRS.Demo.MSQMAdapter/InterProcessBus.cs

@@ -0,0 +1,35 @@
+using System.Configuration;
+using System.Messaging;
+using NLog;
+using SachaBarber.CQRS.Demo.SharedCore.Services;
+
+namespace SachaBarber.CQRS.Demo.MSMQAdapter
+{
+    public class InterProcessBus : IInterProcessBus
+    {
+        private MessageQueue _messageQueue;
+        private readonly string _queueName;
+        private Logger logger = LogManager.GetLogger("MSMQAdapter.InterProcessBus");
+
+        public InterProcessBus()
+        {
+            _queueName = ConfigurationManager.AppSettings["MSMQueueName"];
+
+            logger.Info("Using queue " + _queueName);
+
+            if (!MessageQueue.Exists(_queueName))
+            {
+                MessageQueue.Create(_queueName);
+                logger.Info("Queue created: " + _queueName);
+            }
+
+            _messageQueue = new MessageQueue(_queueName);
+        }
+
+        public void SendMessage(string message)
+        {
+            logger.Info("Sending message '{0}' to queue '{1}' ...", message, _queueName);
+            _messageQueue.Send(message);
+        }
+    }
+}

SachaBarber.CQRS.Demo/BusAdapters/SachaBarber.CQRS.Demo.MSQMAdapter/Properties/AssemblyInfo.cs

@@ -0,0 +1,36 @@
+using System.Reflection;
+using System.Runtime.CompilerServices;
+using System.Runtime.InteropServices;
+
+// General Information about an assembly is controlled through the following 
+// set of attributes. Change these attribute values to modify the information
+// associated with an assembly.
+[assembly: AssemblyTitle("SachaBarber.CQRS.Demo.MSQMAdapter")]
+[assembly: AssemblyDescription("")]
+[assembly: AssemblyConfiguration("")]
+[assembly: AssemblyCompany("")]
+[assembly: AssemblyProduct("SachaBarber.CQRS.Demo.MSQMAdapter")]
+[assembly: AssemblyCopyright("Copyright ©  2015")]
+[assembly: AssemblyTrademark("")]
+[assembly: AssemblyCulture("")]
+
+// Setting ComVisible to false makes the types in this assembly not visible 
+// to COM components.  If you need to access a type in this assembly from 
+// COM, set the ComVisible attribute to true on that type.
+[assembly: ComVisible(false)]
+
+// The following GUID is for the ID of the typelib if this project is exposed to COM
+[assembly: Guid("4068d2da-f0b6-40a8-baa2-f4a93a36c1f6")]
+
+// Version information for an assembly consists of the following four values:
+//
+//      Major Version
+//      Minor Version 
+//      Build Number
+//      Revision
+//
+// You can specify all the values or you can default the Build and Revision Numbers 
+// by using the '*' as shown below:
+// [assembly: AssemblyVersion("1.0.*")]
+[assembly: AssemblyVersion("1.0.0.0")]
+[assembly: AssemblyFileVersion("1.0.0.0")]

SachaBarber.CQRS.Demo/BusAdapters/SachaBarber.CQRS.Demo.MSQMAdapter/SachaBarber.CQRS.Demo.MSQMAdapter.csproj

@@ -0,0 +1,87 @@
+<?xml version="1.0" encoding="utf-8"?>
+<Project ToolsVersion="14.0" DefaultTargets="Build" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
+  <Import Project="$(MSBuildExtensionsPath)\$(MSBuildToolsVersion)\Microsoft.Common.props" Condition="Exists('$(MSBuildExtensionsPath)\$(MSBuildToolsVersion)\Microsoft.Common.props')" />
+  <PropertyGroup>
+    <Configuration Condition=" '$(Configuration)' == '' ">Debug</Configuration>
+    <Platform Condition=" '$(Platform)' == '' ">AnyCPU</Platform>
+    <ProjectGuid>{4068D2DA-F0B6-40A8-BAA2-F4A93A36C1F6}</ProjectGuid>
+    <OutputType>Library</OutputType>
+    <AppDesignerFolder>Properties</AppDesignerFolder>
+    <RootNamespace>SachaBarber.CQRS.Demo.MSQMAdapter</RootNamespace>
+    <AssemblyName>SachaBarber.CQRS.Demo.MSQMAdapter</AssemblyName>
+    <TargetFrameworkVersion>v4.5.1</TargetFrameworkVersion>
+    <FileAlignment>512</FileAlignment>
+    <TargetFrameworkProfile />
+  </PropertyGroup>
+  <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Debug|AnyCPU' ">
+    <DebugSymbols>true</DebugSymbols>
+    <DebugType>full</DebugType>
+    <Optimize>false</Optimize>
+    <OutputPath>bin\Debug\</OutputPath>
+    <DefineConstants>DEBUG;TRACE</DefineConstants>
+    <ErrorReport>prompt</ErrorReport>
+    <WarningLevel>4</WarningLevel>
+  </PropertyGroup>
+  <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Release|AnyCPU' ">
+    <DebugType>pdbonly</DebugType>
+    <Optimize>true</Optimize>
+    <OutputPath>bin\Release\</OutputPath>
+    <DefineConstants>TRACE</DefineConstants>
+    <ErrorReport>prompt</ErrorReport>
+    <WarningLevel>4</WarningLevel>
+  </PropertyGroup>
+  <ItemGroup>
+    <Reference Include="NLog, Version=3.2.0.0, Culture=neutral, PublicKeyToken=5120e14c03d0593c, processorArchitecture=MSIL">
+      <HintPath>..\..\packages\NLog.3.2.0.0\lib\net45\NLog.dll</HintPath>
+      <Private>True</Private>
+    </Reference>
+    <Reference Include="System" />
+    <Reference Include="System.configuration" />
+    <Reference Include="System.Core" />
+    <Reference Include="System.Messaging" />
+    <Reference Include="System.Reactive.Core, Version=2.2.5.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35, processorArchitecture=MSIL">
+      <HintPath>..\..\packages\Rx-Core.2.2.5\lib\net45\System.Reactive.Core.dll</HintPath>
+      <Private>True</Private>
+    </Reference>
+    <Reference Include="System.Reactive.Interfaces, Version=2.2.5.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35, processorArchitecture=MSIL">
+      <HintPath>..\..\packages\Rx-Interfaces.2.2.5\lib\net45\System.Reactive.Interfaces.dll</HintPath>
+      <Private>True</Private>
+    </Reference>
+    <Reference Include="System.Reactive.Linq, Version=2.2.5.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35, processorArchitecture=MSIL">
+      <HintPath>..\..\packages\Rx-Linq.2.2.5\lib\net45\System.Reactive.Linq.dll</HintPath>
+      <Private>True</Private>
+    </Reference>
+    <Reference Include="System.Reactive.PlatformServices, Version=2.2.5.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35, processorArchitecture=MSIL">
+      <HintPath>..\..\packages\Rx-PlatformServices.2.2.5\lib\net45\System.Reactive.PlatformServices.dll</HintPath>
+      <Private>True</Private>
+    </Reference>
+    <Reference Include="System.Xml.Linq" />
+    <Reference Include="System.Data.DataSetExtensions" />
+    <Reference Include="Microsoft.CSharp" />
+    <Reference Include="System.Data" />
+    <Reference Include="System.Net.Http" />
+    <Reference Include="System.Xml" />
+  </ItemGroup>
+  <ItemGroup>
+    <Compile Include="BusSubscriber.cs" />
+    <Compile Include="InterProcessBus.cs" />
+    <Compile Include="Properties\AssemblyInfo.cs" />
+  </ItemGroup>
+  <ItemGroup>
+    <ProjectReference Include="..\..\SachaBarber.CQRS.Demo.SharedCore\SachaBarber.CQRS.Demo.SharedCore.csproj">
+      <Project>{f97fe1c8-53bf-4c69-8694-6a8a6776d86a}</Project>
+      <Name>SachaBarber.CQRS.Demo.SharedCore</Name>
+    </ProjectReference>
+  </ItemGroup>
+  <ItemGroup>
+    <None Include="packages.config" />
+  </ItemGroup>
+  <Import Project="$(MSBuildToolsPath)\Microsoft.CSharp.targets" />
+  <!-- To modify your build process, add your task inside one of the targets below and uncomment it. 
+       Other similar extension points exist, see Microsoft.Common.targets.
+  <Target Name="BeforeBuild">
+  </Target>
+  <Target Name="AfterBuild">
+  </Target>
+  -->
+</Project>

SachaBarber.CQRS.Demo/BusAdapters/SachaBarber.CQRS.Demo.MSQMAdapter/packages.config

@@ -0,0 +1,9 @@
+<?xml version="1.0" encoding="utf-8"?>
+<packages>
+  <package id="NLog" version="3.2.0" targetFramework="net451" userInstalled="true" />
+  <package id="Rx-Core" version="2.2.5" targetFramework="net452" userInstalled="true" />
+  <package id="Rx-Interfaces" version="2.2.5" targetFramework="net452" userInstalled="true" />
+  <package id="Rx-Linq" version="2.2.5" targetFramework="net452" userInstalled="true" />
+  <package id="Rx-Main" version="2.2.5" targetFramework="net452" userInstalled="true" />
+  <package id="Rx-PlatformServices" version="2.2.5" targetFramework="net452" userInstalled="true" />
+</packages>

SachaBarber.CQRS.Demo/BusAdapters/SachaBarber.CQRS.Demo.RabbitMQAdapter/BusSubscriber.cs

@@ -0,0 +1,101 @@
+using System;
+using System.Reflection;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using RabbitMQ.Client;
+using RabbitMQ.Client.Events;
+using System.Reactive.Subjects;
+using System.Configuration;
+using System.Reactive.Linq;
+
+namespace SachaBarber.CQRS.Demo.RabbitMQAdapter
+{
+    public class BusSubscriber : SharedCore.Services.IInterProcessBusSubscriber, IDisposable
+    {
+        private readonly string busName;
+        private readonly string connectionString;
+        private CancellationTokenSource cancellationToken;
+        private Task workerTask;
+        private ISubject<string> eventsSubject = new Subject<string>();
+
+        public BusSubscriber()
+        {
+            this.busName = "InterProcessBus";
+            this.connectionString = ConfigurationManager.AppSettings["RabbitMqHost"];
+            StartMessageListener();
+        }
+
+        private void StartMessageListener()
+        {
+            cancellationToken = new CancellationTokenSource();
+            workerTask = Task.Factory.StartNew(() => ListenForMessage(), cancellationToken.Token);
+        }
+
+        public void Dispose()
+        {
+            CancelWorkerTask();
+        }
+
+        private void CancelWorkerTask()
+        {
+            if (workerTask == null) return;
+            cancellationToken.Cancel();
+            workerTask.Wait();
+            workerTask.Dispose();
+        }
+
+        private void ListenForMessage()
+        {
+            var factory = new ConnectionFactory() { HostName = connectionString };
+            using (var connection = factory.CreateConnection())
+            {
+                using (var channel = connection.CreateModel())
+                {
+                    channel.ExchangeDeclare(busName, "fanout");
+
+                    bool durable = true;
+                    bool exclusive = false;
+                    bool autoDelete = false;
+
+                    var queue = channel.QueueDeclare(
+                        Assembly.GetEntryAssembly().GetName().Name,
+                        durable, exclusive, autoDelete, null);
+                    channel.QueueBind(queue.QueueName, busName, string.Empty);
+                    var consumer = new QueueingBasicConsumer(channel);
+
+                    channel.BasicConsume(queue.QueueName, false, string.Empty, consumer);
+
+                    while (true)
+                    {
+                        if (cancellationToken.IsCancellationRequested)
+                            break;
+                        BasicDeliverEventArgs ea;
+                        consumer.Queue.Dequeue(10, out ea);
+
+                        if (ea == null)
+                            continue;
+
+                        var message = Encoding.ASCII.GetString(ea.Body);
+                        Task.Run(async () =>
+                        {
+                            await Task.Run(() =>
+                            {
+                                eventsSubject.OnNext(message);
+                            });
+                        });
+                        channel.BasicAck(ea.DeliveryTag, false);
+                    }
+
+                }
+
+            }
+        }
+
+
+        public IObservable<string> GetEventStream()
+        {
+            return eventsSubject.AsObservable();
+        }
+    }
+}