//******************************************************************************************************//  ServiceBusService.cs - Gbtc////  Copyright © 2012, Grid Protection Alliance.  All Rights Reserved.////  Licensed to the Grid Protection Alliance (GPA) under one or more contributor license agreements. See//  the NOTICE file distributed with this work for additional information regarding copyright ownership.//  The GPA licenses this file to you under the MIT License (MIT), the "License"; you may//  not use this file except in compliance with the License. You may obtain a copy of the License at:////      http://www.opensource.org/licenses/MIT////  Unless agreed to in writing, the subject software distributed under the License is distributed on an//  "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Refer to the//  License for the specific language governing permissions and limitations.////  Code Modification History://  ----------------------------------------------------------------------------------------------------//  10/06/2010 - Pinal C. Patel//       Generated original version of source code.//  10/26/2010 - Pinal C. Patel//       Added management operations GetClients(), GetQueues() and GetTopics().//  11/23/2010 - Pinal C. Patel//       Added new BufferThreshold and ProcessingMode properties.//       Enhanced thread synchronization using ReaderWriterLockSlim for better performance.//  11/24/2010 - Pinal C. Patel//       Updated the text returned by Status property.//  01/07/2011 - Pinal C. Patel//       Fixed initialization issue that prevented the service from functioning when hosted inside ASP.NET.//  02/03/2011 - Pinal C. Patel//       Added GetLatestMessage() operation that can be used to retrieve the latest message published //       to the subscribers of a topic.//  12/20/2012 - Starlynn Danyelle Gilliam//       Modified Header.//  01/2/2014 - Pinal C. Patel//       Updated to support publishing locally from inheriting class without establishing a communications //       channel over the web service interface.//       Added bubbling up of exception encountered when publishing messages.//       Updated Status to include information about clients, queues and topics.////****************************************************************************************************** using System;using System.Collections.Generic;using System.ServiceModel;using System.ServiceModel.Description;using System.Text;using System.Threading;using GSF.Collections;using GSF.Configuration;using GSF.ServiceModel; namespace GSF.ServiceBus{    #region [ Enumerations ]     /// <summary>    /// Indicates how the distribution of <see cref="Message"/>s is processed by the <see cref="ServiceBusService"/>.    /// </summary>    public enum MessageProcessingMode    {        /// <summary>        /// <see cref="Message"/> distribution is processed in parallel for increased distribution performance.        /// </summary>        Parallel,        /// <summary>        /// <see cref="Message"/> distribution is processed sequentially to preserve <see cref="Message"/> ordering.        /// </summary>        Sequential    }     #endregion     /// <summary>    /// A service bus for event-based messaging between disjoint systems.    /// </summary>    /// <example>    /// This example shows how to host <see cref="ServiceBusService"/> inside a console application:    /// <code>    /// using System;    /// using System.ServiceModel;    /// using System.ServiceModel.Description;    /// using System.Threading;    /// using GSF;    /// using GSF.ServiceBus;    ///     /// class Program    /// {    ///     static void Main(string[] args)    ///     {    ///         // Prompt for security option.    ///         Console.Write("Enable security (Y/N): ");    ///         bool enableSecurity = Console.ReadLine().ParseBoolean();    ///     ///         // Initialize service bus.    ///         ServiceBusService service = new ServiceBusService();    ///         service.Singleton = true;    ///         service.PublishMetadata = true;    ///         service.PersistSettings = false;    ///         service.Endpoints = "http.duplex://localhost:4501; net.tcp://locahost:4502";    ///         if (enableSecurity)    ///             service.SecurityPolicy = typeof(ServiceBusSecurityPolicy).FullName;    ///         service.Initialize();    ///     ///         // Show service bus status.    ///         if (service.ServiceHost.State == CommunicationState.Opened)    ///         {    ///             Console.WriteLine("\r\n{0} is running:", service.GetType().Name);    ///             foreach (ServiceEndpoint endpoint in service.ServiceHost.Description.Endpoints)    ///             {    ///                 Console.WriteLine("- {0} ({1})", endpoint.Address, endpoint.Binding.GetType().Name);    ///             }    ///     ///             new Thread(delegate()     ///                 {    ///                     while (service.ServiceHost.State == CommunicationState.Opened)    ///                     {    ///                         Console.WriteLine();    ///                         Console.WriteLine();    ///                         Console.WriteLine("Status of {0}:", service.Name);    ///                         Console.WriteLine(service.Status);    ///                         Console.Write("Press Enter key to stop...");    ///     ///                         Thread.Sleep(5000);    ///                     }    ///                 }).Start();    ///         }    ///         else    ///         {    ///             Console.WriteLine("\r\n{0} could not be started", service.GetType().Name);    ///         }    ///     ///         // Shutdown.    ///         Console.ReadLine();    ///         service.Dispose();    ///     }    /// }    /// </code>    /// This example shows how to host <see cref="ServiceBusService"/> inside a web application:    /// <code>    /// <![CDATA[    /// <?xml version="1.0"?>    /// <configuration>    ///   <configSections>    ///     <section name="categorizedSettings" type="GSF.Configuration.CategorizedSettingsSection, GSF.Core" />    ///   </configSections>    ///   <categorizedSettings>    ///     <serviceBusService>    ///       <add name="Endpoints" value="" description="Semicolon delimited list of URIs where the web service can be accessed."    ///         encrypted="false" />    ///       <add name="Contract" value="GSF.ServiceBus.IServiceBusService, GSF.ServiceBus"    ///         description="Assembly qualified name of the contract interface implemented by the web service."    ///         encrypted="false" />    ///       <add name="Singleton" value="True" description="True if the web service is singleton; otherwise False."    ///         encrypted="false" />    ///       <add name="SecurityPolicy" value="" description="Assembly qualified name of the authorization policy to be used for securing the web service."    ///         encrypted="false" />    ///       <add name="PublishMetadata" value="True" description="True if the web service metadata is to be published at all the endpoints; otherwise False."    ///         encrypted="false" />    ///       <add name="BufferThreshold" value="-1" description="Maximum number of messages that can be queued for distribution before the oldest ones are discarded."    ///         encrypted="false" />    ///       <add name="ProcessingMode" value="Sequential" description="Processing mode (Parallel; Sequential) to be used for the distribution of messages."    ///         encrypted="false" />    ///     </serviceBusService>    ///   </categorizedSettings>    ///   <system.serviceModel>    ///     <services>    ///       <service name="GSF.ServiceBus.ServiceBusService">    ///         <endpoint address="" contract="GSF.ServiceBus.IServiceBusService" binding="wsDualHttpBinding" />    ///       </service>    ///     </services>    ///     <behaviors>    ///       <serviceBehaviors>    ///         <behavior>    ///           <serviceMetadata httpGetEnabled="true" />    ///           <serviceDebug includeExceptionDetailInFaults="false" />    ///         </behavior>    ///       </serviceBehaviors>    ///     </behaviors>    ///     <serviceHostingEnvironment multipleSiteBindingsEnabled="true">    ///       <serviceActivations>    ///         <add relativeAddress="ServiceBusService.svc" service="GSF.ServiceBus.ServiceBusService, GSF.ServiceBus" />    ///       </serviceActivations>    ///     </serviceHostingEnvironment>    ///   </system.serviceModel>    /// </configuration>    /// ]]>    /// </code>    /// This example shows how to publish <see cref="Message"/>s to <see cref="ServiceBusService"/>:    /// <code>    /// using System;    /// using System.ServiceModel;    /// using System.Threading;    ///     /// class Program : IServiceBusServiceCallback    /// {    ///     static void Main(string[] args)    ///     {    ///         // NOTE: Service reference to the service bus service must be added to generate the service proxy.    ///     ///         // Initialize auto-generated service bus service proxy.    ///         InstanceContext callbackContext = new InstanceContext(new Program());    ///         ServiceBusServiceClient serviceBusService = new ServiceBusServiceClient(callbackContext, "NetTcpBinding_IServiceBusService");    ///     ///         // Create registration request for publishing messages.    ///         RegistrationRequest registration = new RegistrationRequest();    ///         registration.MessageType = MessageType.Topic;    ///         registration.MessageName = "Topic.Frequency";    ///         registration.RegistrationType = RegistrationType.Produce;    ///         serviceBusService.Register(registration);    ///     ///         // Start publishing messages to the bus asynchronously.    ///         new Thread(delegate()     ///             {    ///                 Message message = new Message();    ///                 message.Type = registration.MessageType;    ///                 message.Name = registration.MessageName;    ///                 message.Format = "application/octet-stream";    ///     ///                 Random random = new Random(59);    ///                 while (serviceBusService.State == CommunicationState.Opened)    ///                 {    ///                     message.Time = DateTime.UtcNow;    ///                     message.Content = BitConverter.GetBytes(random.Next(61));    ///                     serviceBusService.Publish(message);    ///     ///                     Thread.Sleep(5000);    ///                 }    ///             }).Start();    ///     ///         // Shutdown.    ///         Console.Write("Press Enter key to stop...");    ///         Console.ReadLine();    ///         serviceBusService.Close();    ///     }    ///     ///     public void ProcessMessage(Message message)    ///     {    ///         // This method will not be invoked since we are not consuming messages.    ///         throw new NotSupportedException();    ///     }    /// }    /// </code>    /// This example shows how to subscribe to <see cref="ServiceBusService"/> for receiving <see cref="Message"/>s:    /// <code>    /// using System;    /// using System.ServiceModel;    ///     /// class Program : IServiceBusServiceCallback    /// {    ///     static void Main(string[] args)    ///     {    ///         // NOTE: Service reference to the service bus service must be added to generate the service proxy.    ///     ///         // Initialize auto-generated service bus service proxy.    ///         InstanceContext callbackContext = new InstanceContext(new Program());    ///         ServiceBusServiceClient serviceBusService = new ServiceBusServiceClient(callbackContext, "NetTcpBinding_IServiceBusService");    ///     ///         // Subscribe with service bus service to receive messages.    ///         RegistrationRequest registration = new RegistrationRequest();    ///         registration.MessageType = MessageType.Topic;    ///         registration.MessageName = "Topic.Frequency";    ///         registration.RegistrationType = RegistrationType.Consume;    ///         serviceBusService.Register(registration);    ///     ///         // Shutdown.    ///         Console.WriteLine("Press Enter key to stop...");    ///         Console.WriteLine();    ///         Console.ReadLine();    ///         serviceBusService.Close();    ///     }    ///     ///     public void ProcessMessage(Message message)    ///     {    ///         if (message.Format != "application/octet-stream")    ///             Console.WriteLine("Message format '{0}' is not supported", message.Format);    ///         else    ///             Console.WriteLine("Message received: {0} Hz", BitConverter.ToInt32(message.Content, 0));    ///     }    /// }    /// </code>    /// This example shows how to monitor <see cref="ServiceBusService"/> remotely:    /// <code>    /// using System;    /// using System.ServiceModel;    /// using System.Threading;    /// using GSF;    ///     /// class Program : IServiceBusServiceCallback    /// {    ///     static void Main(string[] args)    ///     {    ///         // NOTE: Service reference to the service bus service must be added to generate the service proxy.    ///     ///         // Initialize auto-generated service bus service proxy.    ///         InstanceContext callbackContext = new InstanceContext(new Program());    ///         ServiceBusServiceClient serviceBusService = new ServiceBusServiceClient(callbackContext, "NetTcpBinding_IServiceBusService");    ///         serviceBusService.ChannelFactory.Open();    ///     ///         // Start querying service bus service status asynchronously.    ///         new Thread(delegate()    ///         {    ///             while (serviceBusService.State == CommunicationState.Opened)    ///             {    ///                 Console.Clear();    ///                 Console.WriteLine(new string('-', 79));    ///                 Console.WriteLine("|" + "Service Bus Status".CenterText(77) + "|");    ///                 Console.WriteLine(new string('-', 79));    ///                 Console.WriteLine();    ///     ///                 // Show clients.    ///                 Console.Write("Client ID".PadRight(25));    ///                 Console.Write(" ");    ///                 Console.Write("Connected".PadRight(21));    ///                 Console.Write(" ");    ///                 Console.Write("Msg. Produced".PadRight(15));    ///                 Console.Write(" ");    ///                 Console.Write("Msg. Consumed".PadRight(15));    ///                 Console.WriteLine();    ///                 Console.Write(new string('-', 25));    ///                 Console.Write(" ");    ///                 Console.Write(new string('-', 21));    ///                 Console.Write(" ");    ///                 Console.Write(new string('-', 15));    ///                 Console.Write(" ");    ///                 Console.Write(new string('-', 15));    ///                 Console.WriteLine();    ///                 foreach (ClientInfo client in serviceBusService.GetClients())    ///                 {    ///                     Console.Write(client.SessionId.TruncateRight(25).PadRight(25));    ///                     Console.Write(" ");    ///                     Console.Write(client.ConnectedAt.ToString("MM/dd/yy hh:mm:ss tt").PadRight(21));    ///                     Console.Write(" ");    ///                     Console.Write(client.MessagesProduced.ToString().PadRight(15));    ///                     Console.Write(" ");    ///                     Console.Write(client.MessagesConsumed.ToString().PadRight(15));    ///                     Console.WriteLine();    ///                 }    ///                 Console.WriteLine();    ///     ///                 // Show queues.    ///                 Console.Write("Queue Name".PadRight(25));    ///                 Console.Write(" ");    ///                 Console.Write("Producers".PadRight(10));    ///                 Console.Write(" ");    ///                 Console.Write("Consumers".PadRight(10));    ///                 Console.Write(" ");    ///                 Console.Write("Msg. Received".PadRight(15));    ///                 Console.Write(" ");    ///                 Console.Write("Msg. Processed".PadRight(15));    ///                 Console.WriteLine();    ///                 Console.Write(new string('-', 25));    ///                 Console.Write(" ");    ///                 Console.Write(new string('-', 10));    ///                 Console.Write(" ");    ///                 Console.Write(new string('-', 10));    ///                 Console.Write(" ");    ///                 Console.Write(new string('-', 15));    ///                 Console.Write(" ");    ///                 Console.Write(new string('-', 15));    ///                 Console.WriteLine();    ///                 foreach (RegistrationInfo queue in serviceBusService.GetQueues())    ///                 {    ///                     Console.Write(queue.MessageName.PadRight(25));    ///                     Console.Write(" ");    ///                     Console.Write(queue.Producers.Length.ToString().PadRight(15));    ///                     Console.Write(" ");    ///                     Console.Write(queue.Consumers.Length.ToString().PadRight(15));    ///                     Console.Write(" ");    ///                     Console.Write(queue.MessagesReceived.ToString().PadRight(15));    ///                     Console.Write(" ");    ///                     Console.Write(queue.MessagesProcessed.ToString().PadRight(15));    ///                     Console.WriteLine();    ///                 }    ///                 Console.WriteLine();    ///     ///                 // Show topics.    ///                 Console.Write("Topic Name".PadRight(25));    ///                 Console.Write(" ");    ///                 Console.Write("Producers".PadRight(10));    ///                 Console.Write(" ");    ///                 Console.Write("Consumers".PadRight(10));    ///                 Console.Write(" ");    ///                 Console.Write("Msg. Received".PadRight(15));    ///                 Console.Write(" ");    ///                 Console.Write("Msg. Processed".PadRight(15));    ///                 Console.WriteLine();    ///                 Console.Write(new string('-', 25));    ///                 Console.Write(" ");    ///                 Console.Write(new string('-', 10));    ///                 Console.Write(" ");    ///                 Console.Write(new string('-', 10));    ///                 Console.Write(" ");    ///                 Console.Write(new string('-', 15));    ///                 Console.Write(" ");    ///                 Console.Write(new string('-', 15));    ///                 Console.WriteLine();    ///                 foreach (RegistrationInfo topic in serviceBusService.GetTopics())    ///                 {    ///                     Console.Write(topic.MessageName.PadRight(25));    ///                     Console.Write(" ");    ///                     Console.Write(topic.Producers.Length.ToString().PadRight(15));    ///                     Console.Write(" ");    ///                     Console.Write(topic.Consumers.Length.ToString().PadRight(15));    ///                     Console.Write(" ");    ///                     Console.Write(topic.MessagesReceived.ToString().PadRight(15));    ///                     Console.Write(" ");    ///                     Console.Write(topic.MessagesProcessed.ToString().PadRight(15));    ///                     Console.WriteLine();    ///                 }    ///                 Console.WriteLine();    ///                 Console.Write("Press Enter key to stop...");    ///     ///                 Thread.Sleep(5000);    ///             }    ///         }).Start();    ///     ///         // Shutdown.    ///         Console.ReadLine();    ///         serviceBusService.Close();    ///     }    ///     ///     public void ProcessMessage(Message message)    ///     {    ///         // This method will not be invoked since we are not consuming messages.    ///         throw new NotSupportedException();    ///     }    /// }    /// </code>    /// </example>    /// <seealso cref="Message"/>    /// <seealso cref="ClientInfo"/>    /// <seealso cref="RegistrationInfo"/>    /// <seealso cref="RegistrationRequest"/>    /// <seealso cref="ServiceBusSecurityPolicy"/>    [ServiceBehavior(InstanceContextMode = InstanceContextMode.Single, UseSynchronizationContext = false)]    public class ServiceBusService : SelfHostingService, IServiceBusService    {        #region [ Members ]         // Nested Types         private class PublishContext        {            public PublishContext(Message message, RegistrationInfo registration)            {                Message = message;                Registration = registration;            }             public readonly Message Message;             public readonly RegistrationInfo Registration;        }         // Constants         /// <summary>        /// Specifies the default value for the <see cref="BufferThreshold"/> property.        /// </summary>        public const int DefaultBufferThreshold = -1;         /// <summary>        /// Specifies the default value for the <see cref="ProcessingMode"/> property.        /// </summary>        public const MessageProcessingMode DefaultProcessingMode = MessageProcessingMode.Sequential;         // Fields        private int m_bufferThreshold;        private MessageProcessingMode m_processingMode;        private readonly Dictionary<string, ClientInfo> m_clients;        private readonly Dictionary<string, RegistrationInfo> m_queues;        private readonly Dictionary<string, RegistrationInfo> m_topics;        private readonly ReaderWriterLockSlim m_clientsLock;        private readonly ReaderWriterLockSlim m_queuesLock;        private readonly ReaderWriterLockSlim m_topicsLock;        private ProcessQueue<PublishContext> m_publishQueue;        private long m_discardedMessages;        private bool m_disposed;         #endregion         #region [ Constructors ]         /// <summary>        /// Initializes a new instance of the <see cref="ServiceBusService"/> class.        /// </summary>        public ServiceBusService()        {            // Override base class settings.            Singleton = true;            PublishMetadata = true;            PersistSettings = true;             // Initialize member variables.            m_bufferThreshold = DefaultBufferThreshold;            m_processingMode = DefaultProcessingMode;            m_clients = new Dictionary<string, ClientInfo>(StringComparer.CurrentCultureIgnoreCase);            m_queues = new Dictionary<string, RegistrationInfo>(StringComparer.CurrentCultureIgnoreCase);            m_topics = new Dictionary<string, RegistrationInfo>(StringComparer.CurrentCultureIgnoreCase);

Lines 1–500 of 1351. Load more