//******************************************************************************************************
//  Program.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:
//  ----------------------------------------------------------------------------------------------------
//  06/30/2011 - J. Ritchie Carroll
//       Generated original version of source code.
//  12/13/2012 - Starlynn Danyelle Gilliam
//       Modified Header.
//
//******************************************************************************************************

using System;
using System.Collections.Generic;
using System.Data;
using System.Text;
using System.Timers;
using GSF;
using GSF.IO;
using GSF.TimeSeries;
using GSF.TimeSeries.Transport;
using Random = GSF.Security.Cryptography.Random;

namespace DataSubscriberTest
{
    class Program
    {
        // Define subscription modes for testing
        static readonly SynchronizedSubscriptionInfo remotelySynchronizedInfo = new SynchronizedSubscriptionInfo(true, 30);
        static readonly SynchronizedSubscriptionInfo locallySynchronizedInfo = new SynchronizedSubscriptionInfo(false, 30);
        static readonly UnsynchronizedSubscriptionInfo unsynchronizedInfo = new UnsynchronizedSubscriptionInfo(false);
        static readonly UnsynchronizedSubscriptionInfo throttledInfo = new UnsynchronizedSubscriptionInfo(true);

        static readonly DataSubscriber subscriber = new DataSubscriber();
        static long dataCount;
        static readonly Timer timer = new Timer(10000);
        static readonly object displayLock = new object();

        private const int MeasurementCount = 200;

        static void Main(string[] args)
        {
            // Attach to subscriber events
            subscriber.StatusMessage += subscriber_StatusMessage;
            subscriber.ProcessException += subscriber_ProcessException;
            subscriber.ConnectionEstablished += subscriber_ConnectionEstablished;
            subscriber.ConnectionTerminated += subscriber_ConnectionTerminated;
            subscriber.NewMeasurements += subscriber_NewMeasurements;
            subscriber.MetaDataReceived += subscriber_MetaDataReceived;

            StringBuilder connectionString = new StringBuilder();

            // Define connection string
            for (int i = 1; i <= MeasurementCount; i++)
            {
                connectionString.AppendFormat("DEVARCHIVE:{0};", i);
            }

            // Set up subscription info objects
            remotelySynchronizedInfo.LagTime = 0.5D;
            remotelySynchronizedInfo.LeadTime = 1.0D;
            remotelySynchronizedInfo.FilterExpression = connectionString.ToString();

            locallySynchronizedInfo.LagTime = 0.5D;
            locallySynchronizedInfo.LeadTime = 1.0D;
            locallySynchronizedInfo.FilterExpression = connectionString.ToString();

            unsynchronizedInfo.FilterExpression = connectionString.ToString();

            throttledInfo.LagTime = 5.0D;
            throttledInfo.LeadTime = 1.0D;
            throttledInfo.FilterExpression = connectionString.ToString();

            // Initialize subscriber
            subscriber.UseSimpleTcpClient = true;
            //subscriber.ConnectionString = "server=tcp://127.0.0.1:9898; useZeroMQChannel=true";
            subscriber.ConnectionString = "server=localhost:9898";
            subscriber.OperationalModes |= OperationalModes.UseCommonSerializationFormat | OperationalModes.CompressMetadata | OperationalModes.CompressSignalIndexCache | OperationalModes.CompressPayloadData;
            //subscriber.CompressionModes = CompressionModes.TSSC | CompressionModes.GZip;
            subscriber.Initialize();

            // Start subscriber connection cycle
            subscriber.Start();

            timer.Elapsed += timer_Elapsed;
            timer.Start();

            Console.ReadLine();

            subscriber.Unsubscribe();

            subscriber.StatusMessage -= subscriber_StatusMessage;
            subscriber.ProcessException -= subscriber_ProcessException;
            subscriber.ConnectionEstablished -= subscriber_ConnectionEstablished;
            subscriber.ConnectionTerminated -= subscriber_ConnectionTerminated;
            subscriber.NewMeasurements -= subscriber_NewMeasurements;

            timer.Elapsed -= timer_Elapsed;
            timer.Stop();
        }

        static void subscriber_MetaDataReceived(object sender, EventArgs<System.Data.DataSet> e)
        {
            DataSet dataSet = e.Argument;
            dataSet.WriteXml(FilePath.GetAbsolutePath("Metadata.xml"), XmlWriteMode.WriteSchema);
            Console.WriteLine("Data set serialized with {0} tables...", dataSet.Tables.Count);
        }

        static void timer_Elapsed(object sender, ElapsedEventArgs e)
        {
            if (subscriber.IsConnected)
            {
                if (Random.Boolean)
                {
                    if (Random.Boolean)
                    {
                        lock (displayLock)
                        {
                            Console.WriteLine("Initiating remotely synchronized subscription...");
                        }
                        subscriber.SynchronizedSubscribe(remotelySynchronizedInfo);
                    }
                    else
                    {
                        lock (displayLock)
                        {
                            Console.WriteLine("Initiating locally synchronized subscription...");
                        }
                        subscriber.SynchronizedSubscribe(locallySynchronizedInfo);
                    }
                }
                else
                {
                    if (Random.Boolean)
                    {
                        lock (displayLock)
                        {
                            Console.WriteLine("Initiating on-change unsynchronized subscription...");
                        }
                        subscriber.UnsynchronizedSubscribe(unsynchronizedInfo);
                    }
                    else
                    {
                        lock (displayLock)
                        {
                            Console.WriteLine("Initiating throttled unsynchronized subscription...");
                        }
                        subscriber.UnsynchronizedSubscribe(throttledInfo);
                    }
                }
            }
        }

        static void subscriber_NewMeasurements(object sender, EventArgs<ICollection<IMeasurement>> e)
        {
            // Check to see if total number of added points will exceed process interval used to show periodic
            // messages of how many points have been archived so far...
            const int interval = 5 * 60;

            bool showMessage = dataCount + e.Argument.Count >= (dataCount / interval + 1) * interval;

            dataCount += e.Argument.Count;

            if (showMessage)
            {
                lock (displayLock)
                {
                    Console.WriteLine(string.Format("{0:N0} measurements have been processed so far...", dataCount));
                }

                //// Occasionally request another cipher key rotation
                //if (GSF.Security.Cryptography.Random.Boolean)
                //    subscriber.SendServerCommand(ServerCommand.RotateCipherKeys);
            }
        }

        static void subscriber_ConnectionEstablished(object sender, EventArgs e)
        {

            //subscriber.SendServerCommand(ServerCommand.MetaDataRefresh);

            //// Request cipher key rotation
            //subscriber.SendServerCommand(ServerCommand.RotateCipherKeys);

            //subscriber.SynchronizedSubscribe(true, 30, 0.5D, 1.0D, "DEVARCHIVE:1;DEVARCHIVE:2;DEVARCHIVE:3;DEVARCHIVE:4;DEVARCHIVE:5");
            //subscriber.SynchronizedSubscribe(true, 30, 0.5D, 1.0D, "DEVARCHIVE:1");
            //subscriber.UnsynchronizedSubscribe(true, "DEVARCHIVE:1;DEVARCHIVE:2;DEVARCHIVE:3;DEVARCHIVE:4;DEVARCHIVE:5");
        }

        static void subscriber_ConnectionTerminated(object sender, EventArgs e)
        {
            subscriber.Start();

            lock (displayLock)
            {
                Console.WriteLine("Connection to publisher was terminated, restarting connection cycle...");
            }
        }

        static void subscriber_ProcessException(object sender, EventArgs<Exception> e)
        {
            lock (displayLock)
            {
                Console.ForegroundColor = ConsoleColor.Red;
                Console.WriteLine("EXCEPTION: " + e.Argument.Message);
                Console.ResetColor();
            }
        }

        static void subscriber_StatusMessage(object sender, EventArgs<string> e)
        {
            lock (displayLock)
            {
                Console.WriteLine(e.Argument);
            }
        }
    }
}