IServiceOriented.ServiceBus.Samples.Chat/ChatServiceTransformer.cs IServiceOriented.ServiceBus.Samples.Chat/IChatService2.cs IServiceOriented.ServiceBus/PublishRequest.cs IServiceOriented.ServiceBus/TransformationDispatcher.cs IServiceOriented.ServiceBus/iServiceOriented.snk @@ -15,12 +15,17 @@ </behaviors> <client> <endpoint name="ChatClient" contract="IServiceOriented.ServiceBus.Samples.Chat.IChatService" binding="netNamedPipeBinding" bindingConfiguration="namedPipe" address="net.pipe://localhost/chatServer" /> + <endpoint name="ChatClient2" contract="IServiceOriented.ServiceBus.Samples.Chat.IChatService2" binding="netNamedPipeBinding" bindingConfiguration="namedPipe" address="net.pipe://localhost/chatServer2" /> <endpoint contract="IServiceOriented.ServiceBus.IServiceBusManagementService" binding="netNamedPipeBinding" bindingConfiguration="namedPipe" address="net.pipe://localhost/serviceBusManagement" /> </client> <services> <service name="ChatServer" behaviorConfiguration="debug" > <endpoint name="NamedPipeListener" contract="IServiceOriented.ServiceBus.Samples.Chat.IChatService" binding="netNamedPipeBinding" bindingConfiguration="namedPipe" address="send" /> </service> + <service name="ChatServer2" behaviorConfiguration="debug" > + <endpoint name="NamedPipeListener" contract="IServiceOriented.ServiceBus.Samples.Chat.IChatService2" binding="netNamedPipeBinding" bindingConfiguration="namedPipe" address="send" /> + </service> + <service name="IServiceOriented.ServiceBus.Samples.Chat.ChatService" behaviorConfiguration="debug" > <endpoint name="NamedPipe" contract="IServiceOriented.ServiceBus.Samples.Chat.IChatService" binding="netNamedPipeBinding" bindingConfiguration="namedPipe" address="net.pipe://localhost/chatServer" /> </service> IServiceOriented.ServiceBus.Samples.Chat/App.config @@ -9,33 +9,61 @@ namespace IServiceOriented.ServiceBus.Samples.Chat { public class ChatClient { - public ChatClient(string name) + public ChatClient(string name, bool version2) { _from = name; - _handler = new IncomingHandler(); - _host = new ServiceHost(_handler, new Uri("net.pipe://localhost/chat/"+_from)); + + if (version2) + { + _host = new ServiceHost(new IncomingHandler2(), new Uri("net.pipe://localhost/chat/" + _from)); + } + else + { + _host = new ServiceHost(new IncomingHandler(), new Uri("net.pipe://localhost/chat/" + _from)); + } + + _version2 = version2; } + bool _version2; + public void Start() { _host.Open(); Service.Use<IServiceBusManagementService>(serviceBus => { - serviceBus.Subscribe(new SubscriptionEndpoint(Guid.NewGuid(), _from, "ChatClient", "net.pipe://localhost/chat/"+_from+"/send", typeof(IChatService), typeof(WcfDispatcher<IChatService>), new ChatFilter(_from))); + + if (_version2) + { + serviceBus.Subscribe(new SubscriptionEndpoint(Guid.NewGuid(), _from, "ChatClient2", "net.pipe://localhost/chat/" + _from + "/send", typeof(IChatService2), new WcfDispatcher<IChatService2>(), new ChatFilter2(_from))); + } + else + { + serviceBus.Subscribe(new SubscriptionEndpoint(Guid.NewGuid(), _from, "ChatClient", "net.pipe://localhost/chat/" + _from + "/send", typeof(IChatService), new WcfDispatcher<IChatService>(), new ChatFilter(_from))); + } }); } - string _from; - - IncomingHandler _handler; + string _from; + public void Send(string to, string message) { - Service.Use<IChatService>("ChatClient", chatService => + if (_version2) + { + Service.Use<IChatService2>("ChatClient2", chatService => + { + chatService.SendMessage(new SendMessageRequest2("Title", _from, to, message)); + }); + } + else { - chatService.SendMessage(new SendMessageRequest(_from, to, message)); - }); + Service.Use<IChatService>("ChatClient", chatService => + { + chatService.SendMessage(new SendMessageRequest(_from, to, message)); + }); + } } ServiceHost _host; @@ -55,6 +83,18 @@ namespace IServiceOriented.ServiceBus.Samples.Chat Console.WriteLine(request.From + ": " + request.Message); } #endregion + } + + [ServiceBehavior(InstanceContextMode = InstanceContextMode.Single, ConfigurationName = "ChatServer2")] + public class IncomingHandler2 : IChatService2 + { + #region IChatService Members + [OperationBehavior] + public void SendMessage(SendMessageRequest2 request) + { + Console.WriteLine(request.From + ": " + request.Title + "\r\n"+request.Message); + } + #endregion } } @@ -73,23 +113,40 @@ namespace IServiceOriented.ServiceBus.Samples.Chat } [DataMember] public string To; + + public override bool Include(string action, object message) + { + SendMessageRequest request = message as SendMessageRequest; + if (request != null) + { + return String.Compare(request.To, To, true) == 0; + } + return false; + } + } - protected override string CreateInitString() + + + [DataContract] + public class ChatFilter2 : MessageFilter + { + public ChatFilter2() { - return To; } - protected override void InitFromString(string data) + public ChatFilter2(string to) { - To = data; + To = to; } + [DataMember] + public string To; public override bool Include(string action, object message) - { - SendMessageRequest request = message as SendMessageRequest; - if (request != null) + { + SendMessageRequest2 request2 = message as SendMessageRequest2; + if (request2 != null) { - return String.Compare(request.To, To, true) == 0; + return String.Compare(request2.To, To, true) == 0; } return false; } IServiceOriented.ServiceBus.Samples.Chat/ChatClient.cs @@ -10,17 +10,16 @@ namespace IServiceOriented.ServiceBus.Samples.Chat public ChatServer() { _serviceBus = new ServiceBusRuntime(new MsmqMessageDeliveryQueue(".\\private$\\chat_deliver", true), new MsmqMessageDeliveryQueue(".\\private$\\chat_retry", true), new MsmqMessageDeliveryQueue(".\\private$\\chat_fail", true)); - _serviceBus.AddListener(new ListenerEndpoint(Guid.NewGuid(), "Chat Service", "ChatServer", "net.pipe://localhost/chatServer", typeof(IChatService), typeof(WcfListener<IChatService>))); + _serviceBus.AddListener(new ListenerEndpoint(Guid.NewGuid(), "Chat Service", "ChatServer", "net.pipe://localhost/chatServer", typeof(IChatService), new WcfListener<IChatService>())); + _serviceBus.AddListener(new ListenerEndpoint(Guid.NewGuid(), "Chat Service2", "ChatServer2", "net.pipe://localhost/chatServer2", typeof(IChatService2), new WcfListener<IChatService2>())); + _serviceBus.Subscribe(new SubscriptionEndpoint(Guid.NewGuid(), "Chat Service Transformer", null, null, typeof(IChatService), new ChatServiceTransformer(), new TypedMessageFilter(typeof(SendMessageRequest)))); + _serviceBus.Subscribe(new SubscriptionEndpoint(Guid.NewGuid(), "No subscribers", "ChatClient", "", typeof(IChatService), new MethodDispatcher<IChatService>(new UnhandledReplyHandler(_serviceBus)), new UnhandledMessageFilter(typeof(SendMessageRequest)))); _serviceBus.RegisterService(new WcfManagementService()); _serviceBus.UnhandledException+= (o, ex) => { Console.WriteLine("Unhandled Exception: "+ex.ExceptionObject); }; - - IChatService noListenerReply = new UnhandledReplyHandler(_serviceBus); - MethodDispatcherConfiguration.For(_serviceBus).RegisterTarget(typeof(IChatService), noListenerReply); - - _serviceBus.Subscribe(new SubscriptionEndpoint(Guid.NewGuid(), "No subscribers", "ChatClient", "", typeof(IChatService), typeof(MethodDispatcher<IChatService>), new UnhandledMessageFilter(typeof(SendMessageRequest)))); + } class UnhandledReplyHandler : IChatService @@ -33,7 +32,7 @@ namespace IServiceOriented.ServiceBus.Samples.Chat ServiceBusRuntime _serviceBus; public void SendMessage(SendMessageRequest request) { - _serviceBus.Publish(typeof(IChatService), "SendMessage", new SendMessageRequest("System", request.From, request.To + " is an invalid user")); + _serviceBus.Publish(new PublishRequest(typeof(IChatService), "SendMessage", new SendMessageRequest("System", request.From, request.To + " is an invalid user"))); } } IServiceOriented.ServiceBus.Samples.Chat/ChatServer.cs @@ -54,7 +54,9 @@ <ItemGroup> <Compile Include="ChatClient.cs" /> <Compile Include="ChatServer.cs" /> + <Compile Include="ChatServiceTransformer.cs" /> <Compile Include="IChatService.cs" /> + <Compile Include="IChatService2.cs" /> <Compile Include="Program.cs" /> <Compile Include="Properties\AssemblyInfo.cs" /> </ItemGroup> IServiceOriented.ServiceBus.Samples.Chat/IServiceOriented.ServiceBus.Samples.Chat.csproj @@ -10,7 +10,9 @@ namespace IServiceOriented.ServiceBus.Samples.Chat static void Main(string[] args) { MessageDelivery.RegisterKnownType(typeof(ChatFilter)); + MessageDelivery.RegisterKnownType(typeof(ChatFilter2)); MessageDelivery.RegisterKnownType(typeof(SendMessageRequest)); + MessageDelivery.RegisterKnownType(typeof(SendMessageRequest2)); if(args.Length == 0) { @@ -28,13 +30,13 @@ namespace IServiceOriented.ServiceBus.Samples.Chat server.Stop(); } - else if(args[0] == "client") + else if (args[0] == "client" || args[0] == "client2") { Console.WriteLine("Starting client..."); Console.Write("Enter name: "); - ChatClient client = new ChatClient(Console.ReadLine()); + ChatClient client = new ChatClient(Console.ReadLine(), args[0] == "client2"); client.Start(); while (true) IServiceOriented.ServiceBus.Samples.Chat/Program.cs @@ -19,15 +19,6 @@ namespace IServiceOriented.ServiceBus.UnitTests { return _messageFilter; } - - protected override void InitFromString(string data) - { - _messageFilter = Convert.ToBoolean(data); - } - - protected override string CreateInitString() - { - return _messageFilter.ToString(); - } + } } IServiceOriented.ServiceBus.UnitTests/BooleanMessageFilter.cs @@ -130,7 +130,7 @@ namespace IServiceOriented.ServiceBus.UnitTests { MsmqMessageDeliveryQueue queue = new MsmqMessageDeliveryQueue(_testQueuePath); - SubscriptionEndpoint endpoint = new SubscriptionEndpoint(Guid.NewGuid(), "SubscriptionName", "http://localhost/test", "SubscriptionConfigName", typeof(IContract), typeof(WcfDispatcher<IContract>), new PassThroughMessageFilter()); + SubscriptionEndpoint endpoint = new SubscriptionEndpoint(Guid.NewGuid(), "SubscriptionName", "http://localhost/test", "SubscriptionConfigName", typeof(IContract), new WcfDispatcher<IContract>(), new PassThroughMessageFilter()); MessageDelivery enqueued = new MessageDelivery(endpoint.Id, "randomAction","randomMessageData", 3); Assert.IsNull(queue.Peek(TimeSpan.FromSeconds(1)), "Queue must be empty to start transaction test"); @@ -194,7 +194,7 @@ namespace IServiceOriented.ServiceBus.UnitTests { MsmqMessageDeliveryQueue queue = new MsmqMessageDeliveryQueue(_testQueuePath); - SubscriptionEndpoint endpoint = new SubscriptionEndpoint(Guid.NewGuid(), "SubscriptionName", "http://localhost/test", "SubscriptionConfigName", typeof(IContract), typeof(WcfDispatcher<IContract>), new PassThroughMessageFilter()); + SubscriptionEndpoint endpoint = new SubscriptionEndpoint(Guid.NewGuid(), "SubscriptionName", "http://localhost/test", "SubscriptionConfigName", typeof(IContract), new WcfDispatcher<IContract>(), new PassThroughMessageFilter()); MessageDelivery enqueued = new MessageDelivery(endpoint.Id, messageAction, messageData, 3); IServiceOriented.ServiceBus.UnitTests/MsmqMessageDeliveryQueueTest.cs @@ -14,18 +14,6 @@ namespace IServiceOriented.ServiceBus.UnitTests return true; } - protected override void InitFromString(string data) - { - if (data != "passthrough") - { - throw new Exception("Data was invalid"); - } - } - - protected override string CreateInitString() - { - return "passthrough"; - } } } IServiceOriented.ServiceBus.UnitTests/PassThroughMessageFilter.cs @@ -64,7 +64,7 @@ namespace IServiceOriented.ServiceBus.UnitTests void putMessageAndGet(object message) { - SubscriptionEndpoint subscription = new SubscriptionEndpoint(Guid.NewGuid(), "test", "configurationName", "address", typeof(IServiceBusManagementService), typeof(WcfDispatcher<IServiceBusManagementService>), null); + SubscriptionEndpoint subscription = new SubscriptionEndpoint(Guid.NewGuid(), "test", "configurationName", "address", typeof(IServiceBusManagementService), new WcfDispatcher<IServiceBusManagementService>(), null); MessageDelivery failure = new MessageDelivery(subscription.Id, "action", message, 3); NonTransactionalMemoryQueue failureQueue = new NonTransactionalMemoryQueue(); IServiceOriented.ServiceBus.UnitTests/ServiceBusManagementTest.cs @@ -111,14 +111,13 @@ namespace IServiceOriented.ServiceBus.UnitTests serviceBusRuntime.MaxRetries = 1; - serviceBusRuntime.AddListener(new ListenerEndpoint(Guid.NewGuid(), "test", "NamedPipeListener", "net.pipe://localhost/servicebus/test", typeof(IContract), typeof(WcfListener<IContract>))); + serviceBusRuntime.AddListener(new ListenerEndpoint(Guid.NewGuid(), "test", "NamedPipeListener", "net.pipe://localhost/servicebus/test", typeof(IContract), new WcfListener<IContract>())); string message = "Publish this message"; - serviceBusRuntime.Subscribe(new SubscriptionEndpoint(Guid.NewGuid(), "subscription", "", "", typeof(IContract), typeof(MethodDispatcher<IContract>), new BooleanMessageFilter(false))); - ContractImplementation ci = new ContractImplementation(); ci.SetFailCount(1); - MethodDispatcherConfiguration.For(serviceBusRuntime).RegisterTarget(typeof(IContract), ci); + serviceBusRuntime.Subscribe(new SubscriptionEndpoint(Guid.NewGuid(), "subscription", "", "", typeof(IContract), new MethodDispatcher<IContract>(ci), new BooleanMessageFilter(false))); + AutoResetEvent wait = new AutoResetEvent(false); serviceBusRuntime.MessageDelivered += (o, mdea) => { wait.Set(); }; @@ -126,7 +125,7 @@ namespace IServiceOriented.ServiceBus.UnitTests serviceBusRuntime.Start(); - serviceBusRuntime.Publish(typeof(IContract), "PublishThis", message); + serviceBusRuntime.Publish(new PublishRequest(typeof(IContract), "PublishThis", message)); try { @@ -161,14 +160,12 @@ namespace IServiceOriented.ServiceBus.UnitTests serviceBusRuntime.RetryDelay = 1000; serviceBusRuntime.MaxRetries = 1; - serviceBusRuntime.AddListener(new ListenerEndpoint(Guid.NewGuid(), "test", "NamedPipeListener", "net.pipe://localhost/servicebus/test", typeof(IContract), typeof(WcfListener<IContract>))); + serviceBusRuntime.AddListener(new ListenerEndpoint(Guid.NewGuid(), "test", "NamedPipeListener", "net.pipe://localhost/servicebus/test", typeof(IContract), new WcfListener<IContract>())); string message = "Publish this message"; - serviceBusRuntime.Subscribe(new SubscriptionEndpoint(Guid.NewGuid(), "subscription", "", "", typeof(IContract), typeof(MethodDispatcher<IContract>), new BooleanMessageFilter(true))); - ContractImplementation ci = new ContractImplementation(); ci.SetFailCount(0); - MethodDispatcherConfiguration.For(serviceBusRuntime).RegisterTarget(typeof(IContract), ci); + serviceBusRuntime.Subscribe(new SubscriptionEndpoint(Guid.NewGuid(), "subscription", "", "", typeof(IContract), new MethodDispatcher<IContract>(ci), new BooleanMessageFilter(true))); AutoResetEvent wait = new AutoResetEvent(false); serviceBusRuntime.MessageDelivered += (o, mdea) => { wait.Set(); }; @@ -176,7 +173,7 @@ namespace IServiceOriented.ServiceBus.UnitTests serviceBusRuntime.Start(); - serviceBusRuntime.Publish(typeof(IContract), "PublishThis", message); + serviceBusRuntime.Publish(new PublishRequest(typeof(IContract), "PublishThis", message)); // Wait for delivery wait.WaitOne(TimeSpan.FromMinutes(1), false); // give it a minute @@ -202,14 +199,12 @@ namespace IServiceOriented.ServiceBus.UnitTests ServiceBusRuntime serviceBusRuntime = new ServiceBusRuntime(testQueue, retryQueue, failQueue); - serviceBusRuntime.AddListener(new ListenerEndpoint(Guid.NewGuid(), "test", "NamedPipeListener", "net.pipe://localhost/servicebus/test", typeof(IContract), typeof(WcfListener<IContract>))); + serviceBusRuntime.AddListener(new ListenerEndpoint(Guid.NewGuid(), "test", "NamedPipeListener", "net.pipe://localhost/servicebus/test", typeof(IContract), new WcfListener<IContract>())); string message = "Publish this message"; - serviceBusRuntime.Subscribe(new SubscriptionEndpoint(Guid.NewGuid(), "subscription", "", "", typeof(IContract), typeof(MethodDispatcher<IContract>), new PassThroughMessageFilter())); ContractImplementation ci = new ContractImplementation(); ci.SetFailCount(0); - - MethodDispatcherConfiguration.For(serviceBusRuntime).RegisterTarget(typeof(IContract), ci); + serviceBusRuntime.Subscribe(new SubscriptionEndpoint(Guid.NewGuid(), "subscription", "", "", typeof(IContract), new MethodDispatcher<IContract>(ci), new PassThroughMessageFilter())); AutoResetEvent wait = new AutoResetEvent(false); serviceBusRuntime.MessageDelivered += (o, mdea) => { wait.Set(); }; @@ -217,7 +212,7 @@ namespace IServiceOriented.ServiceBus.UnitTests serviceBusRuntime.Start(); - serviceBusRuntime.Publish(typeof(IContract), "PublishThis", message); + serviceBusRuntime.Publish(new PublishRequest(typeof(IContract), "PublishThis", message)); // Wait for delivery wait.WaitOne(TimeSpan.FromMinutes(1), false); // give it a minute @@ -244,20 +239,19 @@ namespace IServiceOriented.ServiceBus.UnitTests ServiceBusRuntime serviceBusRuntime = new ServiceBusRuntime(testQueue, retryQueue, failQueue); serviceBusRuntime.RegisterService(new PerformanceMonitorRuntimeService()); - serviceBusRuntime.AddListener(new ListenerEndpoint(Guid.NewGuid(), "test", "NamedPipeListener", "net.pipe://localhost/servicebus/test", typeof(IContract), typeof(WcfListener<IContract>))); - serviceBusRuntime.Subscribe(new SubscriptionEndpoint(Guid.NewGuid(), "subscription", "", "", typeof(IContract), typeof(MethodDispatcher<IContract>), new PassThroughMessageFilter())); - ContractImplementation ci = new ContractImplementation(); ci.SetFailCount(0); + serviceBusRuntime.AddListener(new ListenerEndpoint(Guid.NewGuid(), "test", "NamedPipeListener", "net.pipe://localhost/servicebus/test", typeof(IContract), new WcfListener<IContract>())); + serviceBusRuntime.Subscribe(new SubscriptionEndpoint(Guid.NewGuid(), "subscription", "", "", typeof(IContract), new MethodDispatcher<IContract>(ci), new PassThroughMessageFilter())); + + int messageCount = 10000; DateTime start = DateTime.Now; CountdownLatch countDown = new CountdownLatch(messageCount); - - MethodDispatcherConfiguration.For(serviceBusRuntime).RegisterTarget(typeof(IContract), ci); - + AutoResetEvent wait = new AutoResetEvent(false); serviceBusRuntime.MessageDelivered += (o, mdea) => { countDown.Tick(); }; serviceBusRuntime.MessageDeliveryFailed += (o, mdfea) => { countDown.Tick(); }; @@ -265,7 +259,7 @@ namespace IServiceOriented.ServiceBus.UnitTests for (int i = 0; i < messageCount; i++) { string message = i.ToString(); - serviceBusRuntime.Publish(typeof(IContract), "PublishThis", message); + serviceBusRuntime.Publish(new PublishRequest(typeof(IContract), "PublishThis", message)); } serviceBusRuntime.Start(); @@ -314,21 +308,21 @@ namespace IServiceOriented.ServiceBus.UnitTests serviceBusRuntime.RegisterService(new PerformanceMonitorRuntimeService()); - serviceBusRuntime.AddListener(new ListenerEndpoint(Guid.NewGuid(), "test", "NamedPipeListener", "net.pipe://localhost/servicebus/test", typeof(IContract), typeof(WcfListener<IContract>))); - serviceBusRuntime.Subscribe(new SubscriptionEndpoint(Guid.NewGuid(), "subscription", "", "", typeof(IContract), typeof(MethodDispatcher<IContract>), new PassThroughMessageFilter())); - ContractImplementation ci = new ContractImplementation(); ci.SetFailCount(0); ci.SetFailInterval(10); + serviceBusRuntime.AddListener(new ListenerEndpoint(Guid.NewGuid(), "test", "NamedPipeListener", "net.pipe://localhost/servicebus/test", typeof(IContract), new WcfListener<IContract>())); + serviceBusRuntime.Subscribe(new SubscriptionEndpoint(Guid.NewGuid(), "subscription", "", "", typeof(IContract), new MethodDispatcher<IContract>(ci), new PassThroughMessageFilter())); + + int messageCount = 1000; DateTime start = DateTime.Now; CountdownLatch countDown = new CountdownLatch(messageCount); - MethodDispatcherConfiguration.For(serviceBusRuntime).RegisterTarget(typeof(IContract), ci); - + AutoResetEvent wait = new AutoResetEvent(false); serviceBusRuntime.MessageDelivered += (o, mdea) => { countDown.Tick(); }; @@ -338,7 +332,7 @@ namespace IServiceOriented.ServiceBus.UnitTests for (int i = 0; i < messageCount; i++) { string message = i.ToString(); - serviceBusRuntime.Publish(typeof(IContract), "PublishThis", message); + serviceBusRuntime.Publish(new PublishRequest(typeof(IContract), "PublishThis", message)); } bool[] results = new bool[messageCount]; @@ -383,14 +377,12 @@ namespace IServiceOriented.ServiceBus.UnitTests serviceBusRuntime.RetryDelay = 1000; serviceBusRuntime.MaxRetries = 1; - serviceBusRuntime.AddListener(new ListenerEndpoint(Guid.NewGuid(), "test", "NamedPipeListener", "net.pipe://localhost/servicebus/test", typeof(IContract), typeof(WcfListener<IContract>))); + serviceBusRuntime.AddListener(new ListenerEndpoint(Guid.NewGuid(), "test", "NamedPipeListener", "net.pipe://localhost/servicebus/test", typeof(IContract), new WcfListener<IContract>())); string message = "Publish this message"; - serviceBusRuntime.Subscribe(new SubscriptionEndpoint(Guid.NewGuid(), "subscription", "", "", typeof(IContract), typeof(MethodDispatcher<IContract>), new PassThroughMessageFilter())); - ContractImplementation ci = new ContractImplementation(); ci.SetFailCount(1); + serviceBusRuntime.Subscribe(new SubscriptionEndpoint(Guid.NewGuid(), "subscription", "", "", typeof(IContract), new MethodDispatcher<IContract>(ci), new PassThroughMessageFilter())); - MethodDispatcherConfiguration.For(serviceBusRuntime).RegisterTarget(typeof(IContract), ci); CountdownLatch latch = new CountdownLatch(2); @@ -406,7 +398,7 @@ namespace IServiceOriented.ServiceBus.UnitTests serviceBusRuntime.Start(); - serviceBusRuntime.Publish(typeof(IContract), "PublishThis", message); + serviceBusRuntime.Publish(new PublishRequest(typeof(IContract), "PublishThis", message)); // Wait for delivery latch.Handle.WaitOne(TimeSpan.FromMinutes(1), false); // give it a minute @@ -440,14 +432,11 @@ namespace IServiceOriented.ServiceBus.UnitTests serviceBusRuntime.RetryDelay = 1000; serviceBusRuntime.MaxRetries = 1; - serviceBusRuntime.AddListener(new ListenerEndpoint(Guid.NewGuid(), "test", "NamedPipeListener", "net.pipe://localhost/servicebus/test", typeof(IContract), typeof(WcfListener<IContract>))); + serviceBusRuntime.AddListener(new ListenerEndpoint(Guid.NewGuid(), "test", "NamedPipeListener", "net.pipe://localhost/servicebus/test", typeof(IContract), new WcfListener<IContract>())); string message = "Publish this message"; - serviceBusRuntime.Subscribe(new SubscriptionEndpoint(Guid.NewGuid(), "subscription", "", "", typeof(IContract), typeof(MethodDispatcher<IContract>), new PassThroughMessageFilter())); - ContractImplementation ci = new ContractImplementation(); ci.SetFailCount(3); - - MethodDispatcherConfiguration.For(serviceBusRuntime).RegisterTarget(typeof(IContract), ci); + serviceBusRuntime.Subscribe(new SubscriptionEndpoint(Guid.NewGuid(), "subscription", "", "", typeof(IContract), new MethodDispatcher<IContract>(ci), new PassThroughMessageFilter())); CountdownLatch latch = new CountdownLatch(3); @@ -462,7 +451,7 @@ namespace IServiceOriented.ServiceBus.UnitTests serviceBusRuntime.Start(); - serviceBusRuntime.Publish(typeof(IContract), "PublishThis", message); + serviceBusRuntime.Publish(new PublishRequest(typeof(IContract), "PublishThis", message)); // Wait for delivery latch.Handle.WaitOne(TimeSpan.FromMinutes(1), false); // give it a minute IServiceOriented.ServiceBus.UnitTests/ServiceBusRuntimeTest.cs @@ -59,12 +59,12 @@ namespace IServiceOriented.ServiceBus.UnitTests [TestMethod] public void TestCrud() { - SqlSubscriptionDB db = new SqlSubscriptionDB(_connectionString); + SqlSubscriptionDB db = new SqlSubscriptionDB(_connectionString, new Type[] { typeof(WcfDispatcher<IContract>) }, new Type[] { typeof(WcfListener<IContract>) }, new Type[] { typeof(PassThroughMessageFilter) }); Assert.AreEqual(0, db.LoadListenerEndpoints().Count(), "Listener endpoints count should be zero"); Assert.AreEqual(0, db.LoadSubscriptionEndpoints().Count(), "Subscription endpoints count should be zero"); - ListenerEndpoint listener = new ListenerEndpoint(Guid.NewGuid(), "listener", "ListenerConfig", "http://localhost/test", typeof(IContract), typeof(WcfListener<IContract>)); + ListenerEndpoint listener = new ListenerEndpoint(Guid.NewGuid(), "listener", "ListenerConfig", "http://localhost/test", typeof(IContract), new WcfListener<IContract>()); db.CreateListener(listener); IEnumerable<ListenerEndpoint> listeners = db.LoadListenerEndpoints(); @@ -78,7 +78,7 @@ namespace IServiceOriented.ServiceBus.UnitTests Assert.AreEqual(listener.ConfigurationName, savedListener.ConfigurationName, "Configuration name does not match"); Assert.AreEqual(listener.Address, savedListener.Address, "Address does not match"); - SubscriptionEndpoint subscription = new SubscriptionEndpoint(Guid.NewGuid(), "subscription", "SubscriptionConfig", "http://localhost/test/subscription", typeof(IContract), typeof(WcfDispatcher<IContract>), new PassThroughMessageFilter()); + SubscriptionEndpoint subscription = new SubscriptionEndpoint(Guid.NewGuid(), "subscription", "SubscriptionConfig", "http://localhost/test/subscription", typeof(IContract), new WcfDispatcher<IContract>(), new PassThroughMessageFilter()); db.CreateSubscription(subscription); IEnumerable<SubscriptionEndpoint> subscriptions = db.LoadSubscriptionEndpoints(); @@ -90,7 +90,7 @@ namespace IServiceOriented.ServiceBus.UnitTests Assert.AreEqual(subscription.Address, savedSubscription.Address, "Address does not match"); Assert.AreEqual(subscription.ConfigurationName, savedSubscription.ConfigurationName, "ConfigurationName does not match"); Assert.AreEqual(subscription.ContractType, savedSubscription.ContractType, "ContractType does not match"); - Assert.AreEqual(subscription.DispatcherType, savedSubscription.DispatcherType, "DispatcherType does not match"); + // TODO: Compare dispatchers Assert.AreEqual(subscription.Id, savedSubscription.Id, "Id does not match"); Assert.AreEqual(subscription.Filter.GetType(), savedSubscription.Filter.GetType(), "Id does not match"); IServiceOriented.ServiceBus.UnitTests/SqlSubscriptionDbUnitTest.cs @@ -1,12 +1,12 @@ -create table listener (id uniqueidentifier primary key, name nvarchar(max), address nvarchar(max) not null, contract_type nvarchar(max) not null, configuration_name nvarchar(max) not null, listener_type nvarchar(max) not null) +create table listener (id uniqueidentifier primary key, name nvarchar(max), address nvarchar(max) not null, contract_type nvarchar(max) not null, configuration_name nvarchar(max) not null, listener_data varbinary(max) not null) go -create table subscription (id uniqueidentifier primary key, filter_type nvarchar(max), filter_data nvarchar(max), name nvarchar(max), address nvarchar(max) not null, contract_type nvarchar(max) not null, configuration_name nvarchar(max) not null, dispatcher_type nvarchar(max)) +create table subscription (id uniqueidentifier primary key, filter_data varbinary(max), name nvarchar(max), address nvarchar(max) not null, contract_type nvarchar(max) not null, configuration_name nvarchar(max) not null, dispatcher_data varbinary(max)) go -create proc sp_listener_create (@id as uniqueidentifier, @name as nvarchar(max), @address as nvarchar(max), @contract_type as nvarchar(max), @configuration_name as nvarchar(max), @listener_type as nvarchar(max)) +create proc sp_listener_create (@id as uniqueidentifier, @name as nvarchar(max), @address as nvarchar(max), @contract_type as nvarchar(max), @configuration_name as nvarchar(max), @listener_data as varbinary(max)) as -insert into listener (id, name, address, contract_type, configuration_name, listener_type) values (@id, @name, @address, @contract_type, @configuration_name, @listener_type) +insert into listener (id, name, address, contract_type, configuration_name, listener_data) values (@id, @name, @address, @contract_type, @configuration_name, @listener_data) go @@ -16,9 +16,9 @@ delete from listener where id = @id go -create proc sp_subscription_create (@id as uniqueidentifier, @name as nvarchar(max), @address as nvarchar(max), @contract_type as nvarchar(max), @configuration_name as nvarchar(max), @filter_type as nvarchar(max), @filter_data as nvarchar(max), @dispatcher_type as nvarchar(max)) +create proc sp_subscription_create (@id as uniqueidentifier, @name as nvarchar(max), @address as nvarchar(max), @contract_type as nvarchar(max), @configuration_name as nvarchar(max), @filter_data as varbinary(max), @dispatcher_data as varbinary(max)) as -insert into subscription (id, name, address, contract_type, configuration_name, filter_type, filter_data, dispatcher_type) values (@id, @name, @address, @contract_type, @configuration_name, @filter_type, @filter_data, @dispatcher_type) +insert into subscription (id, name, address, contract_type, configuration_name, filter_data, dispatcher_data) values (@id, @name, @address, @contract_type, @configuration_name, @filter_data, @dispatcher_data) go IServiceOriented.ServiceBus/CreateSqlSubscriptionPersistenceServiceDb.sql @@ -3,56 +3,54 @@ using System.Collections.Generic; using System.Linq; using System.Text; +using System.Runtime.Serialization; + + namespace IServiceOriented.ServiceBus -{ +{ + [Serializable] + [DataContract] public abstract class Dispatcher : IDisposable { + [NonSerialized] + ServiceBusRuntime _runtime; public ServiceBusRuntime Runtime { - get; - private set; + get + { + return _runtime; + } + internal set + { + _runtime = value; + } } + [NonSerialized] + SubscriptionEndpoint _endpoint; public SubscriptionEndpoint Endpoint { - get; - private set; - } - - internal void StartInternal(ServiceBusRuntime runtime, SubscriptionEndpoint endpoint) - { - try + get { - Runtime = runtime; - Endpoint = endpoint; - - OnStart(); - - Started = true; + return _endpoint; } - finally + internal set { - if (!Started) - { - Runtime = null; - Endpoint = null; - } + _endpoint = value; } } - internal void StopInternal() + internal void StartInternal() { - try - { - OnStop(); - } - finally - { - Runtime = null; - Endpoint = null; + OnStart(); - Started = false; - } + Started = true; + } + + internal void StopInternal() + { + OnStop(); + Started = false; } protected virtual void OnStart() @@ -69,7 +67,7 @@ namespace IServiceOriented.ServiceBus private set; } - [ThreadStatic] + [ThreadStatic, NonSerialized] static DispatchContext _dispatchContext; /// <summary> /// Contains information about the message that is currently being dispatched. IServiceOriented.ServiceBus/Dispatcher.cs @@ -114,6 +114,20 @@ namespace IServiceOriented.ServiceBus _name = value; } } + + bool _transient; + [DataMember] + public bool Transient + { + get + { + return _transient; + } + set + { + _transient = value; + } + } } IServiceOriented.ServiceBus/Endpoint.cs @@ -68,6 +68,11 @@ <Compile Include="InvalidContractException.cs" /> <Compile Include="ListenerEndpoint.cs" /> <Compile Include="IMessageDeliveryQueue.cs" /> + <Compile Include="PublishRequest.cs" /> + <Compile Include="SqlSubscriptionPersistenceService.cs"> + <SubType>Code</SubType> + </Compile> + <Compile Include="TransformationDispatcher.cs" /> <Compile Include="ReaderWriterLockedObject.cs" /> <Compile Include="UnhandledMessageFilter.cs" /> <Compile Include="PerformanceMonitorRuntimeService.cs" /> @@ -81,7 +86,6 @@ <Compile Include="Properties\AssemblyInfo.cs" /> <Compile Include="Service.cs" /> <Compile Include="ServiceBusRuntime.cs" /> - <Compile Include="SqlSubscriptionPersistenceService.cs" /> <Compile Include="SubscriptionEndpoint.cs" /> <Compile Include="SubscriptionNotFoundException.cs" /> <Compile Include="SubscriptionPersistenceService.cs" /> @@ -92,7 +96,6 @@ <Compile Include="WcfManagementService.cs" /> </ItemGroup> <ItemGroup> - <None Include="ClassDiagram.cd" /> <None Include="iServiceOriented.snk" /> <EmbeddedResource Include="CreateSqlSubscriptionPersistenceServiceDb.sql" /> </ItemGroup> IServiceOriented.ServiceBus/IServiceOriented.ServiceBus.csproj @@ -2,63 +2,66 @@ using System.Collections.Generic; using System.Linq; using System.Text; +using System.Runtime.Serialization; namespace IServiceOriented.ServiceBus -{ +{ + [Serializable] + [DataContract] public abstract class Listener : IDisposable { + [NonSerialized] + ServiceBusRuntime _runtime; public ServiceBusRuntime Runtime { - get; - private set; + get + { + return _runtime; + } + internal set + { + _runtime = value; + } } + [NonSerialized] + ListenerEndpoint _endpoint; public ListenerEndpoint Endpoint { - get; - private set; + get + { + return _endpoint; + } + internal set + { + _endpoint = value; + } } + [NonSerialized] + bool _started; public bool Started { - get; - private set; - } - - internal void StartInternal(ServiceBusRuntime host, ListenerEndpoint endpoint) - { - try + get { - Runtime = host; - Endpoint = endpoint; - - OnStart(); - - Started = true; + return _started; } - finally + private set { - if (!Started) - { - Runtime = null; - Endpoint = null; - } + _started = value; } } + internal void StartInternal() + { + OnStart(); + Started = true; + } + internal void StopInternal() { - try - { - OnStop(); - } - finally - { - Started = false; - - Runtime = null; - Endpoint = null; - } + OnStop(); + Started = false; } protected virtual void OnStart() IServiceOriented.ServiceBus/Listener.cs @@ -6,47 +6,35 @@ using System.Runtime.Serialization; namespace IServiceOriented.ServiceBus { - [Serializable] [DataContract] public sealed class ListenerEndpoint : Endpoint { - public ListenerEndpoint(Guid id, string name, string configurationName, string address, Type contractType, Type listenerType) : base(id, name, configurationName, address, contractType) + public ListenerEndpoint(Guid id, string name, string configurationName, string address, Type contractType, Listener listener) : base(id, name, configurationName, address, contractType) { - ListenerType = listenerType; - } - - public Type ListenerType - { - get; - set; + Listener = listener; } + Listener _listener; [DataMember] - public string ListenerTypeName + public Listener Listener { get { - if (ListenerType == null) return null; - return ListenerType.AssemblyQualifiedName; + return _listener; } - set + private set { - if (value == null) + if (value != null && value.Endpoint != null) { - ListenerType = null; + throw new InvalidOperationException("Endpoint is attached to another listener"); } - else + if (_listener != null) { - Type type = Type.GetType(value); - if (type == null) - { - throw new InvalidOperationException("Unknown type"); - } - ListenerType = type; - + _listener.Endpoint = null; } + _listener = value; + if(value != null) value.Endpoint = this; } } - } } IServiceOriented.ServiceBus/ListenerEndpoint.cs @@ -8,11 +8,7 @@ namespace IServiceOriented.ServiceBus [Serializable] public abstract class MessageFilter { - public abstract bool Include(string action, object message); - - protected internal abstract void InitFromString(string data); - protected internal abstract string CreateInitString(); - + public abstract bool Include(string action, object message); } IServiceOriented.ServiceBus/MessageFilter.cs @@ -5,69 +5,21 @@ using System.Linq; using System.Text; namespace IServiceOriented.ServiceBus -{ - public class MethodDispatcherConfiguration +{ + public sealed class MethodDispatcher<T> : Dispatcher { - Dictionary<Type, object> _targets = new Dictionary<Type, object>(); - - public void RegisterTarget(Type type, object target) - { - _targets.Add(type, target); - } - public void UnregisterTarget(Type type) + private MethodDispatcher() { - _targets.Remove(type); - } - public object GetTarget(Type type) - { - return _targets[type]; } - - static Dictionary<WeakReference, MethodDispatcherConfiguration> _configs = new Dictionary<WeakReference, MethodDispatcherConfiguration>(); - - - // Note: this is not optimized - public static MethodDispatcherConfiguration For(ServiceBusRuntime runtime) - { - lock (_configs) + public MethodDispatcher(object target) + { + if (target == null) { - List<WeakReference> kill = new List<WeakReference>(); // remove invalid references - foreach (WeakReference wr in _configs.Keys) - { - if (wr.IsAlive) - { - if (wr.Target == runtime) - { - return _configs[wr]; - } - } - else - { - kill.Add(wr); - } - } + throw new ArgumentNullException("target"); + } - foreach (WeakReference wr in kill) - { - _configs.Remove(wr); - } - - // Configuration not found - WeakReference newRef = new WeakReference(runtime); - MethodDispatcherConfiguration config = new MethodDispatcherConfiguration(); - _configs.Add(newRef, config); - return config; - } - } - - } - - public class MethodDispatcher<T> : Dispatcher - { - - public MethodDispatcher() - { + Target = target; foreach (MethodInfo method in typeof(T).GetMethods()) { if (_actionLookup.ContainsKey(method.Name)) @@ -78,6 +30,12 @@ namespace IServiceOriented.ServiceBus } } + public object Target + { + get; + private set; + } + protected override void Dispatch(SubscriptionEndpoint endpoint, string action, object message) { MethodInfo methodInfo; @@ -95,13 +53,12 @@ namespace IServiceOriented.ServiceBus if (methodInfo != null) { - methodInfo.Invoke(MethodDispatcherConfiguration.For(DispatchContext.Runtime).GetTarget(typeof(T)), new object[] { message }); + methodInfo.Invoke(Target, new object[] { message }); } else { throw new InvalidOperationException("Matching action not found"); } - } Dictionary<string, MethodInfo> _actionLookup = new Dictionary<string, MethodInfo>(); IServiceOriented.ServiceBus/MethodDispatcher.cs @@ -112,45 +112,7 @@ namespace IServiceOriented.ServiceBus ReaderWriterLockedObject<ReadOnlyCollection<RuntimeService>, IList<RuntimeService>> _runtimeServices = new ReaderWriterLockedObject<ReadOnlyCollection<RuntimeService>, IList<RuntimeService>>(new List<RuntimeService>(), l => new ReadOnlyCollection<RuntimeService>(l)); - Dictionary<ListenerEndpoint, Listener> _listeners = new Dictionary<ListenerEndpoint,Listener>(); - - void startDispatching(SubscriptionEndpoint se) - { - Dispatcher dispatcher = (Dispatcher)Activator.CreateInstance(se.DispatcherType); - dispatcher.StartInternal(this, se); - lock (_dispatchers) - { - _dispatchers.Add(se, dispatcher); - } - } - - void stopDispatching(SubscriptionEndpoint se) - { - lock (_dispatchers) - { - _dispatchers[se].StopInternal(); - _dispatchers.Remove(se); - } - } - - void startListening(ListenerEndpoint le) - { - Listener listener = (Listener)Activator.CreateInstance(le.ListenerType); - listener.StartInternal(this, le); - lock (_listeners) - { - _listeners.Add(le, listener); - } - } - - void stopListening(ListenerEndpoint le) - { - lock (_listeners) - { - _listeners[le].StopInternal(); - _listeners.Remove(le); - } - } + public void Start() { @@ -178,7 +140,7 @@ namespace IServiceOriented.ServiceBus { foreach (SubscriptionEndpoint se in subscriptions) { - startDispatching(se); + se.Dispatcher.StartInternal(); } }); @@ -186,7 +148,7 @@ namespace IServiceOriented.ServiceBus { foreach(ListenerEndpoint le in _listenerEndpoints) { - startListening(le); + le.Listener.StopInternal(); } } @@ -342,7 +304,7 @@ namespace IServiceOriented.ServiceBus { foreach (ListenerEndpoint le in _listenerEndpoints) { - stopListening(le); + le.Listener.StopInternal(); } } @@ -350,7 +312,7 @@ namespace IServiceOriented.ServiceBus { foreach (SubscriptionEndpoint se in subscriptions) { - stopDispatching(se); + se.Dispatcher.StopInternal(); } }); @@ -362,7 +324,6 @@ namespace IServiceOriented.ServiceBus } }); - _dispatchers.Clear(); _started = false; } @@ -439,6 +400,11 @@ namespace IServiceOriented.ServiceBus if (endpoint == null) throw new ArgumentNullException("endpoint"); + if (endpoint.Listener.Runtime != null) + { + throw new InvalidOperationException("Listener is attached to a bus already"); + } + VerifyContract(endpoint.ContractType); bool added = false; @@ -448,7 +414,8 @@ namespace IServiceOriented.ServiceBus { lock (_listenerEndpointsLock) { - _listenerEndpoints.Add(endpoint); + endpoint.Listener.Runtime = this; + _listenerEndpoints.Add(endpoint); added = true; } @@ -465,7 +432,7 @@ namespace IServiceOriented.ServiceBus if (_started) { - startListening(endpoint); + endpoint.Listener.StartInternal(); } ts.Complete(); @@ -496,7 +463,7 @@ namespace IServiceOriented.ServiceBus { if (_started) { - stopListening(endpoint); + endpoint.Listener.StopInternal(); } RemoveListener(endpoint); @@ -527,6 +494,7 @@ namespace IServiceOriented.ServiceBus if (endpoint == null) throw new ArgumentNullException("endpoint"); + bool removed = false; try { @@ -535,6 +503,7 @@ namespace IServiceOriented.ServiceBus lock(_listenerEndpointsLock) { _listenerEndpoints.Remove(endpoint); + endpoint.Listener.Runtime = null; } _runtimeServices.Read(runtimeServices => @@ -681,11 +650,12 @@ namespace IServiceOriented.ServiceBus try { - Dispatcher dispatcher; SubscriptionEndpoint endpoint = GetSubscription(delivery.SubscriptionEndpointId); + Dispatcher dispatcher = endpoint.Dispatcher; + if (endpoint != null) // subscriber might have been removed since enqueue { - _dispatchers.TryGetValue(endpoint, out dispatcher); + if (dispatcher != null) { @@ -737,9 +707,7 @@ namespace IServiceOriented.ServiceBus } } } - - Dictionary<SubscriptionEndpoint, Dispatcher> _dispatchers = new Dictionary<SubscriptionEndpoint, Dispatcher>(); - + /// <summary> /// Execute a block of code, passing any unhandled exceptions to the unhandled exception handler /// </summary> @@ -805,20 +773,11 @@ namespace IServiceOriented.ServiceBus return ForEachSafely(list, d => d.DynamicInvoke(parameters)); } - public void Publish(Type contract, string action, object message) + public void Publish(PublishRequest dispatchInformation) { if (_disposed) throw new ObjectDisposedException("ServiceBusRuntime"); - if (contract == null) - { - throw new ArgumentNullException("contract"); - } - - if (message == null) - { - throw new ArgumentNullException("message"); - } - VerifyContract(contract); + VerifyContract(dispatchInformation.Contract); _subscriptions.Read(subscriptions => { @@ -835,7 +794,7 @@ namespace IServiceOriented.ServiceBus if (subscription.Filter is UnhandledMessageFilter) { include = false; - if (subscription.Filter.Include(action, message)) + if (subscription.Filter.Include(dispatchInformation.Action, dispatchInformation.Message)) { unhandledFilters.Add(subscription); } @@ -843,7 +802,7 @@ namespace IServiceOriented.ServiceBus } else { - include = subscription.Filter.Include(action, message); + include = subscription.Filter.Include(dispatchInformation.Action, dispatchInformation.Message); } } else @@ -853,7 +812,7 @@ namespace IServiceOriented.ServiceBus if (include) { - QueueDelivery(subscription.Id, action, message); + QueueDelivery(subscription.Id, dispatchInformation.Action, dispatchInformation.Message); handled = true; } } @@ -863,7 +822,7 @@ namespace IServiceOriented.ServiceBus { foreach (SubscriptionEndpoint subscription in unhandledFilters) { - QueueDelivery(subscription.Id, action, message); + QueueDelivery(subscription.Id, dispatchInformation.Action, dispatchInformation.Message); } } @@ -912,6 +871,11 @@ namespace IServiceOriented.ServiceBus if (subscription == null) throw new ArgumentNullException("subscription"); + if (subscription.Dispatcher != null) + { + throw new InvalidOperationException("Subscription is attached to another bus"); + } + VerifyContract(subscription.ContractType); bool added = false; @@ -922,12 +886,13 @@ namespace IServiceOriented.ServiceBus _subscriptions.Write(subscriptions => { subscriptions.Add(subscription); + subscription.Dispatcher.Runtime = this; added = true; }); if (_started) { - startDispatching(subscription); + subscription.Dispatcher.StopInternal() ; } _runtimeServices.Read(runtimeServices => @@ -987,12 +952,13 @@ namespace IServiceOriented.ServiceBus try { using (TransactionScope ts = new TransactionScope()) - { - stopDispatching(subscription); + { + subscription.Dispatcher.StopInternal(); _subscriptions.Write(subscriptions => { subscriptions.Remove(subscription); + subscription.Dispatcher.Runtime = null; removed = true; }); IServiceOriented.ServiceBus/ServiceBusRuntime.cs @@ -12,12 +12,67 @@ using System.Data.SqlClient; using Microsoft.SqlServer.Management.Common; using Microsoft.SqlServer.Management.Smo; +using System.Runtime.Serialization; + namespace IServiceOriented.ServiceBus { public class SqlSubscriptionDB { - public SqlSubscriptionDB(string connectionString) + public SqlSubscriptionDB(Type[] knownDispatcherTypes, Type[] knownListenerTypes, Type[] knownFilterTypes) + { + _dispatcherSerializer = new DataContractSerializer(typeof(Dispatcher), knownDispatcherTypes); + _listenerSerializer = new DataContractSerializer(typeof(Listener), knownListenerTypes); + _filterSerializer = new DataContractSerializer(typeof(MessageFilter), knownFilterTypes); + } + + DataContractSerializer _dispatcherSerializer; + DataContractSerializer _listenerSerializer; + DataContractSerializer _filterSerializer; + + Listener getListenerFromPersistenceData(byte[] data) + { + MemoryStream ms = new MemoryStream(data); + ms.Position = 0; + return (Listener)_listenerSerializer.ReadObject(ms); + } + + byte[] getListenerPersistenceData(Listener obj) + { + MemoryStream ms = new MemoryStream(); + _listenerSerializer.WriteObject(ms,obj); + return ms.ToArray(); + } + + MessageFilter getFilterFromPersistenceData(byte[] data) + { + MemoryStream ms = new MemoryStream(data); + ms.Position = 0; + return (MessageFilter)_filterSerializer.ReadObject(ms); + } + + byte[] getFilterPersistenceData(MessageFilter obj) + { + MemoryStream ms = new MemoryStream(); + _filterSerializer.WriteObject(ms, obj); + return ms.ToArray(); + } + + + Dispatcher getDispatcherFromPersistenceData(byte[] data) + { + MemoryStream ms = new MemoryStream(data); + ms.Position = 0; + return (Dispatcher)_dispatcherSerializer.ReadObject(ms); + } + + byte[] getDispatcherPersistenceData(Dispatcher obj) + { + MemoryStream ms = new MemoryStream(); + _dispatcherSerializer.WriteObject(ms, obj); + return ms.ToArray(); + } + public SqlSubscriptionDB(string connectionString, Type[] knownDispatcherTypes, Type[] knownListenerTypes, Type[] knownFilterTypes) : this(knownDispatcherTypes, knownListenerTypes, knownFilterTypes) { _connectionString = connectionString; } @@ -58,33 +113,16 @@ namespace IServiceOriented.ServiceBus } - static SubscriptionEndpoint getSubscription(IDataReader dr) - { - MessageFilter filter = null; - - string filterData = dr["filter_data"] as string; - string filterTypeString = dr["filter_type"] as string; - - if (filterTypeString != null) - { - Type filterType = Type.GetType(filterTypeString); - if (filterType == null) - { - throw new InvalidOperationException("Filter type could not be loaded"); - } - - filter = (MessageFilter)Activator.CreateInstance(filterType); - filter.InitFromString(filterData); - } - + SubscriptionEndpoint getSubscription(IDataReader dr) + { SubscriptionEndpoint endpoint = new SubscriptionEndpoint( (Guid)dr["id"], dr["name"] as string, dr["configuration_name"] as string, dr["address"] as string, Type.GetType(dr["contract_type"] as string), - Type.GetType(dr["dispatcher_type"] as string), - filter) + getDispatcherFromPersistenceData((byte[])dr["dispatcher_data"]), + getFilterFromPersistenceData((byte[])dr["filter_data"]) ) ; if (endpoint.ContractType == null) @@ -95,7 +133,7 @@ namespace IServiceOriented.ServiceBus return endpoint; } - static ListenerEndpoint getListener(IDataReader dr) + ListenerEndpoint getListener(IDataReader dr) { ListenerEndpoint endpoint = new ListenerEndpoint( (Guid)dr["id"], @@ -103,8 +141,7 @@ namespace IServiceOriented.ServiceBus dr["configuration_name"] as string, dr["address"] as string, Type.GetType(dr["contract_type"] as string), - Type.GetType(dr["listener_type"] as string) - ); + getListenerFromPersistenceData((byte[])dr["listener_data"])); if (endpoint.ContractType == null) { @@ -169,7 +206,7 @@ namespace IServiceOriented.ServiceBus command.Parameters.AddWithValue("@configuration_name", endpoint.ConfigurationName); command.Parameters.AddWithValue("@contract_type", endpoint.ContractTypeName); command.Parameters.AddWithValue("@name", endpoint.Name); - command.Parameters.AddWithValue("@listener_type", endpoint.ListenerTypeName); + command.Parameters.AddWithValue("@listener_data", getListenerPersistenceData(endpoint.Listener)); command.ExecuteNonQuery(); } } @@ -183,13 +220,12 @@ namespace IServiceOriented.ServiceBus { command.CommandType = CommandType.StoredProcedure; command.Parameters.AddWithValue("@id", subscription.Id); - command.Parameters.AddWithValue("@filter_type", subscription.Filter == null ? DBNull.Value : (object)subscription.Filter.GetType().AssemblyQualifiedName); - command.Parameters.AddWithValue("@filter_data", subscription.Filter == null ? DBNull.Value : (object)subscription.Filter.CreateInitString()); + command.Parameters.AddWithValue("@filter_data", getFilterPersistenceData(subscription.Filter)); command.Parameters.AddWithValue("@address", subscription.Address); command.Parameters.AddWithValue("@configuration_name", subscription.ConfigurationName); command.Parameters.AddWithValue("@contract_type", subscription.ContractTypeName); command.Parameters.AddWithValue("@name", subscription.Name); - command.Parameters.AddWithValue("@dispatcher_type", subscription.DispatcherTypeName); + command.Parameters.AddWithValue("@dispatcher_data", getDispatcherPersistenceData(subscription.Dispatcher)); command.ExecuteNonQuery(); } } @@ -230,13 +266,12 @@ namespace IServiceOriented.ServiceBus public class SqlSubscriptionPersistenceService : SubscriptionPersistenceService { - public SqlSubscriptionPersistenceService(string connectionString) + public SqlSubscriptionPersistenceService(string connectionString, Type[] knownDispatcherTypes, Type[] knownListenerTypes, Type[] knownFilterTypes) { - _db = new SqlSubscriptionDB(connectionString); + _db = new SqlSubscriptionDB(connectionString, knownDispatcherTypes, knownListenerTypes, knownFilterTypes); } SqlSubscriptionDB _db; - protected override IEnumerable<Endpoint> LoadEndpoints() { @@ -249,22 +284,34 @@ namespace IServiceOriented.ServiceBus protected override void CreateListener(ListenerEndpoint endpoint) { - _db.CreateListener(endpoint); + if (!endpoint.Transient) + { + _db.CreateListener(endpoint); + } } protected override void CreateSubscription(SubscriptionEndpoint subscription) { - _db.CreateSubscription(subscription); + if (!subscription.Transient) + { + _db.CreateSubscription(subscription); + } } protected override void DeleteListener(ListenerEndpoint endpoint) { - _db.DeleteListener(endpoint.Id); + if (!endpoint.Transient) + { + _db.DeleteListener(endpoint.Id); + } } protected override void DeleteSubscription(SubscriptionEndpoint subscription) { - _db.DeleteSubscription(subscription.Id); + if (!subscription.Transient) + { + _db.DeleteSubscription(subscription.Id); + } } } } IServiceOriented.ServiceBus/SqlSubscriptionPersistenceService.cs @@ -11,18 +11,18 @@ namespace IServiceOriented.ServiceBus [DataContract] public sealed class SubscriptionEndpoint : Endpoint { - public SubscriptionEndpoint(Guid id, string name, string configurationName, string address, Type contractType, Type dispatcherType, MessageFilter filter) - : base(id, name, configurationName, address, contractType) + public SubscriptionEndpoint(Guid id, string name, string configurationName, string address, Type contractType, Dispatcher dispatcher, MessageFilter filter) + : base(id, name, configurationName, address, contractType) { Filter = filter; - DispatcherType = dispatcherType; + Dispatcher = dispatcher; } - - public SubscriptionEndpoint(Guid id, string name, string configurationName, string address, string contractTypeName, string dispatcherTypeName, MessageFilter filter) + + public SubscriptionEndpoint(Guid id, string name, string configurationName, string address, string contractTypeName, Dispatcher dispatcher, MessageFilter filter) : base(id, name, configurationName, address, contractTypeName) { Filter = filter; - DispatcherTypeName = dispatcherTypeName; + Dispatcher = dispatcher; } MessageFilter _filter; @@ -38,34 +38,27 @@ namespace IServiceOriented.ServiceBus _filter = value; } } - - Type _dispatcherType; - public Type DispatcherType - { - get - { - return _dispatcherType; - } - private set - { - _dispatcherType = value; - } - } + Dispatcher _dispatcher; [DataMember] - public string DispatcherTypeName + public Dispatcher Dispatcher { get { - return _dispatcherType.AssemblyQualifiedName; + return _dispatcher; } private set { - _dispatcherType = Type.GetType(value); - if (_dispatcherType == null) + if (value != null && value.Endpoint != null) + { + throw new InvalidOperationException("Endpoint is attached to another dispatcher"); + } + if (_dispatcher != null) { - throw new InvalidOperationException(value + " is an invalid type"); + _dispatcher.Endpoint = null; } + _dispatcher = value; + if (value != null) value.Endpoint = this; } } } IServiceOriented.ServiceBus/SubscriptionEndpoint.cs @@ -23,34 +23,7 @@ namespace IServiceOriented.ServiceBus } const char TYPE_SEPERATOR = ':'; - - protected internal override string CreateInitString() - { - StringBuilder sb = new StringBuilder(); - for (int i = 0; i < _messageTypes.Length; i++) - { - if (i > 0) - { - sb.Append(TYPE_SEPERATOR); - } - sb.Append(_messageTypes[i]); - } - return sb.ToString(); - } - - protected internal override void InitFromString(string data) - { - string[] typeNames = data.Split(TYPE_SEPERATOR); - Type[] types = new Type[typeNames.Length]; - for (int i = 0; i < typeNames.Length; i++) - { - Type t = Type.GetType(typeNames[i]); - if(t == null) throw new InvalidOperationException("Unknown type"); - types[i] = t; - } - _messageTypes = types; - } - + [DataMember] protected IEnumerable<string> MessageTypeNames { IServiceOriented.ServiceBus/TypedMessageFilter.cs @@ -17,16 +17,7 @@ namespace IServiceOriented.ServiceBus : base(messageType) { } - protected internal override string CreateInitString() - { - return null; - } - - protected internal override void InitFromString(string data) - { - - } - + public override bool Include(string action, object message) { return base.Include(action, message); IServiceOriented.ServiceBus/UnhandledMessageFilter.cs @@ -5,12 +5,21 @@ using System.Text; using System.ServiceModel; using System.Reflection; +using System.Runtime.Serialization; + namespace IServiceOriented.ServiceBus { + [Serializable] + [DataContract] public class WcfDispatcher<T> : Dispatcher { public WcfDispatcher() { + initActionLookup(); + } + + void initActionLookup() + { foreach (MethodInfo method in typeof(T).GetMethods()) { object[] attributes = method.GetCustomAttributes(typeof(OperationContractAttribute), false); @@ -56,6 +65,7 @@ namespace IServiceOriented.ServiceBus } } + [NonSerialized] Dictionary<string, MethodInfo> _actionLookup = new Dictionary<string, MethodInfo>(); } IServiceOriented.ServiceBus/WcfDispatcher.cs @@ -5,8 +5,12 @@ using System.Linq; using System.Text; using System.ServiceModel; +using System.Runtime.Serialization; + namespace IServiceOriented.ServiceBus { + [Serializable] + [DataContract] public class WcfListener<T> : Listener { protected override void OnStart() @@ -21,6 +25,7 @@ namespace IServiceOriented.ServiceBus _host = null; } + [NonSerialized] ServiceHost _host; protected override void Dispose(bool disposing) IServiceOriented.ServiceBus/WcfListener.cs @@ -11,6 +11,7 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution IServiceOriented1.vsmdi = IServiceOriented1.vsmdi IServiceOriented2.vsmdi = IServiceOriented2.vsmdi IServiceOriented3.vsmdi = IServiceOriented3.vsmdi + IServiceOriented4.vsmdi = IServiceOriented4.vsmdi LocalTestRun.testrunconfig = LocalTestRun.testrunconfig EndProjectSection EndProject @@ -19,6 +20,9 @@ EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "IServiceOriented.ServiceBus.Samples.Chat", "IServiceOriented.ServiceBus.Samples.Chat\IServiceOriented.ServiceBus.Samples.Chat.csproj", "{4587E682-FDB8-4C34-8448-48F1F6921591}" EndProject Global + GlobalSection(TestCaseManagementSettings) = postSolution + CategoryFile = IServiceOriented4.vsmdi + EndGlobalSection GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU Release|Any CPU = Release|Any CPU IServiceOriented.sln cd00bc4b636e93f9bc69fdc77fda18fd8442a95a JEzell jezell@gmail.com http://github.com/jezell/iserviceoriented/commit/ede51630428ab6d1174ebcaf06d5c7306b051862 ede51630428ab6d1174ebcaf06d5c7306b051862 2008-08-30T15:52:13-07:00 2008-08-30T15:52:13-07:00 Message transforms / Useability improvements 2ed512ddf554061d44295e3f4896851c7cd7156d JEzell jezell@gmail.com