Added support for using non transactional queues and added a send onl…

…y sample
bfcamara · Nov 3, 2010 · 4789bb2 · 4789bb2
1 parent 5ec8327
commit 4789bb2
Show file tree

Hide file tree

Showing 20 changed files with 233 additions and 92 deletions.
diff --git a/.gitignore b/.gitignore
@@ -15,4 +15,4 @@ _UpgradeReport.*
 *.user
 TestResult.xml
 results
-src/commonassemblyinfo.cs
+CommonAssemblyInfo.cs
diff --git a/Samples/SendOnlyEndpoint/SendOnlyEndpoint.Custom.sln b/Samples/SendOnlyEndpoint/SendOnlyEndpoint.Custom.sln
@@ -0,0 +1,20 @@
+
+Microsoft Visual Studio Solution File, Format Version 11.00
+# Visual Studio 2010
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "SendOnlyEndpoint.Custom", "SendOnlyEndpoint.Custom\SendOnlyEndpoint.Custom.csproj", "{52E814AA-CD2D-48F5-AD5C-38D8335DEBF4}"
+EndProject
+Global
+	GlobalSection(SolutionConfigurationPlatforms) = preSolution
+		Debug|x86 = Debug|x86
+		Release|x86 = Release|x86
+	EndGlobalSection
+	GlobalSection(ProjectConfigurationPlatforms) = postSolution
+		{52E814AA-CD2D-48F5-AD5C-38D8335DEBF4}.Debug|x86.ActiveCfg = Debug|x86
+		{52E814AA-CD2D-48F5-AD5C-38D8335DEBF4}.Debug|x86.Build.0 = Debug|x86
+		{52E814AA-CD2D-48F5-AD5C-38D8335DEBF4}.Release|x86.ActiveCfg = Release|x86
+		{52E814AA-CD2D-48F5-AD5C-38D8335DEBF4}.Release|x86.Build.0 = Release|x86
+	EndGlobalSection
+	GlobalSection(SolutionProperties) = preSolution
+		HideSolutionNode = FALSE
+	EndGlobalSection
+EndGlobal
diff --git a/Samples/SendOnlyEndpoint/SendOnlyEndpoint.Custom/Program.cs b/Samples/SendOnlyEndpoint/SendOnlyEndpoint.Custom/Program.cs
@@ -0,0 +1,32 @@
+using System;
+using NServiceBus;
+
+namespace SendOnlyEndpoint.Custom
+{
+    public class Program
+    {
+        static void Main()
+        {
+            var bus = Configure.With()
+                .DefaultBuilder()
+                .XmlSerializer()
+                .InMemoryFaultManagement()
+                .MsmqTransport()
+                .UnicastBus()
+                .CreateBus()
+                .Start();
+
+            bus.Send("SendOnlyDestination@someserver",new TestMessage());
+
+            Console.WriteLine("Message sent to remote endpoint, you can verify this by looking at the outgoing queues in you msmq MMC-snapin"); 
+            Console.WriteLine("Press any key to exit");
+
+            Console.ReadKey();
+        }
+    }
+
+
+    public class TestMessage : IMessage
+    {
+    }
+}
diff --git a/Samples/SendOnlyEndpoint/SendOnlyEndpoint.Custom/Properties/AssemblyInfo.cs b/Samples/SendOnlyEndpoint/SendOnlyEndpoint.Custom/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("SendOnlyEndpoint.Custom")]
+[assembly: AssemblyDescription("")]
+[assembly: AssemblyConfiguration("")]
+[assembly: AssemblyCompany("Microsoft")]
+[assembly: AssemblyProduct("SendOnlyEndpoint.Custom")]
+[assembly: AssemblyCopyright("Copyright © Microsoft 2010")]
+[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("e96ccad3-a27e-4941-bc3c-b68aec263446")]
+
+// 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")]
diff --git a/Samples/SendOnlyEndpoint/SendOnlyEndpoint.Custom/SendOnlyEndpoint.Custom.csproj b/Samples/SendOnlyEndpoint/SendOnlyEndpoint.Custom/SendOnlyEndpoint.Custom.csproj
@@ -0,0 +1,62 @@
+<?xml version="1.0" encoding="utf-8"?>
+<Project ToolsVersion="4.0" DefaultTargets="Build" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
+  <PropertyGroup>
+    <Configuration Condition=" '$(Configuration)' == '' ">Debug</Configuration>
+    <Platform Condition=" '$(Platform)' == '' ">x86</Platform>
+    <ProductVersion>8.0.30703</ProductVersion>
+    <SchemaVersion>2.0</SchemaVersion>
+    <ProjectGuid>{52E814AA-CD2D-48F5-AD5C-38D8335DEBF4}</ProjectGuid>
+    <OutputType>Exe</OutputType>
+    <AppDesignerFolder>Properties</AppDesignerFolder>
+    <RootNamespace>SendOnlyEndpoint.Custom</RootNamespace>
+    <AssemblyName>SendOnlyEndpoint.Custom</AssemblyName>
+    <TargetFrameworkVersion>v4.0</TargetFrameworkVersion>
+    <TargetFrameworkProfile>
+    </TargetFrameworkProfile>
+    <FileAlignment>512</FileAlignment>
+  </PropertyGroup>
+  <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Debug|x86' ">
+    <PlatformTarget>x86</PlatformTarget>
+    <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|x86' ">
+    <PlatformTarget>x86</PlatformTarget>
+    <DebugType>pdbonly</DebugType>
+    <Optimize>true</Optimize>
+    <OutputPath>bin\Release\</OutputPath>
+    <DefineConstants>TRACE</DefineConstants>
+    <ErrorReport>prompt</ErrorReport>
+    <WarningLevel>4</WarningLevel>
+  </PropertyGroup>
+  <ItemGroup>
+    <Reference Include="log4net">
+      <HintPath>..\..\..\binaries\log4net.dll</HintPath>
+    </Reference>
+    <Reference Include="NServiceBus">
+      <HintPath>..\..\..\binaries\NServiceBus.dll</HintPath>
+    </Reference>
+    <Reference Include="NServiceBus.Core">
+      <HintPath>..\..\..\binaries\NServiceBus.Core.dll</HintPath>
+    </Reference>
+    <Reference Include="System" />
+    <Reference Include="System.Core" />
+  </ItemGroup>
+  <ItemGroup>
+    <Compile Include="Program.cs" />
+    <Compile Include="Properties\AssemblyInfo.cs" />
+  </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>
diff --git a/nant.build b/nant.build
@@ -396,7 +396,7 @@
 
 	<target name="compile_samples" depends="prepare_binaries">
 		<foreach 	item="String"
-						in="AsyncPages,FullDuplex,PubSub,Manufacturing,GenericHost,Versioning,WcfIntegration,Starbucks"
+						in="AsyncPages,FullDuplex,PubSub,Manufacturing,GenericHost,Versioning,WcfIntegration,Starbucks,SendOnlyEndpoint"
 						delim=","
 						property="pathToBuild">
 			<echo message="${pathToBuild}" />

diff --git a/src/distributor/NServiceBus.Distributor/Service.cs b/src/distributor/NServiceBus.Distributor/Service.cs
@@ -65,7 +65,7 @@ void ConfigureDistributor(int numberOfThreads)
                                     {
                                         NumberOfWorkerThreads = numberOfThreads,
                                         IsTransactional = true,
-                                        MessageQueue = new MsmqMessageReceiver()
+                                        MessageReceiver = new MsmqMessageReceiver()
                                     };
 
             var distributor = new Unicast.Distributor.Distributor

diff --git a/src/gateway/NServiceBus.Gateway/EndpointConfig.cs b/src/gateway/NServiceBus.Gateway/EndpointConfig.cs
@@ -36,7 +36,7 @@ public void Init()
 
             transport = new TransactionalTransport
             {
-                MessageQueue = new MsmqMessageReceiver(),
+                MessageReceiver = new MsmqMessageReceiver(),
                 IsTransactional = true,
                 NumberOfWorkerThreads = numberOfWorkerThreads
             };

diff --git a/src/gateway/gateway.suo b/src/gateway/gateway.suo
diff --git a/src/impl/unicast/queuing/NServiceBus.Unicast.Queuing.Azure.Tests/AzureQueueFixture.cs b/src/impl/unicast/queuing/NServiceBus.Unicast.Queuing.Azure.Tests/AzureQueueFixture.cs
@@ -38,7 +38,7 @@ public void Setup()
                             PurgeOnStartup = PurgeOnStartup
                         };
 
-            queue.Init(QueueName);
+            queue.Init(QueueName,true);
         }
 
         protected void AddTestMessage()

diff --git a/...pl/unicast/queuing/NServiceBus.Unicast.Queuing.Azure.Tests/When_initializing_the_queue.cs b/...pl/unicast/queuing/NServiceBus.Unicast.Queuing.Azure.Tests/When_initializing_the_queue.cs
@@ -15,9 +15,9 @@ public void A_purge_can_be_requested()
 
             queue.PurgeOnStartup = true;
 
-            queue.Init(QueueName);
+            queue.Init(QueueName,false);
 
-            Assert.Null(queue.Receive(false));
+            Assert.Null(queue.Receive());
         }
     }
 } 
diff --git a/src/impl/unicast/queuing/NServiceBus.Unicast.Queuing.Azure.Tests/When_receiving_messages.cs b/src/impl/unicast/queuing/NServiceBus.Unicast.Queuing.Azure.Tests/When_receiving_messages.cs
@@ -25,36 +25,23 @@ public void Has_messages_should_indicate_if_messages_exists_int_the_queue()
 
         }
 
-        [Test]
-        public void The_queue_should_poll_for_messages()
-        {
-            //setup a message insertion in 0.5 seconds
-            Action onstart = () =>
-            {
-                Thread.Sleep(100);
-                AddTestMessage();
-            };
-
-            onstart.BeginInvoke(null, null);
-
-            Assert.NotNull(queue.Receive(false));
-        }
         [Test]
         public void Should_throw_if_non_nservicebus_messages_are_received()
         {
             nativeQueue.AddMessage(new CloudQueueMessage("whatever"));
 
-            Assert.Throws<SerializationException>(() => queue.Receive(false));
+            Assert.Throws<SerializationException>(() => queue.Receive());
         }
+
         [Test]
         public void Should_default_to_non_transactionable_if_no_ambient_transaction_exists()
         {
             AddTestMessage();
             queue.MessageInvisibleTime = 1;
 
-            Assert.NotNull(queue.Receive(true));
+            Assert.NotNull(queue.Receive());
             Thread.Sleep(1000);
-            Assert.Null(queue.Receive(true));
+            Assert.Null(queue.Receive());
         }
 
         [Test]
@@ -65,14 +52,14 @@ public void Messages_should_not_reapper_in_the_queue_if_transaction_is_committed
             queue.MessageInvisibleTime = 1;
             using (var scope = new TransactionScope())
             {
-                Assert.NotNull(queue.Receive(true));
+                Assert.NotNull(queue.Receive());
 
                 scope.Complete();
             }
 
             Thread.Sleep(1000);
 
-            Assert.Null(queue.Receive(false));
+            Assert.Null(queue.Receive());
         }
 
         [Test]
@@ -83,16 +70,13 @@ public void The_received_message_should_reappear_in_the_queue_if_transaction_is_
             queue.MessageInvisibleTime = 2;
             using (new TransactionScope())
             {
-                Assert.NotNull(queue.Receive(true));
+                Assert.NotNull(queue.Receive());
 
                 //rollback
             }
-
-            Assert.Null(queue.Receive(false));
-
             Thread.Sleep(1000);
-
-            Assert.NotNull(queue.Receive(false));
+            
+            Assert.NotNull(queue.Receive());
         }
 
         [Test]
@@ -102,19 +86,19 @@ public void Received_messages_should_be_removed_from_the_queue()
 
             queue.MessageInvisibleTime = 1;
 
-            queue.Receive(false);
+            queue.Receive();
 
             Thread.Sleep(1000);
 
-            Assert.Null(queue.Receive(false));
+            Assert.Null(queue.Receive());
         }
 
         [Test]
         public void Send_messages_without_body_should_be_ok()
         {
             AddTestMessage();
 
-            var message = queue.Receive(false);
+            var message = queue.Receive();
 
             Assert.Null(message.Body);
         }
@@ -142,7 +126,7 @@ public void All_properties_should_be_preserved()
                                    };
                 AddTestMessage(original);
 
-                var result = queue.Receive(false);
+                var result = queue.Receive();
 
                 var resultMessage = formatter.Deserialize(new MemoryStream(result.Body)) as TestMessage;
                 Assert.AreEqual(resultMessage.TestProperty,"Test");

diff --git a/src/impl/unicast/queuing/NServiceBus.Unicast.Queuing.Azure/AzureMessageQueue.cs b/src/impl/unicast/queuing/NServiceBus.Unicast.Queuing.Azure/AzureMessageQueue.cs
@@ -44,9 +44,10 @@ public AzureMessageQueue(CloudQueueClient client)
             MaximumWaitTimeWhenIdle = 60000;
         }
 
-        public void Init(string inputqueue)
+        public void Init(string address, bool transactional)
         {
-            queue = client.GetQueueReference(inputqueue);
+            useTransactions = transactional;
+            queue = client.GetQueueReference(address);
             queue.CreateIfNotExist();
 
             if (PurgeOnStartup)
@@ -93,23 +94,23 @@ private void DelayNextPeekWhenThereIsNoMessage(bool hasMessage)
             }
         }
 
-        public TransportMessage Receive(bool transactional)
+        public TransportMessage Receive()
         {
-            var rawMessage = GetMessage(transactional);
+            var rawMessage = GetMessage();
 
             if (rawMessage == null)
                 return null;
 
             return DeserializeMessage(rawMessage);
         }
 
-        private CloudQueueMessage GetMessage(bool transactional)
+        private CloudQueueMessage GetMessage()
         {
             var receivedMessage = queue.GetMessage(TimeSpan.FromMilliseconds(MessageInvisibleTime));
 
             if (receivedMessage != null)
             {
-                if (!transactional || Transaction.Current == null)
+                if (!useTransactions || Transaction.Current == null)
                     queue.DeleteMessage(receivedMessage);
                 else
                     Transaction.Current.EnlistVolatile(new ReceiveResourceManager(queue, receivedMessage),EnlistmentOptions.None);
@@ -147,5 +148,7 @@ public void CreateQueue(string queueName)
         {
             client.GetQueueReference(queueName).CreateIfNotExist();
         }
+
+        private bool useTransactions;
     }
 }
diff --git a/src/impl/unicast/queuing/NServiceBus.Unicast.Queuing.Ftp/FtpMessageQueue.cs b/src/impl/unicast/queuing/NServiceBus.Unicast.Queuing.Ftp/FtpMessageQueue.cs
@@ -29,12 +29,12 @@ public bool HasMessage()
             return (_locQueue.Count > 0);
         }
 
-        public void Init(string inputqueue)
+        public void Init(string inputqueue, bool transactional)
         {
             this.SetupReceiveService();
         }
 
-        public TransportMessage Receive(bool transactional)
+        public TransportMessage Receive()
         {
             lock (_locker)
             {