Skip to content
This repository

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse code

Merge branch 'rq-one-way'

  • Loading branch information...
commit 0629bbad0b186c5acaaa339892ade6da2c14926d 2 parents 841e975 + 5aa7066
mnichols authored
98 Rhino.ServiceBus.Tests/CanSendMsgsFromOneWayBusUsingRhinoQueues.cs
... ... @@ -0,0 +1,98 @@
  1 +using System;
  2 +using System.IO;
  3 +using System.Threading;
  4 +using System.Transactions;
  5 +using Castle.Windsor;
  6 +using Castle.Windsor.Configuration.Interpreters;
  7 +using Rhino.ServiceBus.Impl;
  8 +using Rhino.ServiceBus.Internal;
  9 +using Rhino.ServiceBus.RhinoQueues;
  10 +using Rhino.ServiceBus.Tests.RhinoQueues;
  11 +using Xunit;
  12 +
  13 +namespace Rhino.ServiceBus.Tests
  14 +{
  15 + public class CanSendMsgsFromOneWayBusUsingRhinoQueues : WithDebugging,IDisposable
  16 + {
  17 + private WindsorContainer container;
  18 +
  19 + public CanSendMsgsFromOneWayBusUsingRhinoQueues()
  20 + {
  21 + if (Directory.Exists("one_way.esent"))
  22 + Directory.Delete("one_way.esent", true);
  23 + if (Directory.Exists("test_queue.esent"))
  24 + Directory.Delete("test_queue.esent", true);
  25 + if (Directory.Exists("test_queue_subscriptions.esent"))
  26 + Directory.Delete("test_queue_subscriptions.esent", true);
  27 + container = new WindsorContainer(new XmlInterpreter("OneWayBusRhinoQueues.config"));
  28 + container.Kernel.AddFacility("rhino.esb", new RhinoServiceBusFacility());
  29 + container.AddComponent<StringConsumer>();
  30 + StringConsumer.Value = null;
  31 + StringConsumer.Event = new ManualResetEvent(false);
  32 + }
  33 +
  34 +
  35 +
  36 + [Fact]
  37 + public void SendMessageToRemoteBus()
  38 + {
  39 + using (var bus = container.Resolve<IStartableServiceBus>())
  40 + {
  41 + bus.Start();
  42 + var transport = new RhinoQueuesTransport(new Uri("null://nowhere:24689/middle"),
  43 + new EndpointRouter(), container.Resolve<IMessageSerializer>(),
  44 + 1, "one_way.esent", IsolationLevel.ReadCommitted, 5);
  45 + var oneWay = new RhinoQueuesOneWayBus(new[]
  46 + {
  47 + new MessageOwner
  48 + {
  49 + Endpoint = bus.Endpoint.Uri,
  50 + Name = "System",
  51 + },
  52 + }, transport);
  53 +
  54 + oneWay.Send("hello there, one way");
  55 +
  56 + StringConsumer.Event.WaitOne();
  57 +
  58 + Assert.Equal("hello there, one way", StringConsumer.Value);
  59 + }
  60 + }
  61 +
  62 + [Fact]
  63 + public void SendMessageToRemoteBusFromConfigDrivenOneWayBus()
  64 + {
  65 + using (var bus = container.Resolve<IStartableServiceBus>())
  66 + {
  67 + bus.Start();
  68 +
  69 + using (var c = new WindsorContainer(new XmlInterpreter("OneWayBusRhinoQueues.config")))
  70 + {
  71 + c.Kernel.AddFacility("one.way.rhino.esb", new OnewayRhinoServiceBusFacility());
  72 + c.Resolve<IOnewayBus>().Send("hello there, one way");
  73 + StringConsumer.Event.WaitOne();
  74 + Assert.Equal("hello there, one way", StringConsumer.Value);
  75 + }
  76 +
  77 +
  78 + }
  79 + }
  80 +
  81 + public class StringConsumer : ConsumerOf<string>
  82 + {
  83 + public static ManualResetEvent Event;
  84 + public static string Value;
  85 +
  86 + public void Consume(string pong)
  87 + {
  88 + Value = pong;
  89 + Event.Set();
  90 + }
  91 + }
  92 +
  93 + public void Dispose()
  94 + {
  95 + container.Dispose();
  96 + }
  97 + }
  98 +}
21 Rhino.ServiceBus.Tests/OneWayBusRhinoQueues.config
... ... @@ -0,0 +1,21 @@
  1 +<castle>
  2 + <facilities>
  3 + <facility id="rhino.esb" >
  4 + <bus name="test_queue"
  5 + threadCount="1"
  6 + numberOfRetries="5"
  7 + endpoint="rhino.queues://localhost/test_queue"
  8 + queueIsolationLevel="ReadCommitted" />
  9 + <messages>
  10 + <add name="Rhino.ServiceBus.Tests"
  11 + endpoint="rhino.queues://localhost/test_queue"/>
  12 + </messages>
  13 + </facility>
  14 + <facility id="one.way.rhino.esb" >
  15 + <messages>
  16 + <add name="System.String"
  17 + endpoint="rhino.queues://localhost/test_queue"/>
  18 + </messages>
  19 + </facility>
  20 + </facilities>
  21 +</castle>
4 Rhino.ServiceBus.Tests/Rhino.ServiceBus.Tests.csproj
@@ -112,6 +112,7 @@
112 112 <Compile Include="BusSubscriptionTests.cs" />
113 113 <Compile Include="CanRouteMessageToConsumerThroughContainer.cs" />
114 114 <Compile Include="CanSendMsgsFromOneWayBus.cs" />
  115 + <Compile Include="CanSendMsgsFromOneWayBusUsingRhinoQueues.cs" />
115 116 <Compile Include="DelayedMessages.cs" />
116 117 <Compile Include="DataStructures\LRUSetTest.cs" />
117 118 <Compile Include="DataStructures\HashtableTest.cs" />
@@ -178,6 +179,9 @@
178 179 <None Include="BusOnTransactionalQueue.config">
179 180 <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
180 181 </None>
  182 + <None Include="OneWayBusRhinoQueues.config">
  183 + <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
  184 + </None>
181 185 <None Include="LoadBalancer\BusWithAcceptingWorkLoadBalancer.config">
182 186 <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
183 187 </None>
2  Rhino.ServiceBus/IOnewayBus.cs
... ... @@ -1,3 +1,5 @@
  1 +using System;
  2 +
1 3 namespace Rhino.ServiceBus
2 4 {
3 5 public interface IOnewayBus
6 Rhino.ServiceBus/Impl/MessageOwnersConfigReader.cs
@@ -15,7 +15,7 @@ public MessageOwnersConfigReader(IConfiguration configuration, ICollection<Messa
15 15 this.configuration = configuration;
16 16 this.messageOwners = messageOwners;
17 17 }
18   -
  18 + public string EndpointScheme { get; private set; }
19 19 public void ReadMessageOwners()
20 20 {
21 21 IConfiguration messageConfig = configuration.Children["messages"];
@@ -36,6 +36,10 @@ public void ReadMessageOwners()
36 36 try
37 37 {
38 38 ownerEndpoint = new Uri(uriString);
  39 + if(EndpointScheme==null)
  40 + {
  41 + EndpointScheme = ownerEndpoint.Scheme;
  42 + }
39 43 }
40 44 catch (Exception e)
41 45 {
62 Rhino.ServiceBus/Impl/OnewayRhinoServiceBusFacility.cs
... ... @@ -1,10 +1,13 @@
1 1 using System;
2 2 using System.Collections.Generic;
  3 +using System.IO;
  4 +using System.Transactions;
3 5 using Castle.Core;
4 6 using Castle.MicroKernel.Facilities;
5 7 using Castle.MicroKernel.Registration;
6 8 using Rhino.ServiceBus.Internal;
7 9 using Rhino.ServiceBus.Msmq;
  10 +using Rhino.ServiceBus.RhinoQueues;
8 11 using Rhino.ServiceBus.Serializers;
9 12 using System.Linq;
10 13
@@ -22,27 +25,58 @@ public void UseMessageSerializer<TMessageSerializer>()
22 25
23 26 protected override void Init()
24 27 {
25   - new MessageOwnersConfigReader(FacilityConfig, messageOwners).ReadMessageOwners();
26   -
  28 + var messageOwnersReader = new MessageOwnersConfigReader(FacilityConfig, messageOwners);
  29 + messageOwnersReader.ReadMessageOwners();
  30 + if (IsRhinoQueues(messageOwnersReader.EndpointScheme))
  31 + {
  32 + var path = Path.GetFullPath(AppDomain.CurrentDomain.BaseDirectory);
  33 + Kernel.Register(
  34 + Component.For<ITransport>()
  35 + .LifeStyle.Is(LifestyleType.Singleton)
  36 + .ImplementedBy(typeof (RhinoQueuesTransport))
  37 + .DependsOn(new
  38 + {
  39 + threadCount = 1,
  40 + endpoint = new Uri("null://nowhere:24689/middle"),
  41 + queueIsolationLevel = IsolationLevel.ReadCommitted,
  42 + numberOfRetries = 5,
  43 + path = Path.Combine(path,"one_way.esent")
  44 + }),
  45 + Component.For<IOnewayBus>()
  46 + .LifeStyle.Is(LifestyleType.Singleton)
  47 + .ImplementedBy<RhinoQueuesOneWayBus>()
  48 + .DependsOn(new {messageOwners = messageOwners.ToArray()})
  49 + );
27 50
  51 + }
  52 + else
  53 + {
  54 + Kernel.Register(
  55 + Component.For<IMessageBuilder>()
  56 + .LifeStyle.Is(LifestyleType.Singleton)
  57 + .ImplementedBy<MessageBuilder>(),
  58 + Component.For<IOnewayBus>()
  59 + .LifeStyle.Is(LifestyleType.Singleton)
  60 + .ImplementedBy<OnewayBus>()
  61 + .DependsOn(new {messageOwners = messageOwners.ToArray()}));
  62 +
  63 + }
28 64 Kernel.Register(
29   - Component.For<IMessageBuilder>()
30   - .LifeStyle.Is(LifestyleType.Singleton)
31   - .ImplementedBy<MessageBuilder>(),
32   - Component.For<IOnewayBus>()
33   - .LifeStyle.Is(LifestyleType.Singleton)
34   - .ImplementedBy<OnewayBus>()
35   - .DependsOn(new{messageOwners = messageOwners.ToArray()}),
36 65 Component.For<IReflection>()
37   - .LifeStyle.Is(LifestyleType.Singleton)
38   - .ImplementedBy<DefaultReflection>(),
39   -
  66 + .LifeStyle.Is(LifestyleType.Singleton)
  67 + .ImplementedBy<DefaultReflection>(),
40 68 Component.For<IMessageSerializer>()
41 69 .LifeStyle.Is(LifestyleType.Singleton)
42 70 .ImplementedBy(serializerImpl),
43 71 Component.For<IEndpointRouter>()
44   - .ImplementedBy<EndpointRouter>()
45   - );
  72 + .ImplementedBy<EndpointRouter>()
  73 + );
  74 +
  75 + }
  76 +
  77 + private static bool IsRhinoQueues(string endpointScheme)
  78 + {
  79 + return endpointScheme.Equals("rhino.queues", StringComparison.InvariantCultureIgnoreCase);
46 80 }
47 81 }
48 82 }
1  Rhino.ServiceBus/Rhino.ServiceBus.csproj
@@ -122,6 +122,7 @@
122 122 <Compile Include="Messages\ReadyToWork.cs" />
123 123 <Compile Include="Msmq\AbstractMsmqListener.cs" />
124 124 <Compile Include="Msmq\QueueCreationModule.cs" />
  125 + <Compile Include="RhinoQueues\RhinoQueuesOneWayBus.cs" />
125 126 <Compile Include="Transport\SubQueue.cs" />
126 127 <Compile Include="Msmq\TransportActions\ShutDownAction.cs" />
127 128 <Compile Include="Msmq\TransportState.cs" />
23 Rhino.ServiceBus/RhinoQueues/RhinoQueuesOneWayBus.cs
... ... @@ -0,0 +1,23 @@
  1 +using Rhino.ServiceBus.Impl;
  2 +using Rhino.ServiceBus.Internal;
  3 +
  4 +namespace Rhino.ServiceBus.RhinoQueues
  5 +{
  6 + public class RhinoQueuesOneWayBus : IOnewayBus
  7 + {
  8 + private MessageOwnersSelector messageOwners;
  9 + private ITransport transport;
  10 +
  11 + public RhinoQueuesOneWayBus(MessageOwner[] messageOwners, ITransport transport)
  12 + {
  13 + this.messageOwners = new MessageOwnersSelector(messageOwners, new EndpointRouter());
  14 + this.transport = transport;
  15 + this.transport.Start();
  16 + }
  17 +
  18 + public void Send(params object[] msgs)
  19 + {
  20 + transport.Send(messageOwners.GetEndpointForMessageBatch(msgs), msgs);
  21 + }
  22 + }
  23 +}
3  Rhino.ServiceBus/RhinoQueues/RhinoQueuesTransport.cs
@@ -128,6 +128,9 @@ public IQueue Queue
128 128
129 129 public void Start()
130 130 {
  131 + if (haveStarted)
  132 + return;
  133 +
131 134 shouldContinue = true;
132 135
133 136 var port = endpoint.Port;

0 comments on commit 0629bba

Please sign in to comment.
Something went wrong with that request. Please try again.