IServiceOriented.ServiceBus/DuplicateIdentifierException.cs IServiceOriented.ServiceBus/ListenerEndpointCollection.cs IServiceOriented.ServiceBus/SubscriptionEndpointCollection.cs @@ -15,9 +15,7 @@ namespace IServiceOriented.ServiceBus.Samples.Chat MessageDelivery.RegisterKnownType(typeof(ChatFilter2)); MessageDelivery.RegisterKnownType(typeof(SendMessageRequest)); MessageDelivery.RegisterKnownType(typeof(SendMessageRequest2)); - MessageDelivery.RegisterKnownType(typeof(WcfListener)); - MessageDelivery.RegisterKnownType(typeof(WcfDispatcher)); - + if(args.Length == 0) { Console.WriteLine("usage: Chat.exe [server | client]"); IServiceOriented.ServiceBus.Samples.Chat/Program.cs @@ -132,7 +132,7 @@ namespace IServiceOriented.ServiceBus.UnitTests SubscriptionEndpoint endpoint = new SubscriptionEndpoint(Guid.NewGuid(), "SubscriptionName", "http://localhost/test", "SubscriptionConfigName", typeof(IContract), new WcfDispatcher(), new PassThroughMessageFilter()); - MessageDelivery enqueued = new MessageDelivery(endpoint.Id, "randomAction","randomMessageData", 3, new ReadOnlyDictionary<string,object>()); + MessageDelivery enqueued = new MessageDelivery(endpoint.Id, typeof(MessageDelivery), "randomAction","randomMessageData", 3, new ReadOnlyDictionary<string,object>()); Assert.IsNull(queue.Peek(TimeSpan.FromSeconds(1)), "Queue must be empty to start transaction test"); // Enqueue, but abort transaction @@ -196,7 +196,7 @@ namespace IServiceOriented.ServiceBus.UnitTests SubscriptionEndpoint endpoint = new SubscriptionEndpoint(Guid.NewGuid(), "SubscriptionName", "http://localhost/test", "SubscriptionConfigName", typeof(IContract), new WcfDispatcher(), new PassThroughMessageFilter()); - MessageDelivery enqueued = new MessageDelivery(endpoint.Id, messageAction, messageData, 3, new ReadOnlyDictionary<string,object>()); + MessageDelivery enqueued = new MessageDelivery(endpoint.Id, typeof(MessageDelivery), messageAction, messageData, 3, new ReadOnlyDictionary<string,object>()); using (TransactionScope ts = new TransactionScope()) { IServiceOriented.ServiceBus.UnitTests/MsmqMessageDeliveryQueueTest.cs @@ -65,7 +65,7 @@ namespace IServiceOriented.ServiceBus.UnitTests void putMessageAndGet(object message) { SubscriptionEndpoint subscription = new SubscriptionEndpoint(Guid.NewGuid(), "test", "configurationName", "address", typeof(IServiceBusManagementService), new WcfDispatcher(), null); - MessageDelivery failure = new MessageDelivery(subscription.Id, "action", message, 3, new ReadOnlyDictionary<string,object>()); + MessageDelivery failure = new MessageDelivery(subscription.Id, typeof(MessageDelivery), "action", message, 3, new ReadOnlyDictionary<string,object>()); NonTransactionalMemoryQueue failureQueue = new NonTransactionalMemoryQueue(); failureQueue.Enqueue(failure); IServiceOriented.ServiceBus.UnitTests/ServiceBusManagementTest.cs @@ -7,6 +7,9 @@ using System.Threading; namespace IServiceOriented.ServiceBus { + /// <summary> + /// Used for countdown events. When the count reaches zero, the event will be set. + /// </summary> public class CountdownLatch : IDisposable { public CountdownLatch(int count) @@ -17,8 +20,13 @@ namespace IServiceOriented.ServiceBus int _count; ManualResetEvent _handle = new ManualResetEvent(false); + /// <summary> + /// Decrement the count. Set the event to signalled if the count is zero; + /// </summary> + /// <returns>The new count</returns> public int Tick() { + if (_disposed) throw new ObjectDisposedException("CoundownLatch"); int count = Interlocked.Decrement(ref _count); if (count == 0) { @@ -31,12 +39,24 @@ namespace IServiceOriented.ServiceBus { get { + if (_disposed) throw new ObjectDisposedException("CoundownLatch"); return _handle; } } volatile bool _disposed; + /// <summary> + /// Reset the count and set the event to non signalled. + /// </summary> + public void Reset(int count) + { + if (_disposed) throw new ObjectDisposedException("CoundownLatch"); + + _count = count; + _handle.Reset(); + } + protected virtual void Dispose(bool disposing) { if (disposing && !_disposed) @@ -56,5 +76,10 @@ namespace IServiceOriented.ServiceBus GC.SuppressFinalize(this); } + + ~CountdownLatch() + { + Dispose(false); + } } } IServiceOriented.ServiceBus/CountdownLatch.cs @@ -8,12 +8,21 @@ using System.Runtime.Serialization; namespace IServiceOriented.ServiceBus { + /// <summary> + /// Base class for all service bus dispatchers. + /// </summary> + /// <remarks> + /// Dispatchers are responsible for delivering messages to subscription endpoints. + /// </remarks> [Serializable] [DataContract] public abstract class Dispatcher : IDisposable { [NonSerialized] ServiceBusRuntime _runtime; + /// <summary> + /// Gets the service bus instance associated with this dispatcher. + /// </summary> public ServiceBusRuntime Runtime { get @@ -28,6 +37,9 @@ namespace IServiceOriented.ServiceBus [NonSerialized] SubscriptionEndpoint _endpoint; + /// <summary> + /// Gets the subscription endpoint associated with this dispatcher. + /// </summary> public SubscriptionEndpoint Endpoint { get @@ -103,6 +115,10 @@ namespace IServiceOriented.ServiceBus /// </summary> protected abstract void Dispatch(SubscriptionEndpoint endpoint, string action, object message); + /// <summary> + /// Dispose any resources held by this dispatcher. + /// </summary> + /// <param name="disposing">Indicates whether this method is being called as a result of a direct call to Dispose (true) or by the object's finalizer (false).</param> protected virtual void Dispose(bool disposing) { @@ -117,6 +133,11 @@ namespace IServiceOriented.ServiceBus } #endregion + + ~Dispatcher() + { + Dispose(false); + } } IServiceOriented.ServiceBus/Dispatcher.cs @@ -8,6 +8,9 @@ using System.Runtime.Serialization; namespace IServiceOriented.ServiceBus { + /// <summary> + /// Base class for service bus endpoints. + /// </summary> [Serializable] [DataContract] public abstract class Endpoint @@ -31,6 +34,10 @@ namespace IServiceOriented.ServiceBus } Guid _endpointId = Guid.NewGuid(); + + /// <summary> + /// Gets the unique identifier of the endpoint. + /// </summary> [DataMember] public Guid Id { @@ -43,7 +50,11 @@ namespace IServiceOriented.ServiceBus _endpointId = value; } } + Type _contractType; + /// <summary> + /// Gets the Type of the contract supported by the endpoint. + /// </summary> public Type ContractType { get @@ -56,6 +67,9 @@ namespace IServiceOriented.ServiceBus } } + /// <summary> + /// Gets the name of the contract type supported by the endpoint. + /// </summary> [DataMember] public string ContractTypeName { @@ -74,6 +88,9 @@ namespace IServiceOriented.ServiceBus } string _address; + /// <summary> + /// Gets the address of the endpoint. + /// </summary> [DataMember] public string Address { @@ -88,6 +105,9 @@ namespace IServiceOriented.ServiceBus } string _configurationName; + /// <summary> + /// Gets the name of the configuration used by the endpoint. + /// </summary> [DataMember] public string ConfigurationName { @@ -102,6 +122,9 @@ namespace IServiceOriented.ServiceBus } string _name; + /// <summary> + /// Gets the name of this endpoint. + /// </summary> [DataMember] public string Name { @@ -116,6 +139,9 @@ namespace IServiceOriented.ServiceBus } bool _transient; + /// <summary> + /// Gets a boolean value indicating whether this endpoint is transient (true) or should be persisted (false). + /// </summary> [DataMember] public bool Transient { @@ -128,8 +154,5 @@ namespace IServiceOriented.ServiceBus _transient = value; } } - } - - - + } } IServiceOriented.ServiceBus/Endpoint.cs @@ -7,11 +7,28 @@ using System.Reflection; namespace IServiceOriented.ServiceBus { + /// <summary> + /// Gaurds an object with locks. + /// </summary> + /// <typeparam name="TRead">Read only type of the contained data.</typeparam> + /// <typeparam name="TWrite">Writeable type of the contained data.</typeparam> public interface ILockedObject<TRead, TWrite> : IDisposable { + /// <summary> + /// Access the contained value for read access. + /// </summary> + /// <param name="action"></param> void Read(Action<TRead> action); + /// <summary> + /// Access the contained value for write access. + /// </summary> + /// <param name="action"></param> void Write(Action<TWrite> action); + /// <summary> + /// Access the contained value with write access and set to a new value. + /// </summary> + /// <param name="value"></param> void SetValue(TWrite value); } IServiceOriented.ServiceBus/ILockedObject.cs @@ -6,35 +6,64 @@ using System.Text; namespace IServiceOriented.ServiceBus { + /// <summary> + /// A read-only dictionary. + /// </summary> + /// <typeparam name="K">Key type</typeparam> + /// <typeparam name="V">Value type</typeparam> public interface IReadOnlyDictionary<K,V> : IEnumerable<KeyValuePair<K,V>> { + /// <summary> + /// Get the keys associated with this dictionary instance. + /// </summary> IEnumerable<K> Keys { get; } + /// <summary> + /// Get the values contained by this dictionary instance. + /// </summary> IEnumerable<V> Values { get; } + /// <summary> + /// Get the value associated with a specific key. + /// </summary> V this[K key] { get; } + /// <summary> + /// Get the number of items contained by this dictionary. + /// </summary> int Count { get; } + /// <summary> + /// Check to see if the dictionary contains a specific key. + /// </summary> bool ContainsKey(K key); - bool Contains(KeyValuePair<K,V> value); + /// <summary> + /// Check to see if the dictionary contains a specific key value pair. + /// </summary> + bool Contains(KeyValuePair<K, V> value); } + /// <summary> + /// Extension methods for read only dicionaries + /// </summary> public static class ReadOnlyDictionaryExtensions { + /// <summary> + /// Convert a read only dictionary to a writeable dictionary. + /// </summary> public static Dictionary<K, V> ToDictionary<K,V>(this IReadOnlyDictionary<K, V> readOnlyDictionary) { Dictionary<K, V> dict = new Dictionary<K, V>(); @@ -45,11 +74,19 @@ namespace IServiceOriented.ServiceBus return dict; } + /// <summary> + /// Convert a dictionary to a read only dictionary. + /// </summary> public static ReadOnlyDictionary<K, V> MakeReadOnly<K, V>(this IDictionary<K, V> dictionary) { return dictionary.MakeReadOnly(false); } - public static ReadOnlyDictionary<K, V> MakeReadOnly<K,V>(this IDictionary<K, V> dictionary, bool copy) + + /// <summary> + /// Convert a dictionary to a read only dictionary. + /// </summary> + /// <param name="copy">Specifies if the dictionary should be copied (true) or wrapped (false).</param> + public static ReadOnlyDictionary<K, V> MakeReadOnly<K, V>(this IDictionary<K, V> dictionary, bool copy) { return new ReadOnlyDictionary<K, V>(dictionary, copy); } IServiceOriented.ServiceBus/IReadOnlyDictionary.cs @@ -62,6 +62,7 @@ <ItemGroup> <Compile Include="CountdownLatch.cs" /> <Compile Include="Dispatcher.cs" /> + <Compile Include="DuplicateIdentifierException.cs" /> <Compile Include="Endpoint.cs" /> <Compile Include="IReadOnlyDictionary.cs" /> <Compile Include="Listener.cs" /> @@ -69,11 +70,13 @@ <Compile Include="InvalidContractException.cs" /> <Compile Include="ListenerEndpoint.cs" /> <Compile Include="IMessageDeliveryQueue.cs" /> + <Compile Include="ListenerEndpointCollection.cs" /> <Compile Include="PublishRequest.cs" /> <Compile Include="ReadOnlyDictionary.cs" /> <Compile Include="SqlSubscriptionPersistenceService.cs"> <SubType>Code</SubType> </Compile> + <Compile Include="SubscriptionEndpointCollection.cs" /> <Compile Include="TransformationDispatcher.cs" /> <Compile Include="ReaderWriterLockedObject.cs" /> <Compile Include="UnhandledMessageFilter.cs" /> IServiceOriented.ServiceBus/IServiceOriented.ServiceBus.csproj @@ -6,12 +6,21 @@ using System.Runtime.Serialization; namespace IServiceOriented.ServiceBus { + /// <summary> + /// Base class for all service host listeners. + /// </summary> + /// <remarks> + /// Listeners are responsible for receiving messages and publishing them to the bus. + /// </remarks> [Serializable] [DataContract] public abstract class Listener : IDisposable { [NonSerialized] ServiceBusRuntime _runtime; + /// <summary> + /// Gets the service bus instance that this listener is associated with. + /// </summary> public ServiceBusRuntime Runtime { get @@ -26,6 +35,9 @@ namespace IServiceOriented.ServiceBus [NonSerialized] ListenerEndpoint _endpoint; + /// <summary> + /// Gets the listener endpoint that this listener is associated with. + /// </summary> public ListenerEndpoint Endpoint { get @@ -40,6 +52,9 @@ namespace IServiceOriented.ServiceBus [NonSerialized] bool _started; + /// <summary> + /// Gets a boolean value indicating whether this listener has been started. + /// </summary> public bool Started { get @@ -64,14 +79,24 @@ namespace IServiceOriented.ServiceBus Started = false; } + /// <summary> + /// Perform any actions that should be performed when this listener starts. + /// </summary> protected virtual void OnStart() { } + /// <summary> + /// Perform any actions that should be performed when this listener stops. + /// </summary> protected virtual void OnStop() { } + /// <summary> + /// Dispose any resources held by this listener. + /// </summary> + /// <param name="disposing">Indicates whether this method is being called as a result of a direct call to Dispose (true) or by the object's finalizer (false).</param> protected virtual void Dispose(bool disposing) { @@ -86,6 +111,11 @@ namespace IServiceOriented.ServiceBus } #endregion + + ~Listener() + { + Dispose(false); + } } } IServiceOriented.ServiceBus/Listener.cs @@ -6,15 +6,24 @@ using System.Runtime.Serialization; namespace IServiceOriented.ServiceBus { + /// <summary> + /// Represents a listener endpoint. + /// </summary> [DataContract] public sealed class ListenerEndpoint : Endpoint { public ListenerEndpoint(Guid id, string name, string configurationName, string address, Type contractType, Listener listener) : base(id, name, configurationName, address, contractType) - { + { Listener = listener; } Listener _listener; + /// <summary> + /// The listener used by this endpoint + /// </summary> + /// <remarks> + /// Listeners cannot be shared by multiple endpoints. + /// </remarks> [DataMember] public Listener Listener { IServiceOriented.ServiceBus/ListenerEndpoint.cs @@ -8,12 +8,15 @@ using System.Runtime.Serialization; namespace IServiceOriented.ServiceBus { + /// <summary> + /// Represents a message to be delivered. + /// </summary> [Serializable] [DataContract] [KnownType("GetKnownTypes")] public class MessageDelivery { - public MessageDelivery(Guid subscriptionEndpointId, string action, object message, int maxRetries, ReadOnlyDictionary<string,object> context) + public MessageDelivery(Guid subscriptionEndpointId, Type contractType, string action, object message, int maxRetries, ReadOnlyDictionary<string,object> context) { _messageId = Guid.NewGuid().ToString(); _subscriptionEndpointId = subscriptionEndpointId; @@ -21,9 +24,10 @@ namespace IServiceOriented.ServiceBus _message = message; _maxRetries = maxRetries; _context = context; + ContractType = contractType; } - public MessageDelivery(string messageId, Guid subscriptionEndpointId, string action, object message, int maxRetries, int retryCount, DateTime? timeToProcess, int queueCount, ReadOnlyDictionary<string, object> context) + public MessageDelivery(string messageId, Guid subscriptionEndpointId, Type contractType, string action, object message, int maxRetries, int retryCount, DateTime? timeToProcess, int queueCount, ReadOnlyDictionary<string, object> context) { _messageId = messageId; _subscriptionEndpointId = subscriptionEndpointId; @@ -34,8 +38,12 @@ namespace IServiceOriented.ServiceBus _queueCount = queueCount; _maxRetries = maxRetries; _context = context; + ContractType = contractType; } + /// <summary> + /// Gets a list of types that have been registered for message delivery. + /// </summary> public static Type[] GetKnownTypes() { lock (_knownTypes) @@ -44,7 +52,10 @@ namespace IServiceOriented.ServiceBus } } - static List<Type> _knownTypes = new List<Type>(new Type[] { typeof(UnhandledMessageFilter), typeof(Guid[]) } ); + static List<Type> _knownTypes = new List<Type>(new Type[] { typeof(UnhandledMessageFilter), typeof(Guid[]), typeof(WcfListener), typeof(WcfDispatcher), typeof(TypedMessageFilter) } ); + /// <summary> + /// Clears the list of types registered for message delivery. + /// </summary> public static void ClearKnownTypes() { lock (_knownTypes) @@ -53,6 +64,9 @@ namespace IServiceOriented.ServiceBus } } + /// <summary> + /// Register a type for message delivery. + /// </summary> public static void RegisterKnownType(Type type) { lock (_knownTypes) @@ -64,6 +78,9 @@ namespace IServiceOriented.ServiceBus } } + /// <summary> + /// Unregister a type for message delivery. + /// </summary> public static void UnregisterKnownType(Type type) { lock (_knownTypes) @@ -77,6 +94,9 @@ namespace IServiceOriented.ServiceBus int _queueCount; + /// <summary> + /// The number of times that this message has been queued + /// </summary> [DataMember] public int QueueCount { @@ -90,6 +110,11 @@ namespace IServiceOriented.ServiceBus } } + /// <summary> + /// Increment the queue count. + /// </summary> + /// <returns>The new queue count.</returns> + /// <remarks>This method is not thread safe.</remarks> public int IncrementQueueCount() { return _queueCount++; @@ -97,6 +122,9 @@ namespace IServiceOriented.ServiceBus private string _messageId; + /// <summary> + /// Gets the unique identifier of this message + /// </summary> [DataMember] public string MessageId { @@ -105,14 +133,65 @@ namespace IServiceOriented.ServiceBus } private Guid _subscriptionEndpointId; + /// <summary> + /// Gets the identifier of the subscription that this message is associated with. + /// </summary> [DataMember] public Guid SubscriptionEndpointId { get { return _subscriptionEndpointId; } private set { _subscriptionEndpointId = value; } - } + } + + private Type _contractType; + /// <summary> + /// Gets the type of contract associated with this message delivery. + /// </summary> + public Type ContractType + { + get + { + return _contractType; + } + set + { + _contractType = value; + } + } + + /// <summary> + /// Gets the name of the contract type associated with this message delivery. + /// </summary> + [DataMember] + public string ContractTypeName + { + get + { + if (_contractType == null) return null; + return _contractType.AssemblyQualifiedName; + } + set + { + if (value == null) + { + ContractType = null; + } + else + { + Type type = Type.GetType(value); + if (type == null) + { + throw new InvalidOperationException("The type specified does not exist"); + } + ContractType = type; + } + } + } private string _action; + /// <summary> + /// Gets the action associated with this message. + /// </summary> [DataMember] public string Action { @@ -121,7 +200,9 @@ namespace IServiceOriented.ServiceBus } private object _message; - + /// <summary> + /// Gets the message contents. + /// </summary> [DataMember] public object Message { @@ -130,6 +211,9 @@ namespace IServiceOriented.ServiceBus } private int _retryCount; + /// <summary> + /// Gets the number of times this message has been retried. + /// </summary> [DataMember] public int RetryCount { @@ -138,6 +222,9 @@ namespace IServiceOriented.ServiceBus } private DateTime? _timeToProcess; + /// <summary> + /// Gets the time that this message should be processed (or null if immediately). + /// </summary> [DataMember] public DateTime? TimeToProcess { @@ -146,6 +233,9 @@ namespace IServiceOriented.ServiceBus } + /// <summary> + /// Gets the maximum number of times this message will be retried. + /// </summary> [DataMember] public int MaxRetries { @@ -154,6 +244,9 @@ namespace IServiceOriented.ServiceBus } + /// <summary> + /// Gets a boolean value indicating whether the maximum number of retries has been met. + /// </summary> public bool RetriesMaxed { get @@ -164,6 +257,9 @@ namespace IServiceOriented.ServiceBus ReadOnlyDictionary<string, object> _context = new ReadOnlyDictionary<string, object>(); + /// <summary> + /// Gets the context associated with this message. + /// </summary> [DataMember] public ReadOnlyDictionary<string, object> Context { @@ -179,10 +275,16 @@ namespace IServiceOriented.ServiceBus private int _maxRetries; + /// <summary> + /// Create a retry message based off of this message. + /// </summary> + /// <param name="resetRetryCount">Whether or not to reset the retry count.</param> + /// <param name="timeToDeliver">Time to deliver the retry message.</param> + /// <returns>A new MessageDelivery.</returns> public MessageDelivery CreateRetry(bool resetRetryCount, DateTime timeToDeliver) { int retryCount = resetRetryCount ? 0 : (_retryCount + 1); - return new MessageDelivery(_messageId, _subscriptionEndpointId, _action, _message, _maxRetries, retryCount, timeToDeliver, QueueCount+1, _context); + return new MessageDelivery(_messageId, _subscriptionEndpointId, _contractType, _action, _message, _maxRetries, retryCount, timeToDeliver, QueueCount+1, _context); } } } IServiceOriented.ServiceBus/MessageDelivery.cs @@ -7,10 +7,18 @@ using System.Runtime.Serialization; namespace IServiceOriented.ServiceBus { + /// <summary> + /// Base class for message filters. + /// </summary> [Serializable] [DataContract] public abstract class MessageFilter { + /// <summary> + /// Determine whether a message should be included or skipped. + /// </summary> + /// <param name="request"></param> + /// <returns>true if the message should be included, false if the message should be skipped.</returns> public abstract bool Include(PublishRequest request); } IServiceOriented.ServiceBus/MessageFilter.cs @@ -130,7 +130,13 @@ namespace IServiceOriented.ServiceBus GC.SuppressFinalize(this); } + #endregion + + ~ReaderWriterLockedObject() + { + Dispose(false); + } } } IServiceOriented.ServiceBus/ReaderWriterLockedObject.cs @@ -351,7 +351,7 @@ namespace IServiceOriented.ServiceBus public event EventHandler<EndpointEventArgs> ListenerAdded; public event EventHandler<EndpointEventArgs> ListenerRemoved; - List<ListenerEndpoint> _listenerEndpoints = new List<ListenerEndpoint>(); + ListenerEndpointCollection _listenerEndpoints = new ListenerEndpointCollection(); object _listenerEndpointsLock = new object(); public IEnumerable<Endpoint> ListeningEndpoints @@ -362,7 +362,7 @@ namespace IServiceOriented.ServiceBus } } - ReaderWriterLockedObject<IEnumerable<SubscriptionEndpoint>, IList<SubscriptionEndpoint>> _subscriptions = new ReaderWriterLockedObject<IEnumerable<SubscriptionEndpoint>, IList<SubscriptionEndpoint>>(new List<SubscriptionEndpoint>(), l => l); + ReaderWriterLockedObject<IEnumerable<SubscriptionEndpoint>, SubscriptionEndpointCollection> _subscriptions = new ReaderWriterLockedObject<IEnumerable<SubscriptionEndpoint>, SubscriptionEndpointCollection>(new SubscriptionEndpointCollection(), l => l); public static void VerifyContract(Type contractType) { @@ -576,11 +576,11 @@ namespace IServiceOriented.ServiceBus return endpoint; } - protected void QueueDelivery(Guid subscriptionEndpointId, string action, object message, ReadOnlyDictionary<string, object> context) + protected void QueueDelivery(Guid subscriptionEndpointId, Type contractType, string action, object message, ReadOnlyDictionary<string, object> context) { if (message == null) throw new ArgumentNullException("message"); - MessageDelivery delivery = new MessageDelivery(subscriptionEndpointId, action, message, _maxRetries, context); + MessageDelivery delivery = new MessageDelivery(subscriptionEndpointId, contractType, action, message, _maxRetries, context); _messageDeliveryQueue.Enqueue(delivery); } @@ -819,7 +819,7 @@ namespace IServiceOriented.ServiceBus if (include) { - QueueDelivery(subscription.Id, publishRequest.Action, publishRequest.Message, publishRequest.Context); + QueueDelivery(subscription.Id, publishRequest.Contract, publishRequest.Action, publishRequest.Message, publishRequest.Context); handled = true; } } @@ -829,7 +829,7 @@ namespace IServiceOriented.ServiceBus { foreach (SubscriptionEndpoint subscription in unhandledFilters) { - QueueDelivery(subscription.Id, publishRequest.Action, publishRequest.Message, publishRequest.Context); + QueueDelivery(subscription.Id, publishRequest.Contract, publishRequest.Action, publishRequest.Message, publishRequest.Context); } } @@ -1064,6 +1064,11 @@ namespace IServiceOriented.ServiceBus GC.SuppressFinalize(this); } + + ~ServiceBusRuntime() + { + Dispose(false); + } } IServiceOriented.ServiceBus/ServiceBusRuntime.cs @@ -7,6 +7,9 @@ using System.Runtime.Serialization; namespace IServiceOriented.ServiceBus { + /// <summary> + /// Represents a subscription endpoint. + /// </summary> [Serializable] [DataContract] public sealed class SubscriptionEndpoint : Endpoint @@ -26,6 +29,9 @@ namespace IServiceOriented.ServiceBus } MessageFilter _filter; + /// <summary> + /// The filter to be applied to this subscription or null if all messages should be included. + /// </summary> [DataMember] public MessageFilter Filter { @@ -40,6 +46,9 @@ namespace IServiceOriented.ServiceBus } Dispatcher _dispatcher; + /// <summary> + /// The Dispatcher used to send messages to this endpoint. + /// </summary> [DataMember] public Dispatcher Dispatcher { IServiceOriented.ServiceBus/SubscriptionEndpoint.cs @@ -6,6 +6,9 @@ using System.Text; using System.Runtime.Serialization; namespace IServiceOriented.ServiceBus { + /// <summary> + /// Provides support for message transformation. The TransformationDispatcher transforms a PublishRequest and publishes the transformed request to the bus. + /// </summary> [Serializable] [DataContract] public abstract class TransformationDispatcher : Dispatcher @@ -15,7 +18,12 @@ namespace IServiceOriented.ServiceBus } - protected abstract PublishRequest Transform(PublishRequest information); + /// <summary> + /// Performs a transformation of a PublishRequest + /// </summary> + /// <param name="request">An incoming publish request.</param> + /// <returns>The transformed request or null if the message cannot be transformed.</returns> + protected abstract PublishRequest Transform(PublishRequest request); public const string TransformedByKeyName = "TransformedBy"; @@ -33,7 +41,7 @@ namespace IServiceOriented.ServiceBus } // Don't transform this message more than once - if (!oldTransformedByList.Contains(endpoint.Id)) + if (!oldTransformedByList.Contains(endpoint.Id) || AllowMultipleTransforms) { Guid[] newTransformedByList = new Guid[oldTransformedByList.Length + 1]; newTransformedByList[newTransformedByList.Length - 1] = endpoint.Id; @@ -42,12 +50,28 @@ namespace IServiceOriented.ServiceBus context = newContext.MakeReadOnly(); PublishRequest result = Transform(new PublishRequest(endpoint.ContractType, action, message, context)); - Runtime.Publish(new PublishRequest(result.Contract, result.Action, result.Message, context)); + if (result != null) + { + Runtime.Publish(new PublishRequest(result.Contract, result.Action, result.Message, context)); + } } else { System.Diagnostics.Trace.TraceInformation("Skipping already transformed message (" + DispatchContext.MessageDelivery.MessageId +")"); } } + + /// <summary> + /// Gets or sets whether this endpoint can transform a message multiple times. + /// </summary> + /// <remarks> + /// Setting this property to true can cause cycles if it transforms both to and from a certain message type. + /// </remarks> + [DataMember] + public bool AllowMultipleTransforms + { + get; + set; + } } } IServiceOriented.ServiceBus/TransformationDispatcher.cs @@ -9,6 +9,9 @@ using System.Runtime.Serialization; namespace IServiceOriented.ServiceBus { + /// <summary> + /// Provides support for dispatching messages to WCF endpoints. + /// </summary> [Serializable] [DataContract] public class WcfDispatcher : Dispatcher IServiceOriented.ServiceBus/WcfDispatcher.cs @@ -9,6 +9,12 @@ using System.Runtime.Serialization; namespace IServiceOriented.ServiceBus { + /// <summary> + /// Provides support for hosting WCF service contracts with the service bus. + /// </summary> + /// <remarks> + /// Currently, contracts are restricted to one way, single parameter, methods. + /// </remarks> [Serializable] [DataContract] public class WcfListener : Listener IServiceOriented.ServiceBus/WcfListener.cs 46870d58158e9bda7494e71b5e61ce751cc9e19e JEzell jezell@gmail.com http://github.com/jezell/iserviceoriented/commit/cffe8a8a6f1b41efaf0f8237913db8f0cc984e4c cffe8a8a6f1b41efaf0f8237913db8f0cc984e4c 2008-08-31T18:10:54-07:00 2008-08-31T18:10:54-07:00 Adding some xml comments 66139a42d72402ff5022f78e86ad80301265cf0f JEzell jezell@gmail.com