libcore/plugins/tcpsst/TCPStream.hpp

Go to the documentation of this file.

/*  Sirikata Network Utilities
 *  TCPStream.hpp
 *
 *  Copyright (c) 2009, Daniel Reiter Horn
 *  All rights reserved.
 *
 *  Redistribution and use in source and binary forms, with or without
 *  modification, are permitted provided that the following conditions are
 *  met:
 *  * Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 *  * Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in
 *    the documentation and/or other materials provided with the
 *    distribution.
 *  * Neither the name of Sirikata nor the names of its contributors may
 *    be used to endorse or promote products derived from this software
 *    without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
 * IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
 * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
 * PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER
 * OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
 * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
 * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
#ifndef SIRIKATA_TCPStream_HPP__
#define SIRIKATA_TCPStream_HPP__

#include <sirikata/core/network/Stream.hpp>
#include <sirikata/core/util/AtomicTypes.hpp>
#include "TCPSSTDecls.hpp"

namespace Sirikata {
namespace Network {

class TCPSetCallbacks;
class IOStrand;

class TCPStream:public Stream {
public:
    class Callbacks;
    static const char * STRING_PREFIX() {
        return "SSTTCP";
    }
    static const char * WEBSOCKET_STRING_PREFIX() {
        return "SST:WS";
    }
    enum HeaderSizeEnumerant {
        STRING_PREFIX_LENGTH=6,
        TcpSstHeaderSize=24,
        MaxWebSocketHeaderSize=2048
    };
    enum TCPStreamControlCodes {
        TCPStreamCloseStream=1,
        TCPStreamAckCloseStream=2
    };
    enum StreamType{
        UNKNOWN,
        BASE64_ZERODELIM,
        LENGTH_DELIM,
        RFC_6455
    };
    //if !=0 and type is RFC_6455, arbitrarily fragment packets 2 indicates more aggressive testing of fragmentation than 1 (testing option)
    static int sFragmentPackets;
private:
    friend class MultiplexedSocket;
    friend class TCPSetCallbacks;
    IOStrand* mIOStrand;
    MultiplexedSocketPtr mSocket;
    void addCallbacks(Callbacks*);
    StreamID mID;
    enum {
        SendStatusClosing=(1<<29)
    };
    std::tr1::shared_ptr<AtomicValue<int> >mSendStatus;
    unsigned char mNumSimultaneousSockets;
    bool mNoDelay;
    StreamType mStreamType;
    unsigned int mSendBufferSize;
    unsigned int mKernelSendBufferSize;
    unsigned int mKernelReceiveBufferSize;

    TCPStream(IOStrand*,unsigned char mNumSimultaneousSockets, unsigned int mSendBufferSize, bool noDelay, StreamType streamType, unsigned int kernelSendBufferSize, unsigned int kernelReceiveBufferSize);


public:
    static bool closeSendStatus(AtomicValue<int>&vSendStatus);
    StreamID getID()const {return mID;}
    class Callbacks:public Noncopyable {
    public:
        Stream::ConnectionCallback mConnectionCallback;
        Stream::ReceivedCallback mBytesReceivedCallback;
        Stream::ReadySendCallback mReadySendCallback;
        std::tr1::weak_ptr<AtomicValue<int> > mSendStatus;
        Callbacks(const Stream::ConnectionCallback &connectionCallback,
                  const Stream::ReceivedCallback &bytesReceivedCallback,
                  const Stream::ReadySendCallback &readySendCallback,
                  const std::tr1::weak_ptr<AtomicValue<int> >&sendStatus):
            mConnectionCallback(connectionCallback),
            mBytesReceivedCallback(bytesReceivedCallback),
            mReadySendCallback(readySendCallback),
            mSendStatus(sendStatus){
        }
    };
    TCPStream(IOStrand*, OptionSet*);
    TCPStream(const MultiplexedSocketPtr &shared_socket, const Stream::StreamID&);
    virtual Stream*factory();
    virtual void readyRead();
    virtual void requestReadySendCallback();
    WARN_UNUSED
    virtual bool send(MemoryReference, StreamReliability);
    WARN_UNUSED
    virtual bool send(MemoryReference, MemoryReference, StreamReliability);
    WARN_UNUSED
    virtual bool send(const Chunk&data,StreamReliability);
    virtual bool canSend(size_t dataSize)const;
    virtual void connect(
        const Address& addy,
        const SubstreamCallback &substreamCallback,
        const ConnectionCallback &connectionCallback,
        const ReceivedCallback&chunkReceivedCallback,
        const ReadySendCallback&readySend);

    static TCPStream* construct(Network::IOStrand*io, OptionSet*options) {
        return new TCPStream(io,options);
    }
    virtual Stream* clone(const SubstreamCallback&cb);
    virtual Stream* clone(const ConnectionCallback &connectionCallback,
                          const ReceivedCallback&chunkReceivedCallback,
                          const ReadySendCallback&readySendCallback);

    Address getRemoteEndpoint() const;
    Address getLocalEndpoint() const;


    //Shuts down the socket, allowing StreamID to be reused and opposing stream to get disconnection callback
    virtual void close();
    ~TCPStream();

    virtual Duration averageSendLatency() const;
    virtual Duration averageReceiveLatency() const;
};

} // namespace Network
} // namespace Sirikata

#endif //SIRIKATA_TCPStream_HPP__