void ClientStub::onTimerExpired()
    {
        AsyncOpType opType = mAsyncOpType;
        mAsyncOpType = None;

        if (opType == Wait)
        {
            scheduleAmiNotification();
        }
        else
        {
            switch(opType)
            {
            case Connect:
                RCF_ASSERT(mEndpoint.get());
                
                onError(RCF::Exception(_RcfError_ClientConnectTimeout(
                    mConnectTimeoutMs, 
                    mEndpoint->asString())));

                break;

            case Write:
                onError(RCF::Exception(_RcfError_ClientWriteTimeout()));
                break;

            case Read: 
                onError(RCF::Exception(_RcfError_ClientReadTimeout()));
                break;

            default:
                RCF_ASSERT(0)(opType);
            };
        }        
    }

    void ClientStub::createFilterSequence(
        std::vector<FilterPtr> & filters)
    {
        filters.clear();
    
        // Setup compression if configured.
        if (mEnableCompression)
        {
#if RCF_FEATURE_ZLIB==1
            FilterPtr filterPtr( new ZlibStatefulCompressionFilter() );
            filters.push_back(filterPtr);
#else
            RCF_ASSERT(0);
#endif
        }

        FilterPtr filterPtr;
        if (mTransportProtocol != Tp_Clear && mTransportProtocol != Tp_Unspecified)
        {
            switch (mTransportProtocol)
            {
#if RCF_FEATURE_SSPI==1
            case Tp_Ntlm:       filterPtr.reset( new NtlmFilter(this) ); break;
            case Tp_Kerberos:   filterPtr.reset( new KerberosFilter(this) ); break;
            case Tp_Negotiate:  filterPtr.reset( new NegotiateFilter(this) ); break;
#endif

#if RCF_FEATURE_OPENSSL==1 && RCF_FEATURE_SSPI==1
            case Tp_Ssl:        if (mSslImplementation == Si_Schannel)
                                {
                                    filterPtr.reset( new SchannelFilter(this) ); 
                                }
                                else
                                {
                                    RCF_ASSERT(mSslImplementation == Si_OpenSsl);
                                    filterPtr.reset( new OpenSslEncryptionFilter(this) ); 
                                }
                                break;
#elif RCF_FEATURE_OPENSSL==1
            case Tp_Ssl:        filterPtr.reset( new OpenSslEncryptionFilter(this) ); break;
#elif RCF_FEATURE_SSPI==1
            case Tp_Ssl:        filterPtr.reset( new SchannelFilter(this) ); break;
#else
            // Single case just to keep the compiler warnings quiet.
            case Tp_Ssl:
#endif

            default:
                RCF_THROW( Exception( _RcfError_TransportProtocolNotSupported( getTransportProtocolName(mTransportProtocol)) ) );
            }
        }
        if (filterPtr)
        {
            filters.push_back(filterPtr);
        }
    }

void AsioSessionState::onAppReadWriteCompleted(
    size_t bytesTransferred)
{
    RCF_ASSERT(!mReflecting);
    switch(mState)
    {
    case ReadingDataCount:
    case ReadingData:

        if (mTransport.mCustomFraming)
        {
            doCustomFraming(bytesTransferred);
        }
        else
        {
            doRegularFraming(bytesTransferred);
        }

        break;

    case WritingData:

        RCF_ASSERT_LTEQ(bytesTransferred , mWriteBufferRemaining);

        mWriteBufferRemaining -= bytesTransferred;
        if (mWriteBufferRemaining > 0)
        {
            beginWrite();
        }
        else
        {
            if (mCloseAfterWrite)
            {
                // For TCP sockets, call shutdown() so client receives
                // the message before we close the connection.

                implCloseAfterWrite();
            }
            else
            {
                mState = Ready;

                mSlicedWriteByteBuffers.resize(0);
                mWriteByteBuffers.resize(0);

                mTransport.getSessionManager().onWriteCompleted(
                    getSessionPtr());
            }
        }
        break;

    default:
        RCF_ASSERT(0);
    }
}

    void ObjectPool::put(VecPtr & vecPtr)
    {
        RCF_ASSERT(vecPtr);
        RCF_ASSERT(vecPtr.unique());

        vecPtr->resize(0);

        Lock lock(mVecPtrPoolMutex);
        mVecPtrPool.push_back(vecPtr);
        vecPtr.reset();
    }

    HttpsClientTransport::HttpsClientTransport(const HttpsEndpoint & httpsEndpoint) : 
        TcpClientTransport(httpsEndpoint.getIp(), httpsEndpoint.getPort())
    {
        std::vector<FilterPtr> wireFilters;

        // HTTP framing.
        wireFilters.push_back( FilterPtr( new HttpFrameFilter(
            getRemoteAddr().getIp(), 
            getRemoteAddr().getPort())));

        // SSL.
        ClientStub * pClientStub = getTlsClientStubPtr();
        RCF_ASSERT(pClientStub);

        FilterPtr sslFilterPtr;

#if RCF_FEATURE_SSPI==1 && RCF_FEATURE_OPENSSL==1

        if (pClientStub->getSslImplementation() == Si_Schannel)
        {
            sslFilterPtr.reset( new SchannelFilter(pClientStub) );
        }
        else
        {
            RCF_ASSERT(pClientStub->getSslImplementation() == Si_OpenSsl);
            sslFilterPtr.reset( new OpenSslEncryptionFilter(pClientStub) );
        }

#elif RCF_FEATURE_SSPI==1

        sslFilterPtr.reset( new SchannelFilter(pClientStub) );

#elif RCF_FEATURE_OPENSSL==1

        sslFilterPtr.reset( new OpenSslEncryptionFilter(pClientStub) );

#endif

        if (!sslFilterPtr)
        {
            RCF_THROW( Exception(_RcfError_SslNotSupported()) );
        }

        wireFilters.push_back(sslFilterPtr);

        // HTTP CONNECT filter for passing through a proxy.
        wireFilters.push_back( FilterPtr( new HttpConnectFilter(
            getRemoteAddr().getIp(), 
            getRemoteAddr().getPort())));

        setWireFilters(wireFilters);
    }

    void ServerObjectService::customDeleter(const std::string & objectKey, void * pt)
    {
        RCF_UNUSED_VARIABLE(pt);

        Lock lock(mMutex);

        ServerObjectMap::iterator iter = mServerObjectMap.find(objectKey);
        RCF_ASSERT(iter != mServerObjectMap.end());
        ServerObjectHolder & holder = iter->second;
        RCF_ASSERT(holder.mUseCount > 0);
        --holder.mUseCount;
        holder.mLastTouchMs = getCurrentTimeMs();
    }

    bool AsioServerTransport::cycle(
        int timeoutMs,
        const volatile bool &,// stopFlag
        bool returnEarly)
    {
        RCF2_TRACE("Entering cycle()");

        RCF_ASSERT(timeoutMs >= -1)(timeoutMs);

        mInterrupt = returnEarly;
        if (timeoutMs != -1)
        {
            mCycleTimerPtr->mImpl.cancel();

            mCycleTimerPtr->mImpl.expires_from_now(
                boost::posix_time::milliseconds(timeoutMs));

            mCycleTimerPtr->mImpl.async_wait( boost::bind(
                &AsioServerTransport::stopCycle, 
                this, 
                boost::asio::placeholders::error));
        }

        mDemuxerPtr->reset();
        mDemuxerPtr->run();

        RCF2_TRACE("Exiting cycle()");

        return false;
    }

    void SubscriptionService::onServerStop(RcfServer &server)
    {
        RCF_UNUSED_VARIABLE(server);

        mPeriodicTimer.stop();

        Subscriptions subs;

        {
            Lock writeLock(mSubscriptionsMutex);
            subs = mSubscriptions;
        }

        for (Subscriptions::iterator iter = subs.begin();
            iter != subs.end();
            ++iter)
        {
            SubscriptionPtr subscriptionPtr = iter->lock();
            if (subscriptionPtr)
            {
                subscriptionPtr->close();
            }
        }

        {
            Lock writeLock(mSubscriptionsMutex);
            RCF_ASSERT(mSubscriptions.empty());
        }

        mSubscriptions.clear();
        subs.clear();

        mpServer = NULL;
    }

    void Win32NamedPipeSessionState::reconnect()
    {
        RCF_ASSERT(mEnableReconnect && mOwnFd);

        BOOL ok = DisconnectNamedPipe(mhPipe);
        DWORD dwErr = GetLastError();
        RCF_ASSERT(ok)(dwErr);
        
        mSessionPtr.reset();
        mSessionPtr = mTransport.mpSessionManager->createSession();
        mSessionPtr->setProactor(*this);

        resetState();

        accept();
    }

std::string getWorkingDir()
{
    std::vector<char> vec(1024);
    char * szRet = getcwd(&vec[0], static_cast<int>( vec.size() ));
    RCF_ASSERT(szRet);
    return std::string(&vec[0]);
}

    std::size_t Win32NamedPipeClientTransport::implWrite(
        const std::vector<ByteBuffer> &byteBuffers)
    {
        // For now, can't go back to sync calls after doing an async call.
        // Limitations with Windows IOCP.
        RCF_ASSERT(!mAsyncMode);

        // Not using overlapped I/O here because it interferes with the
        // server session that might be coupled to this transport.

        const ByteBuffer & byteBuffer = byteBuffers.front();

        DWORD count = 0;
        DWORD dwBytesToWrite = static_cast<DWORD>(byteBuffer.getLength());

        BOOL ok = WriteFile( 
            mhPipe,
            byteBuffer.getPtr(),
            dwBytesToWrite,
            &count,
            NULL);

        DWORD dwErr = GetLastError();

        RCF_VERIFY(ok, Exception(_RcfError_ClientWriteFail(), dwErr));

        // Strangely, WriteFile() sometimes returns 1, but at the same time a much too big value in count.
        RCF_VERIFY(count <= dwBytesToWrite, Exception(_RcfError_ClientWriteFail(), dwErr))(count)(dwBytesToWrite);

        RCF_VERIFY(count > 0, Exception(_RcfError_ClientWriteFail(), dwErr))(count)(dwBytesToWrite);

        onTimedSendCompleted( RCF_MIN(count, dwBytesToWrite), 0);

        return count;
    }

    void SubscriptionService::createSubscriptionImplBegin(
        RcfClientPtr rcfClientPtr, 
        const SubscriptionParms & parms,
        const std::string & defaultPublisherName)
    {
        ClientStub & clientStub = const_cast<ClientStub &>(parms.mClientStub);
        OnSubscriptionDisconnect onDisconnect = parms.mOnDisconnect;
        std::string publisherName = parms.mPublisherName;
        OnAsyncSubscribeCompleted onCompletion = parms.mOnAsyncSubscribeCompleted;

        if (publisherName.empty())
        {
            publisherName = defaultPublisherName;
        }
        
        RCF_ASSERT(onCompletion);

        if (clientStub.getRuntimeVersion() <= 11)
        {
            doRequestSubscriptionAsync_Legacy(
                clientStub, 
                publisherName, 
                rcfClientPtr,
                parms);
        }
        else
        {
            doRequestSubscriptionAsync(
                clientStub, 
                publisherName, 
                rcfClientPtr,
                parms);
        }
    }

 T & createSessionObject()
 {
     deleteSessionObject<T>();
     T * pt = getSessionObjectImpl<T>(true);
     RCF_ASSERT(pt);
     return *pt; 
 }

        void read(T &t)
        {
            try
            {
                switch (mProtocol)
                {
                case 1: mInProtocol1 >> t; break;
                case 2: mInProtocol2 >> t; break;
                case 3: mInProtocol3 >> t; break;
                case 4: mInProtocol4 >> t; break;

#ifdef RCF_USE_BOOST_XML_SERIALIZATION
                case 5: mInProtocol5 >> boost::serialization::make_nvp("Dummy", t); break;
#else
                case 5: mInProtocol5 >> t; break;
#endif

                default: RCF_ASSERT(0)(mProtocol);
                }
            }
            catch(const RCF::Exception &e)
            {
                RCF_UNUSED_VARIABLE(e);
                throw;
            }
            catch(const std::exception &e)
            {
                RCF::SerializationException se( _RcfError_Deserialization(
                    typeid(t).name(), 
                    typeid(e).name(), 
                    e.what()));

                RCF_THROW(se);
            }
        }

        T * getSessionObjectImpl(bool createIfDoesntExist)
        {
            typedef boost::shared_ptr<T> TPtr;

            const std::type_info & whichType = typeid(T);
            const std::type_info * pWhichType = &whichType;

            SessionObjectMap::iterator iter = mSessionObjects.find(pWhichType);
            if (iter != mSessionObjects.end())
            {
                boost::any & a = iter->second;
                TPtr * ptPtr = boost::any_cast<TPtr>(&a);
                RCF_ASSERT(ptPtr && *ptPtr);
                return ptPtr->get();
            }
            else if (createIfDoesntExist)
            {
                TPtr tPtr( new T() );
                mSessionObjects[pWhichType] = tPtr;
                return tPtr.get();
            }
            else
            {
                return NULL;
            }
        }

    void PublishingService::onServerStop(RcfServer &server)
    {
        RCF_UNUSED_VARIABLE(server);
        mPeriodicTimer.stop();

        // Close all publishers.

        Publishers publishers;
        {
            Lock writeLock(mPublishersMutex);
            publishers = mPublishers;
        }

        Publishers::iterator iter;
        for (iter = publishers.begin(); iter != publishers.end(); ++iter)
        {
            PublisherPtr publisherPtr = iter->second.lock();
            if (publisherPtr)
            {
                publisherPtr->close();
            }
        }

        {
            Lock writeLock(mPublishersMutex);
            RCF_ASSERT(mPublishers.empty());
        }
    }

 UInt32 IStream::read_int(UInt32 &n)
 {
     if (mRuntimeVersion < 9)
     {
         UInt32 bytesRead = read( reinterpret_cast<SF::Byte8 *>(&n), 4);
         RCF::networkToMachineOrder( &n, 4, 1);
         return bytesRead;
     }
     else
     {
         // Integers less than 128 are stored as a single byte.
         Byte8 byte = 0;
         boost::uint8_t ubyte = 0;
         UInt32 bytesRead = read_byte(byte);
         ubyte = byte;
         if (ubyte < 128)
         {
             n = ubyte;
         }
         else
         {
             RCF_ASSERT(ubyte == 128);
             bytesRead += read( reinterpret_cast<SF::Byte8 *>(&n), 4);
             RCF::networkToMachineOrder( &n, 4, 1);
         }
         return bytesRead;
     }
 }

        void read(T &t)
        {
            try
            {
                switch (mProtocol)
                {
                case 1: mInProtocol1 >> t; break;
                case 2: mInProtocol2 >> t; break;
                case 3: mInProtocol3 >> t; break;
                case 4: mInProtocol4 >> t; break;
                case 5: mInProtocol5 >> t; break;

                default: RCF_ASSERT(0)(mProtocol);
                }
            }
            catch(const RCF::Exception &e)
            {
                RCF_UNUSED_VARIABLE(e);
                throw;
            }
            catch(const std::exception &e)
            {
                RCF::SerializationException se( _RcfError_Deserialization(
                    typeid(t).name(), 
                    typeid(e).name(), 
                    e.what()));

                RCF_THROW(se);
            }
        }

    void CallbackConnectionService::CreateCallbackConnection()
    {
        // TODO: regular error message.
        // ...
        RCF_ASSERT( mOnCallbackConnectionCreated );

        RCF::convertRcfSessionToRcfClient( mOnCallbackConnectionCreated );
    }

 void InProcessTransport::setTimer(
     boost::uint32_t timeoutMs,
     I_ClientTransportCallback * pClientStub)
 {
     RCF_ASSERT(0);
     RCF_UNUSED_VARIABLE(timeoutMs);
     RCF_UNUSED_VARIABLE(pClientStub);
 }