@@ -11,7 +11,7 @@ namespace IServiceOriented.ServiceBus.UnitTests [ServiceContract] public interface IContract { - [OperationContract(IsOneWay=true)] + [OperationContract(IsOneWay=true, Action="PublishThis")] void PublishThis(string message); } @@ -25,6 +25,7 @@ namespace IServiceOriented.ServiceBus.UnitTests volatile int failCount = 0; int failInterval = 0; + public void PublishThis(string message) { int value = Interlocked.Increment(ref callCount); IServiceOriented.ServiceBus.UnitTests/IContract.cs @@ -6,6 +6,7 @@ using NUnit.Framework; using IServiceOriented.ServiceBus.Delivery; using IServiceOriented.ServiceBus.Dispatchers; using IServiceOriented.ServiceBus.Delivery.Formatters; +using System.ServiceModel.Description; namespace IServiceOriented.ServiceBus.UnitTests { @@ -16,6 +17,12 @@ namespace IServiceOriented.ServiceBus.UnitTests { } + public void TEst() + { + ContractDescription description = ContractDescription.GetContract(typeof(ISendMessageContract)); + Console.WriteLine(description); + } + [Test] public void QueuedDeliveryCore_Publishes_Response_Messages_For_TwoWay_Operation() IServiceOriented.ServiceBus.UnitTests/TestRequestResponse.cs @@ -41,7 +41,7 @@ namespace IServiceOriented.ServiceBus.UnitTests string message = "blah blah test test"; - contractDispatcher.Dispatch(new MessageDelivery(endpoint.Id, typeof(IContract), "http://tempuri.org/PublishThis", message, 3, new MessageDeliveryContext())); + contractDispatcher.Dispatch(new MessageDelivery(endpoint.Id, typeof(IContract), "PublishThis", message, 3, new MessageDeliveryContext())); Assert.AreEqual(1, ci.PublishedCount); Assert.AreEqual(message, ci.PublishedMessages[0]); @@ -102,8 +102,7 @@ namespace IServiceOriented.ServiceBus.UnitTests runtime.Start(); - - string action = "http://tempuri.org/IContract/PublishThis"; + string action = "PublishThis"; string body = "blah blah test test"; XmlDocument document = new XmlDocument(); IServiceOriented.ServiceBus.UnitTests/TestWcfDispatcher.cs @@ -19,45 +19,71 @@ namespace IServiceOriented.ServiceBus.Delivery.Formatters { internal class DataContractMessageDeliveryConverter : MessageDeliveryConverter { - public DataContractMessageDeliveryConverter(Type interfaceType) + public DataContractMessageDeliveryConverter(Type contractType) + : this(ContractDescription.GetContract(contractType)) { - foreach (WcfMessageInformation information in WcfUtils.GetMessageInformation(interfaceType)) + + } + + public DataContractMessageDeliveryConverter(ContractDescription contract) + { + foreach (OperationDescription operation in contract.Operations) { - cacheSerializer(interfaceType, information.MessageType); + foreach (MessageDescription message in operation.Messages) + { + cacheSerializer(operation, message); + } } } - Dictionary<Type, DataContractSerializer> _serializers = new Dictionary<Type, DataContractSerializer>(); + Dictionary<string, DataContractSerializer> _serializers = new Dictionary<string, DataContractSerializer>(); - void cacheSerializer(Type interfaceType, Type messageType) + void cacheSerializer(OperationDescription operation, MessageDescription message) { - if (!_serializers.ContainsKey(messageType)) + if (!_serializers.ContainsKey(message.Action)) { - _serializers.Add(messageType, new DataContractSerializer(messageType, WcfUtils.GetServiceKnownTypes(interfaceType))); + _serializers.Add(message.Action, message.Body.Parts.Count == 0 ? null : new DataContractSerializer(message.Body.Parts[0].Type, operation.KnownTypes)); } } - DataContractSerializer getSerializer(Type interfaceType) + DataContractSerializer getSerializer(string action) { try { - return _serializers[interfaceType]; + return _serializers[action]; } catch (KeyNotFoundException) { - throw new InvalidOperationException("Unsupported interface or action"); + throw new InvalidOperationException("Unsupported action"); } } protected override Message ToMessageCore(MessageDelivery delivery) { + if (!_serializers.ContainsKey(delivery.Action)) + { + throw new InvalidOperationException("Unsupported action"); + } + return System.ServiceModel.Channels.Message.CreateMessage(MessageVersion.Default, delivery.Action, delivery.Message); } protected override object GetMessageObject(Message message) { - var serializer = getSerializer(Type.GetType(message.Headers.GetHeader<string>(MessageTypeHeader, MessagingNamespace))); - return serializer.ReadObject(message.GetReaderAtBodyContents()); + var serializer = getSerializer(message.Headers.Action); + if (serializer != null) + { + return serializer.ReadObject(message.GetReaderAtBodyContents()); + } + else + { + return null; + } + } + + public override IEnumerable<string> SupportedActions + { + get { return _serializers.Keys; } } } } IServiceOriented.ServiceBus/Delivery/Formatters/DataContractMessageDeliveryConverter.cs @@ -19,49 +19,78 @@ namespace IServiceOriented.ServiceBus.Delivery.Formatters { internal class MessageContractMessageDeliveryConverter : MessageDeliveryConverter { - public MessageContractMessageDeliveryConverter(Type contractType) + public MessageContractMessageDeliveryConverter(Type contractType) : this(ContractDescription.GetContract(contractType)) { - foreach (WcfMessageInformation information in WcfUtils.GetMessageInformation(contractType)) - { - cacheConverter(information.MessageType, information.Action); + + } + + public MessageContractMessageDeliveryConverter(ContractDescription contract) + { + foreach (OperationDescription operation in contract.Operations) + { + foreach (MessageDescription message in operation.Messages) + { + if (message.MessageType == null && message.Body.Parts.Count > 0) + { + throw new InvalidOperationException("Unsupported message contract format"); + } + cacheConverter(message.Action, message.MessageType); + } } } Dictionary<string, TypedMessageConverter> _converterHash = new Dictionary<string, TypedMessageConverter>(); - void cacheConverter(Type objType, string action) + void cacheConverter(string action, Type messageType) { - string key = objType + ":" + action; - if (!_converterHash.ContainsKey(key)) - { - _converterHash.Add(key, (TypedMessageConverter)TypedMessageConverter.Create(objType, action)); + if (!_converterHash.ContainsKey(action)) + { + _converterHash.Add(action, messageType == null ? null : (TypedMessageConverter)TypedMessageConverter.Create(messageType, action)); } } - TypedMessageConverter getCachedConverter(Type objType, string action) + TypedMessageConverter getCachedConverter(string action) { - string key = objType + ":" + action; try { - return (TypedMessageConverter)_converterHash[key]; + return (TypedMessageConverter)_converterHash[action]; } catch (KeyNotFoundException) { - throw new InvalidOperationException("Unsupported interface or action"); + throw new InvalidOperationException("Unsupported action"); } } protected override object GetMessageObject(Message message) { - TypedMessageConverter converter = getCachedConverter(Type.GetType(message.Headers.GetHeader<string>(MessageTypeHeader, MessagingNamespace)), message.Headers.Action); + TypedMessageConverter converter = getCachedConverter(message.Headers.Action); return converter.FromMessage(message); } protected override Message ToMessageCore(MessageDelivery delivery) { - Type objType = delivery.Message.GetType(); - TypedMessageConverter converter = getCachedConverter(objType, delivery.Action); - return converter.ToMessage(delivery.Message); + if (!_converterHash.ContainsKey(delivery.Action)) + { + throw new InvalidOperationException("Unsupported action"); + } + + TypedMessageConverter converter = getCachedConverter(delivery.Action); + if (converter != null) + { + return converter.ToMessage(delivery.Message); + } + else + { + return null; + } + } + + public override IEnumerable<string> SupportedActions + { + get + { + return _converterHash.Keys; + } } } IServiceOriented.ServiceBus/Delivery/Formatters/MessageContractMessageDeliveryConverter.cs @@ -20,7 +20,6 @@ namespace IServiceOriented.ServiceBus.Delivery.Formatters { public abstract class MessageDeliveryConverter { - public const string MessageTypeHeader = "messageType"; public const string MaxRetriesHeader = "maxRetries"; public const string RetryCountHeader = "retryCount"; public const string MessageIdHeader = "messageId"; @@ -42,7 +41,6 @@ namespace IServiceOriented.ServiceBus.Delivery.Formatters Message msg = ToMessageCore(delivery); msg.Headers.Add(System.ServiceModel.Channels.MessageHeader.CreateHeader(ContextHeader, MessagingNamespace, delivery.Context)); - msg.Headers.Add(System.ServiceModel.Channels.MessageHeader.CreateHeader(MessageTypeHeader, MessagingNamespace, objType.AssemblyQualifiedName)); msg.Headers.Add(System.ServiceModel.Channels.MessageHeader.CreateHeader(MaxRetriesHeader, MessagingNamespace, delivery.MaxRetries)); msg.Headers.Add(System.ServiceModel.Channels.MessageHeader.CreateHeader(RetryCountHeader, MessagingNamespace, delivery.RetryCount)); msg.Headers.Add(System.ServiceModel.Channels.MessageHeader.CreateHeader(MessageIdHeader, MessagingNamespace, delivery.MessageDeliveryId)); @@ -74,13 +72,24 @@ namespace IServiceOriented.ServiceBus.Delivery.Formatters return delivery; } + public abstract IEnumerable<string> SupportedActions + { + get; + } + + public static MessageDeliveryConverter CreateConverter(Type interfaceType) { - if (WcfUtils.UsesMessages(interfaceType)) + ContractDescription description = ContractDescription.GetContract(interfaceType); + + int rawMessageOperationCount = description.Operations.Where( o => o.Messages.First().Body.Parts.First().Type == typeof(Message)).Count(); + int messageContractOperationCount = description.Operations.Where(o => o.Messages.First().MessageType != null).Count(); + + if (rawMessageOperationCount > 0) { return new RawMessageMessageDeliveryConverter(interfaceType); } - else if (WcfUtils.UsesMessageContracts(interfaceType)) + else if (messageContractOperationCount > 0) { return new MessageContractMessageDeliveryConverter(interfaceType); } @@ -89,6 +98,18 @@ namespace IServiceOriented.ServiceBus.Delivery.Formatters return new DataContractMessageDeliveryConverter(interfaceType); } } + + + public static MessageDeliveryConverter CreateConverter(params Type[] contractTypes) + { + MessageDeliveryConverter[] converters = new MessageDeliveryConverter[contractTypes.Length]; + + for(int i = 0; i < contractTypes.Length; i ++) + { + converters[i] = CreateConverter(contractTypes[i]); + } + return new CompositeMessageDeliveryConverter(converters); + } } IServiceOriented.ServiceBus/Delivery/Formatters/MessageDeliveryConverter.cs @@ -215,7 +215,7 @@ namespace IServiceOriented.ServiceBus.Delivery } const int FUTURE_SLEEP_MS = 100; const int RETRY_SLEEP_MS = 100; - const int DEQUEUE_TIMEOUT_SECONDS = 5; + const int DEQUEUE_TIMEOUT_SECONDS = 1; List<Thread> _workerThreads = new List<Thread>(); object _workerThreadsLock = new object(); IServiceOriented.ServiceBus/Delivery/QueuedDeliveryCore.cs @@ -10,6 +10,7 @@ using IServiceOriented.ServiceBus.Threading; using IServiceOriented.ServiceBus.Collections; using IServiceOriented.ServiceBus.Listeners; using System.ServiceModel.Channels; +using System.ServiceModel.Description; namespace IServiceOriented.ServiceBus.Dispatchers { @@ -42,24 +43,28 @@ namespace IServiceOriented.ServiceBus.Dispatchers Dictionary<string, MethodInfo> actionLookup = new Dictionary<string, MethodInfo>(); Dictionary<string, string> replyActionLookup = new Dictionary<string, string>(); - foreach (MethodInfo method in endpoint.ContractType.GetMethods()) + foreach (OperationDescription operation in ContractDescription.GetContract(endpoint.ContractType).Operations) { - if (WcfUtils.IsServiceMethod(method)) + MessageDescription requestMessage = operation.Messages.Where(md => md.Direction == MessageDirection.Input).First(); + MessageDescription responseMessage = operation.Messages.Where(md => md.Direction == MessageDirection.Output).FirstOrDefault(); + + if (requestMessage.Action == "*") { - string action = WcfUtils.GetAction(endpoint.ContractType, method); - if (action == "*") + _passThrough = true; + } + + actionLookup.Add(requestMessage.Action, operation.SyncMethod); + + if (responseMessage != null) + { + string replyAction = responseMessage.Action; + + if (replyAction != null) { - _passThrough = true; + replyActionLookup.Add(requestMessage.Action, replyAction); } - actionLookup.Add(action, method); - - string replyAction = WcfUtils.GetReplyAction(endpoint.ContractType, method); - - if(replyAction != null) - { - replyActionLookup.Add(action, replyAction); - } } + } lookup.MethodLookup = actionLookup; lookup.ReplyActionLookup = replyActionLookup; @@ -176,9 +181,18 @@ namespace IServiceOriented.ServiceBus.Dispatchers [NonSerialized] ActionLookup _lookup; - public static MessageFilter CreateMessageFilter(Type interfaceType) - { - return new TypedMessageFilter(WcfUtils.GetMessageInformation(interfaceType).Select(mi => mi.MessageType).ToArray()); + public static MessageFilter CreateMessageFilter(Type contractType) + { + ContractDescription description = ContractDescription.GetContract(contractType); + List<Type> messageTypes = new List<Type>(); + foreach (OperationDescription operation in description.Operations) + { + foreach (MessageDescription message in operation.Messages) + { + messageTypes.Add(message.Body.Parts[0].Type); + } + } + return new TypedMessageFilter(messageTypes.ToArray()); } } IServiceOriented.ServiceBus/Dispatchers/WcfProxyDispatcher.cs @@ -93,6 +93,7 @@ </ItemGroup> <ItemGroup> <Compile Include="DeliveryException.cs" /> + <Compile Include="Delivery\Formatters\CompositeMessageDeliveryConverter.cs" /> <Compile Include="Delivery\Formatters\ConverterMessageDeliveryReader.cs" /> <Compile Include="Delivery\Formatters\ConverterMessageDeliveryReaderFactory.cs" /> <Compile Include="Delivery\Formatters\ConverterMessageDeliveryWriter.cs" /> @@ -165,7 +166,6 @@ <Compile Include="Listeners\WcfListener.cs" /> <Compile Include="Listeners\WcfServiceHostFactory.cs" /> <Compile Include="Services\WcfManagementService.cs" /> - <Compile Include="WcfUtils.cs" /> </ItemGroup> <ItemGroup> <None Include="ClassDiagram.cd" /> IServiceOriented.ServiceBus/IServiceOriented.ServiceBus.csproj @@ -27,8 +27,11 @@ namespace IServiceOriented.ServiceBus.Listeners protected override void OnStop() { - CommunicationObject.Close(); - CommunicationObject = null; + if (CommunicationObject != null) + { + CommunicationObject.Close(); + CommunicationObject = null; + } base.OnStop(); } IServiceOriented.ServiceBus/Listeners/WcfListener.cs @@ -12,6 +12,7 @@ using System.Reflection.Emit; using System.Security.Principal; using IServiceOriented.ServiceBus.Collections; +using System.ServiceModel.Description; namespace IServiceOriented.ServiceBus.Listeners { @@ -68,19 +69,16 @@ namespace IServiceOriented.ServiceBus.Listeners CustomAttributeBuilder attributeBuilder = new CustomAttributeBuilder(typeof(ServiceBehaviorAttribute).GetConstructor(new Type[] { }), new object[] { }, new PropertyInfo[] { typeof(ServiceBehaviorAttribute).GetProperty("InstanceContextMode"), typeof(ServiceBehaviorAttribute).GetProperty("ConcurrencyMode"), typeof(ServiceBehaviorAttribute).GetProperty("Namespace") }, new object[] { InstanceContextMode.Single, ConcurrencyMode.Multiple, "http://iserviceoriented.com/serviceBus/2008/" }); typeBuilder.SetCustomAttribute(attributeBuilder); - foreach (MethodInfo methodInfo in interfaceType.GetMethods()) + foreach (OperationDescription od in ContractDescription.GetContract(interfaceType).Operations) { - // Only add methods with OperationContractAttribute - if(WcfUtils.IsServiceMethod(methodInfo)) + MethodInfo method = od.SyncMethod; + MessageDescription message = od.Messages.Where(m => m.Direction == MessageDirection.Input).First(); + if (message.Body.Parts.Count > 1) { - string action = WcfUtils.GetAction(interfaceType, methodInfo); - if (methodInfo.GetParameters().Length != 1) - { - continue; // skip methods that don't accept a single parameter - } - - definePublishOverride(interfaceType, typeBuilder, methodInfo, action); + throw new InvalidOperationException("Only single parameter methods are currently supported"); } + + definePublishOverride(interfaceType, typeBuilder, method, message.Action); } Type impType = typeBuilder.CreateType(); return impType; IServiceOriented.ServiceBus/Listeners/WcfServiceHostFactory.cs IServiceOriented.ServiceBus/WcfUtils.cs bbc32282b9f135da2fa1c26eb10df08a933ba1ad jezell jezell@gmail.com http://github.com/jezell/iserviceoriented/commit/a13d552fd34aff3ff8c7ed61e0be8031a72922eb a13d552fd34aff3ff8c7ed61e0be8031a72922eb 2008-10-16T07:51:45-07:00 2008-10-16T07:51:45-07:00 Replace WcfUtils with calls to WCF metadata classes, support for ContractDescription in serialization, support for multiple contract serialization a26d7a9277541d45912009231760eb42ec0d5bb4 jezell jezell@gmail.com