/*
 *
 *   Copyright (c) 1999 - 2011 my-Channels Ltd
 *   Copyright (c) 2012 - 2017 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.
 *
 */
#include "nSampleApp.h"
#include "nDataGroupListener.h"
#include "nDataStreamListener.h"
#include "nSession.h"
#include "nDataGroup.h"
#include "nEventProperties.h"
#include "nConsumeEvent.h"
#include "nSessionNotConnectedException.h"
#include "nBaseClientException.h"
#include "nSessionPausedException.h"
#include "nSecurityException.h"
#include "nSessionFactory.h"
#include "nUnexpectedResponseException.h"
#include "nRequestTimedOutException.h"

#include <iostream>
#include <thread>
#include "Poco/Thread.h"
#include <sys/timeb.h>

using namespace com::pcbsys::nirvana::client;
using namespace com::pcbsys::nirvana::apps;


class DataGroupPublisher : public nSampleApp, public nDataGroupListener, public nDataStreamListener {
  private:
    bool m_bOk;
    nBaseClientException *m_pAsyncException;
    nDataGroup *m_pGroup;
    bool m_bHasStreams;

    static DataGroupPublisher *m_pSelf;

    /**
    * This method demonstrates the Nirvana API calls necessary to publish to
    * an nDataGroup.
    * It is called after all command line arguments have been received and
    * validated
    *
    * @param realmDetails a String[] containing the possible RNAME values
    * @param groupName the data group name to publish to
    * @param count the number of messages to publish
    * @param size the size (bytes) of each message to be published
    */
  private:
    void doit(std::string *pRealmDetails, int nRealmDetail, const std::string& groupName, int count, int size) {
        m_pSelf->constructSession(pRealmDetails, nRealmDetail, this);

        try {
            std::list<nDataGroup*> *pGrps = m_pSession->getDataGroups(this);

            for (std::list<nDataGroup*>::iterator iterator = pGrps->begin();
                    iterator != pGrps->end(); iterator++) {
                nDataGroup *pGrp = *iterator;

                if (pGrp->getName().compare(groupName) == 0) {
                    m_pGroup = pGrp;
                    m_pGroup->addListener(this);
                    break;
                }

                if (pGrp->delRef())
                    delete pGrp;
            }

            delete pGrps;

            // if group not found, create it
            if (m_pGroup == NULL)
                m_pGroup = m_pSession->createDataGroup(groupName, this);

            // get the default group to determine how many stream users are connected
            nDataGroup *pDefaultGroup = m_pSession->getDefaultDataGroup(this);		// TODO: refs

            // just add each connected stream to the group
            std::list<nDataStream*> *pStreams = pDefaultGroup->getStreams();

            m_bHasStreams = (m_pGroup->size() > 0);

            for (std::list<nDataStream*>::iterator streamIterator = pStreams->begin();
                    streamIterator != pStreams->end(); streamIterator++) {
                nDataStream *pStream = *streamIterator;
                if (pStream->getName().compare(m_pSession->getStreamId()) != 0) {
                    m_pGroup->add(pStream);
                }
            }

            //Create a byte array filled with characters equal to
            // the message size specified. This could be a result
            //of String.getBytes() call in a real world scenario.
            unsigned char *pBuffer = new unsigned char[size];
            for (int x = 0; x < size; x++) {
                pBuffer[x] = (unsigned char)((x % 90) + 32);
            }

            //Instantiate the message to be published with the specified nEventPropeties / byte[]

            nEventProperties *pProps = new nEventProperties();
            //You can add other types in a dictionary object
            pProps->put("key0string", "1");
            pProps->put("key1int", (int)1);
            pProps->put("key2long", (longlong)-11);

            nConsumeEvent *pEvt1 = new nConsumeEvent(pProps, pBuffer, size);

            delete[] pBuffer;

            //Inform the user that publishing is about to start
            std::cout << "Starting publish of "<<count <<" events with a size of "<< size<<" bytes each" <<std::endl;

            //Get a timestamp to be used to calculate the message publishing rates
            struct timeb tm;
            ftime (&tm);
            unsigned long start = (unsigned long) tm.time * 1000 + tm.millitm;

            //Loop as many times as the number of messages we want to publish
            for (int x = 0; x < count; x++) {
                std::cout << "No streams are registered, waiting for notification" << std::endl;
                while (!m_bHasStreams) {
                    Poco::Thread::sleep(1000);
                    if (m_bHasStreams) {
                        std::cout << "Streams are registered starting to publish" << std::endl;
                    }
                }
                try {
                    //Publish the event
                    m_pSession->writeDataGroup(pEvt1, m_pGroup);
                } catch (nSessionNotConnectedException ex) {
                    while (!m_pSession->isConnected()) {
                        std::cout << "Disconnected from Nirvana, Sleeping for 1 second..." <<std::endl;
                        try {
                            Poco::Thread::sleep(1000);
                        } catch (Exception e) {
                        }
                    }
                    x--;
                    //We need to repeat the publish for the event publish that caused the exception,
                    //so we reduce the counter
                } catch (nBaseClientException ex) {
                    std::cout << "Exception : " << ex.message() <<std::endl;
                    throw ex;
                }

                //Check if an asynchronous exception has been received
                if (!m_bOk) {
                    //If it has, then throw it
                    throw *m_pAsyncException;
                }
            }
            //Check if an asynchronous exception has been received
            if (!m_bOk) {
                //If it has, then throw it
                throw *m_pAsyncException;
            }

            if (pProps->delRef())
                delete pProps;

            if (pEvt1->delRef())
                delete pEvt1;

            ftime (&tm);
            unsigned long end = (unsigned long) tm.time * 1000 + tm.millitm;

            //Get a timestamp to calculate the publishing rates
            //Calculate the events / sec rate
            unsigned long dif = end - start;
            long eventPerSec = ((count * 1000) / dif);

            //Calculate the bytes / sec rate
            long bytesPerSec = eventPerSec * size;

            //Inform the user of the resulting rates
            std::cout <<"Publish Completed in "<<dif <<" ms :\n"
                      << "[Events Published = "<<count <<"]  [Events/Second = "<< count <<"]  [Bytes/Second = "<< bytesPerSec<<"]\n"
                      << "Bandwidth data : Bytes Tx ["<< m_pSession->getOutputByteCount() <<"] Bytes Rx ["<< m_pSession->getInputByteCount() <<"]" <<std::endl;
        } catch (nSessionPausedException ps) {
            std::cout << "Session has been paused, please resume the session\n" <<std::endl;
            exit(1);
        } catch (nSecurityException se) {
            std::cout << "Unsufficient permissions for the requested operation.\n"
                      << "Please check the ACL settings on the server" <<std::endl;
            exit(1);
        } catch (nSessionNotConnectedException snce) {
            std::cout << "The session object used is not physically connected to the Nirvana realm.\n"
                      << "Please ensure the realm is up and check your RNAME value" <<std::endl;
            exit(1);
        } catch (nUnexpectedResponseException ure) {
            std::cout << "The Nirvana REALM has returned an unexpected response.\n"
                      << "Please ensure the Nirvana REALM and client API used are compatible" <<std::endl;
            exit(1);
        } catch (nRequestTimedOutException rtoe) {
            std::cout << "The requested operation has timed out waiting for a response from the REALM.\n"
                      <<"If this is a very busy REALM ask your administrator to increase the client timeout values" <<std::endl;
            exit(1);
        }
        //Close the session we opened
        try {
            nSessionFactory::close(m_pSession);
        } catch (Exception ex) {
        }
        //Close any other sessions so that we can exit
        nSessionFactory::shutdown();
    }

    // callback for DataStreamListener
  public:
    ~DataGroupPublisher() {
        if (m_pGroup && m_pGroup->delRef())
            delete m_pGroup;
    }

    void onMessage(nConsumeEvent *pEvt) {
    }

    // new stream added to default data group
    void addedStream(nDataGroup *pGroup, nDataStream *pStream, int count) {
        std::cout << pGroup->getName() <<" : New stream "<< pStream->getName() <<" : "<< pStream->getSubject() <<" added, size now "<< count <<std::endl;
        if ((pGroup->getName().compare(m_pGroup->getName()) == 0) && (count > 0)) {
            m_bHasStreams = true;
            std::cout << pGroup->getName() <<" : "<< count <<" Streams are registered, starting publisher thread" <<std::endl;
        } else {
            if (pStream->getName().compare(m_pSession->getStreamId()) != 0) {
                m_pGroup->add(pStream);
            }
        }
    }

    // stream removed from data group
    void deletedStream(nDataGroup *pGroup, nDataStream *pStream, int count, bool serverRemoved) {
        std::cout << pGroup->getName() <<" : Data stream "<< pStream->getName() <<" : "<< pStream->getSubject() <<" removed, size now "<< count <<std::endl;
        if ((pGroup->getName().compare(m_pGroup->getName()) == 0) && (count == 0)) {
            m_bHasStreams = false;
            std::cout << "No streams are registered, stopping publisher thread" <<std::endl;
        } else if ((pGroup->getName().compare(m_pGroup->getName()) == 0) && (count > 0)) {
            m_bHasStreams = true;
            std::cout << pGroup->getName() <<" : "<< count <<" Streams are registered, starting publisher thread" <<std::endl;
        }
    }

    // new data group added
    void addedGroup(nDataGroup *pTo, nDataGroup *pGroup, int count) {
        std::cout << pTo->getName() << " : Data group " << pGroup->getName() << " added, size now "<<count <<std::endl;
    }

    // data group removed
    void removedGroup(nDataGroup *pFrom, nDataGroup *pGroup, int count) {
        std::cout << pFrom->getName() << " : Data group " << pGroup->getName() << " added, size now "<<count <<std::endl;
    }

    void deletedGroup(nDataGroup *pGroup) {
        std::cout << "Data group deleted : " << pGroup->getName() << std::endl;
    }

    void createdGroup(nDataGroup *pGroup) {
        std::cout << "Data group created : " << pGroup->getName() <<std::endl;
    }

  protected:
    virtual void processArgs(int argc, char** argv) {
        //
        // Need a min of 3, rname, group name
        if (argc < 3) {
            Usage();
            exit(2);
        }
        std::string RNAME = argv[1];

        std::string groupName = argv[2];

        int count = 1000;
        if (argc > 3) {
            count = atoi (argv[3]);
        }

        int size = 1024;
        if (argc > 4) {
            size = atoi (argv[4]);
        }

        //
        // Run the sample app
        //
        int nRproperty = 0;
        std::string *pRproperties = parseRealmProperties(RNAME, nRproperty);

        m_pSelf->doit(pRproperties, nRproperty, groupName, count, size);

        delete[] pRproperties;
    }

  public:
    static int Main(int argc, char** argv) {
        //Create an instance for this class
        m_pSelf = new DataGroupPublisher();

        //Process command line arguments
        m_pSelf->processArgs(argc, argv);

        return 0;
    }

    /**
    * Prints the usage message for this class
    */
  private:
    static void Usage() {
        std::cout <<"Usage ...\n\n"
                  <<"DataGroupPublish <rname> <group name> [count] [size] \n\n"
                  <<"<Required Arguments> \n\n"
                  <<"<rname> - the rname of the server to connect to\n"
                  <<"<group name> - Data group name parameter to publish to\n"
                  <<"\n[Optional Arguments] \n\n"
                  <<"[count] -The number of events to publish (default: 10)\n"
                  <<"[size] - The size (bytes) of the event to publish (default: 100)" <<std::endl;
    }

    DataGroupPublisher() : m_bOk(true), m_pAsyncException(NULL), m_pGroup(NULL),
        m_bHasStreams(false) {
    }
};

DataGroupPublisher* DataGroupPublisher::m_pSelf = NULL;

int main (int argc, char** argv) {
    return DataGroupPublisher::Main (argc, argv);
}