//******************************************************************************
// RCF - Remote Call Framework
// Copyright (c) 2005 - 2007. All rights reserved.
// Consult your license for conditions of use.
// Developed by Jarl Lindrud.
// Contact: jlindrud@hotmail.com .
//******************************************************************************

#ifndef INCLUDE_RCF_TCPIOCPSERVERTRANSPORT_HPP
#define INCLUDE_RCF_TCPIOCPSERVERTRANSPORT_HPP

#include <map>
#include <memory>
#include <vector>

#include <boost/noncopyable.hpp>
#include <boost/scoped_ptr.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/weak_ptr.hpp>

#include <RCF/AsyncFilter.hpp>
#include <RCF/IpAddress.hpp>
#include <RCF/IpServerTransport.hpp>
#include <RCF/ServerTask.hpp>
#include <RCF/ServerTransport.hpp>
#include <RCF/Service.hpp>
#include <RCF/ThreadLibrary.hpp>

namespace RCF {

    class RcfServer;

    namespace TcpIocp {

        class Iocp;

        class                                           SessionState;
        typedef boost::shared_ptr<SessionState>         SessionStatePtr;
        typedef boost::weak_ptr<SessionState>           SessionStateWeakPtr;
        class                                           ServerTransport;

        typedef int                                     Fd;
        typedef std::vector<ByteBuffer>                 ByteBuffers;
        typedef std::vector<FilterPtr>                  Filters;

        // Iocp

        class Iocp
        {
        public:
            Iocp(int nMaxConcurrency = -1);
            ~Iocp();

            BOOL Create(
                int nMaxConcurrency = 0);

            BOOL AssociateDevice(
                HANDLE hDevice,
                ULONG_PTR CompKey);

            BOOL AssociateSocket(
                SOCKET hSocket,
                ULONG_PTR CompKey);

            BOOL PostStatus(
                ULONG_PTR CompKey,
                DWORD dwNumBytes = 0,
                OVERLAPPED* po = NULL) ;

            BOOL GetStatus(
                ULONG_PTR* pCompKey,
                PDWORD pdwNumBytes,
                OVERLAPPED** ppo,
                DWORD dwMilliseconds = INFINITE);

        private:
            HANDLE m_hIOCP;
        };

        // FilterProxy

        class FilterProxy : public RCF::IdentityFilter
        {
        public:
            FilterProxy(
                SessionState &sessionState,
                Filter &filter,
                bool top);

        private:
            void read(
                const ByteBuffer &byteBuffer,
                std::size_t bytesRequested);

            void write(
                const std::vector<ByteBuffer> &byteBuffers);

            void onReadCompleted(
                const ByteBuffer &byteBuffer,
                int error);

            void onWriteCompleted(
                std::size_t bytesTransferred,
                int error);

            const FilterDescription &getFilterDescription() const
            {
                RCF_ASSERT(0);
                return * (const FilterDescription *) NULL;
            }

            SessionState &                      mSessionState;
            Filter &                            mFilter;
            bool                                mTop;
        };

        // SessionState
        class SessionState :
            public OVERLAPPED,
            public boost::enable_shared_from_this<SessionState>
        {
        public:

            enum State
            {
                Accepting,
                ReadingDataCount,
                ReadingData,
                WritingData,
                Ready
            };

            enum PostState
            {
                Reading,
                Writing
            };

            SessionState(
                ServerTransport &transport,
                Fd fd);

            ~SessionState();

            void setTransportFilters(
                const std::vector<FilterPtr> &filters);

            const std::vector<FilterPtr> &getTransportFilters();

            void clearOverlapped();

            int read(
                ByteBuffer &byteBuffer,
                std::size_t bufferLen);

            int write(
                const std::vector<ByteBuffer> &byteBuffers);

            void onReadWriteCompleted(
                std::size_t bytesTransferred,
                int error);

            void onFilteredReadCompleted(
                const ByteBuffer &byteBuffer,
                int error);

            void onFilteredWriteCompleted(
                std::size_t bytesTransferred,
                int error);

            void onAcceptCompleted();

        private:

            friend class ServerTransport;
            friend class Proactor;
            friend class FilterProxy;

            int                         mError;
            std::vector<WSABUF>         mWsabufs;

            SessionStateWeakPtr         mReflectionSessionStateWeakPtr;

            State                       mState;
            PostState                   mPostState;
            SessionPtr                  mSessionPtr;

            boost::shared_ptr<std::vector<char> > mReadBufferPtr;
            boost::shared_ptr<std::vector<char> > mReadBufferSecondaryPtr;

            ByteBuffer                  mTempByteBuffer;
            std::vector<ByteBuffer>     mWriteByteBuffers;
            std::size_t                 mReadBufferRemaining;
            std::vector<char>           mWriteBuffer;
            std::size_t                 mWriteBufferRemaining;
            const Fd                    mFd;
            std::vector<FilterPtr>      mTransportFilters;
            ServerTransport &           mTransport;
            IpAddress                   mLocalAddress;
            IpAddress                   mRemoteAddress;

            SessionStateWeakPtr         mWeakThisPtr;
            SessionStatePtr             mThisPtr;

            void wsaRecv(
                const ByteBuffer &,
                std::size_t);

            void wsaSend(
                const std::vector<ByteBuffer> &);

            void reflect(
                std::size_t bytesRead);

            std::vector<char> &         getReadBuffer();
            std::vector<char> &         getUniqueReadBuffer();
            ByteBuffer                  getReadByteBuffer() const;

            std::vector<char> &         getReadBufferSecondary();
            std::vector<char> &         getUniqueReadBufferSecondary();
            ByteBuffer                  getReadByteBufferSecondary() const;

        private:
            bool                        mCloseAfterWrite;
            bool                        mReflected;

            bool                        mSynchronized;
            bool                        mOwnFd;
            bool                        mHasBeenClosed;
            MutexPtr                    mMutexPtr;
        };

        // Proactor

        class Proactor : public I_Proactor
        {
        public:
            Proactor(
                ServerTransport &transport,
                const SessionStatePtr &sessionStatePtr);

            void                        postRead();
            void                        postWrite(const ByteBuffers &byteBuffers);
            void                        postClose();
            ByteBuffer                  getReadByteBuffer();
            I_ServerTransport &         getServerTransport();
            SessionState &              getSessionState();
            SessionStatePtr             getSessionStatePtr() const;
            const I_RemoteAddress &     getRemoteAddress();
            void                        setTransportFilters(const Filters &filters);
            const std::vector<FilterPtr> &getTransportFilters();

        private:
            // need to break the cycle SessionState->Session->Proactor->SessionState
            boost::weak_ptr<SessionState>   sessionStatePtr;
            ServerTransport &               transport;
        };


        // ServerTransport

        class ServerTransport :
            public I_ServerTransport,
            public I_ServerTransportEx,
            public I_IpServerTransport,
            public I_Service,
            boost::noncopyable
        {
        private:

            typedef TcpIocp::SessionState               SessionState;
            typedef TcpIocp::Proactor                   Proactor;
            typedef TcpIocp::FilterProxy                FilterProxy;
            typedef TcpIocp::Iocp                       Iocp;
            typedef TcpIocp::SessionStatePtr            SessionStatePtr;
            typedef TcpIocp::SessionStateWeakPtr        SessionStateWeakPtr;
            typedef TcpIocp::Fd                         Fd;

            friend class TcpIocp::SessionState;
            friend class TcpIocp::Proactor;

        public:
            ServerTransport(int port = 0);
            ServerTransport(const std::string &networkInterface, int port = 0);

            ServerTransportPtr clone();

            void open();

            void close();

            void cycle(
                int timeoutMs,
                const volatile bool &stopFlag);

            bool cycleTransportAndServer(
                RcfServer &server,
                int timeoutMs,
                const volatile bool &stopFlag);

            void setPort(
                int port);

            int getPort() const;

            // this is the size going into wsasend()/wsarecv(), not the max message size!
            void setMaxSendRecvSize(
                std::size_t maxSendRecvSize);

            std::size_t getMaxSendRecvSize() const;

            void setMaxPendingConnectionCount(
                std::size_t maxPendingConnectionCount);

            std::size_t getMaxPendingConnectionCount() const;

            void setSessionManager(
                I_SessionManager &sessionManager);

            I_SessionManager &getSessionManager();

        private:

            SessionStatePtr createSession(
                int fd);

            void closeSession(
                const SessionStateWeakPtr &sessionStateWeakPtr,
                int fd = -1);

            void transition(
                const SessionStatePtr &sessionStatePtr);

            void stopAccepts();

            bool cycleAccepts(
                int timeoutMs,
                const volatile bool &stopFlag);

            void generateAccepts();

            void onReadWriteCompleted(
                const SessionStateWeakPtr &sessionStateWeakPtr,
                std::size_t bytesTransferred,
                int error);

            void onFilteredReadCompleted(
                const SessionStateWeakPtr &sessionStateWeakPtr,
                const ByteBuffer &byteBuffer,
                int error);

            void onFilteredWriteCompleted(
                const SessionStateWeakPtr &sessionStateWeakPtr,
                std::size_t bytesTransferred,
                int error);

            void postWrite(
                const SessionStatePtr &sessionStatePtr);

            void postWrite(
                const SessionStatePtr &sessionStatePtr,
                const std::vector<ByteBuffer> &byteBuffers);

            void postRead(
                const SessionStatePtr &sessionStatePtr);

            void flushIocp() const;

            // I_ServerTransportEx implementation
        private:

            ClientTransportAutoPtr createClientTransport(
                const I_Endpoint &endpoint);

            ClientTransportAutoPtr createClientTransport(
                SessionPtr sessionPtr);

            SessionPtr createServerSession(
                ClientTransportAutoPtr clientTransportAutoPtr);

            bool reflect(
                const SessionPtr &sessionPtr1,
                const SessionPtr &sessionPtr2);

            bool reflect(
                const SessionStatePtr &sessionStatePtr1,
                const SessionStatePtr &sessionStatePtr2);

            bool isConnected(const SessionPtr &sessionPtr);

            // I_Service implementation
        private:
            void onServiceAdded(RcfServer &server);
            void onServiceRemoved(RcfServer &server);
            void onServerStart(RcfServer &server);
            void onServerStop(RcfServer &server);
            void onServerOpen(RcfServer &server);
            void onServerClose(RcfServer &server);
            bool mOpen;

            // member variables
        private:

            I_SessionManager *          mpSessionManager;
            std::string                 mAcceptorInterface;
            int                         mAcceptorPort;
            Fd                          mAcceptorFd;
            int                         mPort;
            std::size_t                 mMaxPendingConnectionCount;
            std::size_t                 mMaxSendRecvSize;
            std::auto_ptr<Iocp>         mIocpAutoPtr;

            Mutex                       mQueuedAcceptsMutex;
            Condition                   mQueuedAcceptsCondition;

            // access via InterlockedIncrement()/InterlockedDecrement()
            unsigned int                mQueuedAccepts;

            const unsigned int          mQueuedAcceptsThreshold;
            const unsigned int          mQueuedAcceptsAugment;

            LPFN_ACCEPTEX               mlpfnAcceptEx;
            LPFN_GETACCEPTEXSOCKADDRS   mlpfnGetAcceptExSockAddrs;

            volatile bool               mStopFlag;
        };

    } // namespace TcpIocp

    typedef TcpIocp::ServerTransport                    TcpIocpServerTransport;

    typedef boost::shared_ptr<TcpIocpServerTransport>   TcpIocpServerTransportPtr;

} // namespace RCF

#endif // ! INCLUDE_RCF_TCPIOCPSERVERTRANSPORT_HPP