//******************************************************************************************************
//  IClientSubscription.cs - Gbtc
//
//  Copyright © 2010, 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:
//  ----------------------------------------------------------------------------------------------------
//  06/24/2011 - Ritchie
//       Generated original version of source code.
//
//******************************************************************************************************

using GSF.Collections;
using GSF.TimeSeries.Adapters;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Data;
using System.Linq;
using System.Text;

namespace GSF.TimeSeries.Transport
{
    /// <summary>
    /// Represents a common set of interfaces for a client adapter subscription to the <see cref="DataPublisher"/>.
    /// </summary>
    public interface IClientSubscription : IActionAdapter
    {
        /// <summary>
        /// Indicates to the host that processing for an input adapter (via temporal session) has completed.
        /// </summary>
        /// <remarks>
        /// This event is expected to only be raised when an input adapter has been designed to process
        /// a finite amount of data, e.g., reading a historical range of data during temporal procesing.
        /// </remarks>
        event EventHandler<EventArgs<IClientSubscription, EventArgs>> ProcessingComplete;

        /// <summary>
        /// Gets the <see cref="Guid"/> client TCP connection identifier of this <see cref="IClientSubscription"/>.
        /// </summary>
        Guid ClientID
        {
            get;
        }

        /// <summary>
        /// Gets the <see cref="Guid"/> based subscriber ID of this <see cref="IClientSubscription"/>.
        /// </summary>
        Guid SubscriberID
        {
            get;
        }

        /// <summary>
        /// Gets the current signal index cache of this <see cref="IClientSubscription"/>.
        /// </summary>
        SignalIndexCache SignalIndexCache
        {
            get;
        }

        /// <summary>
        /// Gets the input filter requested by the subscriber when establishing this <see cref="IClientSubscription"/>.
        /// </summary>
        string RequestedInputFilter
        {
            get;
        }

        /// <summary>
        /// Gets or sets flag that determines if the compact measurement format should be used in data packets of this <see cref="IClientSubscription"/>.
        /// </summary>
        bool UseCompactMeasurementFormat
        {
            get;
            set;
        }

        /// <summary>
        /// Gets size of timestamp in bytes.
        /// </summary>
        int TimestampSize
        {
            get;
        }

        /// <summary>
        /// Gets or sets host name used to identify connection source of client subscription.
        /// </summary>
        string HostName
        {
            get;
            set;
        }

        /// <summary>
        /// Explictly raises the <see cref="IAdapter.StatusMessage"/> event.
        /// </summary>
        /// <param name="status">New status message.</param>
        void OnStatusMessage(string status);

        /// <summary>
        /// Explictly raises the <see cref="IAdapter.ProcessException"/> event.
        /// </summary>
        /// <param name="ex">Processing <see cref="Exception"/>.</param>
        void OnProcessException(Exception ex);

        /// <summary>
        /// Explictly raises the <see cref="IInputAdapter.ProcessingComplete"/> event.
        /// </summary>
        /// <param name="sender"><see cref="IInputAdapter"/> raising the notification.</param>
        /// <param name="e">Event arguments for notification, if any.</param>
        void OnProcessingCompleted(object sender, EventArgs e);
    }

    /// <summary>
    /// Defines static extension functions for <see cref="IClientSubscription"/> implementations.
    /// </summary>
    public static class IClientSubscriptionExtensions
    {
        // Define cache of dynamically defined event handlers associated with each client subscription
        private static readonly ConcurrentDictionary<IClientSubscription, EventHandler<EventArgs<string, UpdateType>>> s_statusMessageHandlers = new ConcurrentDictionary<IClientSubscription, EventHandler<EventArgs<string, UpdateType>>>();
        private static readonly ConcurrentDictionary<IClientSubscription, EventHandler<EventArgs<Exception>>> s_processExceptionHandlers = new ConcurrentDictionary<IClientSubscription, EventHandler<EventArgs<Exception>>>();
        private static readonly ConcurrentDictionary<IClientSubscription, EventHandler> s_processingCompletedHandlers = new ConcurrentDictionary<IClientSubscription, EventHandler>();

        /// <summary>
        /// Returns a new temporal <see cref="IaonSession"/> for a <see cref="IClientSubscription"/>.
        /// </summary>
        /// <param name="clientSubscription"><see cref="IClientSubscription"/> instance to create temporal <see cref="IaonSession"/> for.</param>
        /// <returns>New temporal <see cref="IaonSession"/> for a <see cref="IClientSubscription"/>.</returns>
        public static IaonSession CreateTemporalSession(this IClientSubscription clientSubscription)
        {
            IaonSession session;

            // Cache the specified input measurement keys requested by the remote subscription
            // internally since these will only be needed in the private Iaon session
            MeasurementKey[] inputMeasurementKeys = clientSubscription.InputMeasurementKeys;
            IMeasurement[] outputMeasurements = clientSubscription.OutputMeasurements;

            // Since historical data is requested, we "turn off" interaction with the outside real-time world
            // by removing this adapter from external routes. To accomplish this we expose I/O demands for an
            // undefined measurement. Note: assigning to null would mean "broadcast" of all data is desired.
            clientSubscription.InputMeasurementKeys = new MeasurementKey[] { MeasurementKey.Undefined };
            clientSubscription.OutputMeasurements = new IMeasurement[] { Measurement.Undefined };

            // Create a new Iaon session
            session = new IaonSession();
            session.Name = "<" + clientSubscription.HostName.ToNonNullString("unavailable") + ">@" + clientSubscription.StartTimeConstraint.ToString("yyyy-MM-dd HH:mm:ss");

            // Assign requested input measurement keys as a routing restriction
            session.InputMeasurementKeysRestriction = inputMeasurementKeys;

            // Setup default bubbling event handlers associated with the client session adapter
            EventHandler<EventArgs<string, UpdateType>> statusMessageHandler = (sender, e) =>
            {
                if (e.Argument2 == UpdateType.Information)
                    clientSubscription.OnStatusMessage(e.Argument1);
                else
                    clientSubscription.OnStatusMessage("0x" + (int)e.Argument2 + e.Argument1);
            };
            EventHandler<EventArgs<Exception>> processExceptionHandler = (sender, e) => clientSubscription.OnProcessException(e.Argument);
            EventHandler processingCompletedHandler = (sender, e) => clientSubscription.OnProcessingCompleted(sender, e);

            // Cache dynamic event handlers so they can be detached later
            s_statusMessageHandlers[clientSubscription] = statusMessageHandler;
            s_processExceptionHandlers[clientSubscription] = processExceptionHandler;
            s_processingCompletedHandlers[clientSubscription] = processingCompletedHandler;

            // Attach handlers to new session - this will proxy all temporal session messages through the client session adapter
            session.StatusMessage += statusMessageHandler;
            session.ProcessException += processExceptionHandler;
            session.ProcessingComplete += processingCompletedHandler;

            // Send the first message indicating a new temporal session is being established
            statusMessageHandler(null, new EventArgs<string, UpdateType>(
                string.Format("Initializing temporal session for host \"{0}\" spanning {1} to {2} processing data {3}...",
                    clientSubscription.HostName.ToNonNullString("unknown"),
                    clientSubscription.StartTimeConstraint.ToString("yyyy-MM-dd HH:mm:ss.fff"),
                    clientSubscription.StopTimeConstraint.ToString("yyyy-MM-dd HH:mm:ss.fff"),
                    clientSubscription.ProcessingInterval == 0 ? "as fast as possible" :
                    clientSubscription.ProcessingInterval == -1 ? "at the default rate" : "at " + clientSubscription.ProcessingInterval + "ms intervals"),
                UpdateType.Information));

            // Duplicate current real-time session configuration for adapters that report temporal support
            session.DataSource = IaonSession.ExtractTemporalConfiguration(clientSubscription.DataSource);

            // Define an in-situ action adapter for the temporal Iaon session used to proxy data back to the client subscription
            DataTable actionAdapters = session.DataSource.Tables["ActionAdapters"];
            DataRow proxyAdapterRow = actionAdapters.NewRow();

            // Define connection string for proxy adapter based on original inputs and ouputs as requested by client subscription
            StringBuilder connectionString = new StringBuilder();

            if (inputMeasurementKeys != null && inputMeasurementKeys.Length > 0)
                connectionString.AppendFormat("inputMeasurementKeys={{{0}}}", inputMeasurementKeys.Select(key => key.SignalID).ToDelimitedString(";"));

            if (outputMeasurements != null && outputMeasurements.Length > 0)
            {
                if (connectionString.Length > 0)
                    connectionString.Append("; ");

                connectionString.AppendFormat("outputMeasurements={{{0}}}", outputMeasurements.Select(m => m.ID).ToDelimitedString(";"));
            }

            // Assign critical adapter properties
            proxyAdapterRow["ID"] = 0;
            proxyAdapterRow["AdapterName"] = "PROXY!SERVICES";
            proxyAdapterRow["AssemblyName"] = "GSF.TimeSeries.dll";
            proxyAdapterRow["TypeName"] = "GSF.TimeSeries.Transport.TemporalClientSubscriptionProxy";
            proxyAdapterRow["ConnectionString"] = connectionString.ToString();

            // Add proxy row to Iaon action adapter definitions
            actionAdapters.Rows.Add(proxyAdapterRow);

            // Initialize temporal session adapters without starting them
            session.Initialize(false);

            // Get reference to temporal session proxy adapter to it can be associated with the client subscription
            TemporalClientSubscriptionProxy proxyAdapter = null;
            IActionAdapter adapter0;

            // Lookup proxy adapter defined with reserved ID zero
            if (session.ActionAdapters.TryGetAdapterByID(0, out adapter0))
                proxyAdapter = adapter0 as TemporalClientSubscriptionProxy;

            if ((object)proxyAdapter == null)
                throw new InvalidOperationException("Failed to define temporal subscription proxy adapter - cannot complete temporal session initialization.");

            // Associate proxy with client subscription
            proxyAdapter.Parent = clientSubscription;

            // Load current temporal constraint parameters
            Dictionary<string, string> settings = clientSubscription.Settings;
            string startTime, stopTime, parameters;

            settings.TryGetValue("startTimeConstraint", out startTime);
            settings.TryGetValue("stopTimeConstraint", out stopTime);
            settings.TryGetValue("timeConstraintParameters", out parameters);

            // Assign requested temporal constraints to all private session adapters
            session.AllAdapters.SetTemporalConstraint(startTime, stopTime, parameters);
            session.AllAdapters.ProcessingInterval = clientSubscription.ProcessingInterval;

            // Start temporal session adapters
            session.AllAdapters.Start();

            // Recalculate routing tables to accomodate addtion of proxy adapter and handle
            // input measurement keys restriction
            session.RecalculateRoutingTables();

            return session;
        }

        /// <summary>
        /// Disposes a temporal <see cref="IaonSession"/> created using <see cref="CreateTemporalSession"/>.
        /// </summary>
        /// <param name="adapter"><see cref="IClientSubscription"/> source instance.</param>
        /// <param name="session"><see cref="IaonSession"/> instance to dispose.</param>
        public static void DisposeTemporalSession(this IClientSubscription adapter, ref IaonSession session)
        {
            if (session != null)
            {
                EventHandler<EventArgs<string, UpdateType>> statusMessageFunction;
                EventHandler<EventArgs<Exception>> processExceptionFunction;
                EventHandler processingCompletedFunction;

                // Remove and detatch from event handlers
                if (s_statusMessageHandlers.TryRemove(adapter, out statusMessageFunction))
                    session.StatusMessage -= statusMessageFunction;

                if (s_processExceptionHandlers.TryRemove(adapter, out processExceptionFunction))
                    session.ProcessException -= processExceptionFunction;

                if (s_processingCompletedHandlers.TryRemove(adapter, out processingCompletedFunction))
                    session.ProcessingComplete -= processingCompletedFunction;

                session.Dispose();
            }

            session = null;
        }
    }
}