/*
Copyright 1999-2011 (c) My-Channels
Copyright (c) 2012-2014 Software AG, Darmstadt, Germany and/or Software AG USA Inc., Reston, VA, USA, and/or its subsidiaries and/or its affiliates and/or their licensors.

Use, reproduction, transfer, publication or disclosure is prohibited except as specifically provided for in your License Agreement with Software AG.
*/

using System;
using System.Threading;

using com.pcbsys.nirvana.client;

namespace com.pcbsys.nirvana.apps
{
    /**
     * Uses a channel iterator to consume events from a nirvana channel.
     * 
     * This is a synchronous method of event retrieval
     */
    class channeliterator : nSampleApp
    {

        private static System.Text.UTF8Encoding encoding = new System.Text.UTF8Encoding();

        private long lastEID = 0;
        private DateTime startTime;
        private long byteCount = 0;

        private int logLevel = 0;
        private int count = -1;
        private int totalMsgs = 0;
        private int reportCount = 1000;

        private nChannel myChannel;
        private nChannelIterator iterator = null;
        private Thread channelReader = null;
        private static channeliterator mySelf = null;

        /**
         * This method demonstrates the Nirvana API calls necessary to consume 
         * events from a channel using a channel iterator
         * 
         * It is called after all command line arguments have been received and
         * validated
         *
         * @param realmDetails a String[] containing the possible RNAME values
         * @param achannelName the channel name to create
         * @param selector the subscription selector filter
         * @param startEid the eid to start subscribing from
         * @param loglvl the specified log level
         * @param repCount the specified report count
         */
        private void doit(String[] realmDetails, String achannelName, String sel, long start, int loglvl, int repCount)
        {
            logLevel = loglvl;
            reportCount = repCount;

            mySelf.constructSession(realmDetails);

            //Creates the iterator on specified channel
            try
            {
                //Create a channel attributes object
                nChannelAttributes nca = new nChannelAttributes();
                nca.setName(achannelName);

                //Obtain the channel reference
                myChannel = mySession.findChannel(nca);

                //if the latest event has been implied (by specifying -1)
                if (start == -1)
                {
                    //Get the last eid on the channel and reset the start eid with that value
                    start = myChannel.getLastEID() + 1;
                }

                //create the channel iterator object and the read thread
                iterator = myChannel.createIterator(sel, start);
                ThreadStart ts = new ThreadStart(run);
                channelReader = new Thread(ts);
                channelReader.IsBackground = true;
                channelReader.Start();
                channelReader.Name = "IteratorReadThread";

                //Stay subscribed until the user presses any key
                Console.WriteLine("Press any key to quit !");
                try
                {
                    Console.ReadLine();
                }
                catch (Exception read) { } //Ignore this

                Console.WriteLine("Finished. Consumed total of " + totalMsgs);
                iterator.close();
            }
            //Handle errors
            catch (nChannelNotFoundException cnfe)
            {
                Console.WriteLine("The channel specified could not be found.");
                Console.WriteLine("Please ensure that the channel exists in the REALM you connect to.");
                Environment.Exit(1);
            }
            catch (nSecurityException se)
            {
                Console.WriteLine("Unsufficient permissions for the requested operation.");
                Console.WriteLine("Please check the ACL settings on the server.");
                Environment.Exit(1);
            }
            catch (nSessionNotConnectedException snce)
            {
                Console.WriteLine("The session object used is not physically connected to the Nirvana Realm.");
                Console.WriteLine("Please ensure the realm is running and check your RNAME value.");
                Environment.Exit(1);
            }
            catch (nUnexpectedResponseException ure)
            {
                Console.WriteLine("The Nirvana REALM has returned an unexpected response.");
                Console.WriteLine("Please ensure the Nirvana REALM and client API used are compatible.");
                Environment.Exit(1);
            }
            catch (nUnknownRemoteRealmException urre)
            {
                Console.WriteLine("The channel specified resided in a remote realm which could not be found.");
                Console.WriteLine("Please ensure the channel name specified is correct.");
                Environment.Exit(1);
            }
            catch (nRequestTimedOutException rtoe)
            {
                Console.WriteLine("The requested operation has timed out waiting for a response from the REALM.");
                Console.WriteLine("If this is a very busy REALM ask your administrator to increase the client timeout values.");
                Environment.Exit(1);
            }
            catch (nBaseClientException nbce)
            {
                Console.WriteLine("An error occured while creating the Channel Attributes object.");
                Environment.Exit(1);
            }
            //Close the session we opened
            try
            {
                nSessionFactory.close(mySession);
            }
            catch (Exception ex) { }
            //Close any other sessions so that we can exit
            nSessionFactory.shutdown();
        }

        public void run()
        {
            while (true)
            {
                try
                {
                    nConsumeEvent evt = iterator.getNext(-1);
                    if (evt != null)
                    {
                        go(evt);
                    }
                }
                catch (nRequestTimedOutException timeOut)
                {
                    // no event to consume
                }
                catch (Exception e)
                {
                    break;
                }
            }
        }

        /**
         * Prints the usage message for this class
         */
        private static void Usage()
        {

            Console.WriteLine("Usage ...\n");
            Console.WriteLine("channeliterator <rname> <channel name> [start eid] [debug] [count] [selector] \n");
            Console.WriteLine(
              "<Required Arguments> \n");
            Console.WriteLine(
              "<rname> - the rname of the server to connect to");
            Console.WriteLine(
              "<channel name> - Channel name parameter for the channel to subscribe to");
            Console.WriteLine(
              "\n[Optional Arguments] \n");
            Console.WriteLine(
              "[start eid] - The Event ID to start subscribing from");
            Console.WriteLine(
              "[debug] - The level of output from each event, 0 - none, 1 - summary, 2 - EIDs, 3 - All");
            Console.WriteLine(
              "[count] - The number of events to wait before printing out summary information");
            Console.WriteLine(
              "[selector] - The event filter string to use");
        }

        protected override void processArgs(String[] args)
        {
            //
            // Need a min of 2, rname, channel name
            if( args.Length < 2) 
            {
                Usage();
                Environment.Exit(2);
            }
            String RNAME = args[0]; ;

            var channelName = args[1];

            var start = 0;
            if (args.Length > 2)
            {
                start = Convert.ToInt32(args[2]);
            }

            var loglvl = 3;
            if (args.Length > 3)
            {
                loglvl = Convert.ToInt32(args[3]);
            }
            var report = 1000;
            if (args.Length > 4)
            {
                report = Convert.ToInt32(args[4]);
            }
            String sel = null;
            if (args.Length > 5)
            {
                sel = args[5];
            }

            //
            // Run the sample app
            //
            mySelf.doit(parseRealmProperties(RNAME), channelName, sel, start, loglvl, report);
        }

        static void Main(string[] args)
        {
            //Create an instance for this class
            mySelf = new channeliterator();

            //Process command line arguments
            mySelf.processArgs(args);
        }

        /**
         * A callback is received by the API to this method each time an event is received from
         * the nirvana channel. Be carefull not to spend too much time processing the message
         * inside this method, as until it exits the next message can not be pushed.
         *
         * @param evt An nConsumeEvent object containing the message received from the channel
         */
        public void go(nConsumeEvent evt)
        {
            //If this is the first message we receive
            if (count == -1)
            {
                //Get a timestamp to be used for message rate calculations
                startTime = DateTime.Now;
                count = 0;
            }

            //Increment he counter
            count++;
            totalMsgs++;

            //Have we reached the point where we need to report the rates?
            if (count == reportCount)
            {

                //Reset the counter
                count = 0;

                //Get a timestamp to calculate the rates
                DateTime end = DateTime.Now;

                // Does the specified log level permit us to print on the screen?
                if (logLevel >= 1)
                {

                    //Dump the rates on the screen
                    if (!end.Equals(startTime))
                    {
                        TimeSpan ts = end - startTime;
                        Console.WriteLine("Received " + reportCount + " in " + (ts.Milliseconds) + " Evt/Sec = " + ((reportCount * 1000) / (ts.Milliseconds)) + " Bytes/sec=" + ((byteCount * 1000) / (ts.Milliseconds)));
                        Console.WriteLine("Bandwidth data : Bytes Tx [" + mySession.getOutputByteCount() + "] Bytes Rx [" + mySession.getInputByteCount() + "]");
                    }
                    else
                    {
                        Console.WriteLine("Received " + reportCount + " faster than the system millisecond counter");
                    }
                }
                //Set the startTime for the next report equal to the end timestamp for the previous one
                startTime = end;

                //Reset the byte counter
                byteCount = 0;
            }

            //If the last EID counter is not equal to the current event ID
            if (lastEID != evt.getEventID())
            {
                //If yes, maybe we have missed an event, so print a message on the screen.
                //This message could be printed for a number of other reasons.
                //One of them would be someone purging a range creating an 'eid gap'.
                //As eids are never reused within a channel you could have a situation
                //where this gets printed but nothing is missed.
                Console.WriteLine("Expired event range " + (lastEID) + " - " + (evt.getEventID() - 1));
                //Reset the last eid counter
                lastEID = evt.getEventID() + 1;
            }
            else
            {

                //Increment the last eid counter
                lastEID++;
            }
            //Get the data of the message
            byte[] buffer = evt.getEventData();
            if (buffer != null)
            {
                //Add its length to the byte counter
                byteCount += buffer.Length;
            }
            //If the loglevel permits printing on the screen
            if (logLevel >= 2)
            {
                //Print the eid
                Console.WriteLine("Event id : " + evt.getEventID());
                if (evt.isEndOfChannel())
                {
                    Console.WriteLine("End of channel reached");
                }
                //If the loglevel permits printing on the screen
                if (logLevel >= 3)
                {
                    //Print the message tag
                    Console.WriteLine("Event tag : " + evt.getEventTag());
                    //Print the message data
                    Console.WriteLine("Event data : " + encoding.GetString(evt.getEventData()));
                    if (evt.hasAttributes())
                    {
                        displayEventAttributes(evt.getAttributes());
                    }

                    nEventProperties prop = evt.getProperties();
                    if (prop != null)
                    {
                        displayEventProperties(prop);
                    }
                }
            }
        }
    }
}