//******************************************************************************************************
//  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 Eclipse Public License -v 1.0 (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/eclipse-1.0.php
//
//  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 bug 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.
//
//******************************************************************************************************

using System;
using System.Collections.Generic;
using System.ServiceModel;
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)]
    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 ProcessList<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);
            m_clientsLock = new ReaderWriterLockSlim();
            m_queuesLock = new ReaderWriterLockSlim();
            m_topicsLock = new ReaderWriterLockSlim();
        }

        #endregion

Lines 1–500 of 1236. Load more