/Users/baford/proj/netsteria/sst/lib/stream.h

Go to the documentation of this file.

#ifndef SST_STREAM_H
#define SST_STREAM_H

#include <QSet>
#include <QQueue>
#include <QIODevice>

#include "key.h"
#include "strm/proto.h"

namespace SST {

class Host;
class Flow;
class Ident;
class AbstractStream;
class BaseStream;
class StreamPeer;
class Endpoint;
class RegInfo;
class RegClient;


class Stream : public QIODevice
{
        friend class AbstractStream;
        friend class BaseStream;
        friend class StreamServer;
        Q_OBJECT

public:
        enum ShutdownModeFlag {
                Read    = 1,    // Read (incoming data) direction
                Write   = 2,    // Write (outgoing data) direction
                Close   = 3,    // Both directions (Read|Write)
                Reset   = 4,    // Forceful reset
        };
        Q_DECLARE_FLAGS(ShutdownMode, ShutdownModeFlag)

private:
        Host            *host;          // Per-host SST state
        AbstractStream  *as;            // Internal stream control object


public:
        Stream(Host *host, QObject *parent = NULL);
        virtual ~Stream();



        bool connectTo(const QByteArray &dstid,
                        const QString &service, const QString &protocol,
                        const Endpoint &dstep = Endpoint());

        bool connectTo(const Ident &dstid,
                        const QString &service, const QString &protocol,
                        const Endpoint &dstep = Endpoint());

        void disconnect();

        bool isConnected();



        qint64 bytesAvailable() const;

        inline bool hasBytesAvailable() const { return bytesAvailable() > 0; }

        virtual qint64 readData(char *data, qint64 maxSize);


        int pendingMessages() const;

        inline bool hasPendingMessages() const
                { return pendingMessages() > 0; }

        qint64 pendingMessageSize() const;

        qint64 readMessage(char *data, int maxSize);

        QByteArray readMessage(int maxSize = 1 << 30);


        bool atEnd() const;



        qint64 writeData(const char *data, qint64 size);

        qint64 writeMessage(const char *data, qint64 size);

        inline qint64 writeMessage(const QByteArray &msg)
                { return writeMessage(msg.data(), msg.size()); }

        // Send and receive unordered, unreliable datagrams on this stream.
        int readDatagram(char *data, int maxSize);
        QByteArray readDatagram(int maxSize = 1 << 30);
        int writeDatagram(const char *data, int size);
        inline int writeDatagram(const QByteArray &dgram)
                { return writeDatagram(dgram.data(), dgram.size()); }


        // Check for pending datagrams
        bool hasPendingDatagrams() const;
        qint32 pendingDatagramSize() const;

        // XX bytesToWrite()



        Stream *openSubstream();

        void listen();

        bool isListening() const;

        Stream *acceptSubstream();



        QByteArray localHostId();

        QByteArray remoteHostId();

        bool isLinkUp();

        void setPriority(int pri);

        int priority();

        // void setReceiveWindow(int size);

        // inline int receiveWindow();

        void shutdown(ShutdownMode mode);

        inline void close() { shutdown(Close); }


#ifndef QT_NO_DEBUG

        void dump();
#endif


signals:

        void readyReadMessage();

#if 0   // XXX not sure if this actually useful, maybe just readyReadMessage...

        void readyReadComplete();
#endif

        void newSubstream();

        void readyReadDatagram();

        void linkUp();

        //void linkStalled();   XX implement

        void linkDown();

        void receiveBlocked();

        void reset(const QString &errorString);

        void error(const QString &errorString);


protected:
        // Set an error condition on this Stream and emit the error() signal.
        void setError(const QString &errorString);

private:
        // Private constructor used internally
        // to create a Stream wrapper for an existing BaseStream.
        Stream(AbstractStream *as, QObject *parent);
};

Q_DECLARE_OPERATORS_FOR_FLAGS(Stream::ShutdownMode)



class StreamServer : public QObject, public StreamProtocol
{
        friend class BaseStream;
        Q_OBJECT

private:
        Host *const h;                  // Our per-host state
        QQueue<BaseStream*> rconns;     // Received connection stream queue
        QString svname;                 // Service name
        QString svdesc;                 // Longer service description
        QString prname;                 // Protocol name
        QString prdesc;                 // Longer protocol description
        QString err;
        bool active;


public:
        StreamServer(Host *host, QObject *parent = NULL);

        bool listen(const QString &serviceName, const QString &serviceDesc,
                const QString &protocolName, const QString &protocolDesc);

        inline bool isListening() { return active; }

        Stream *accept();

        inline Stream *accept(QByteArray &originHostId) {
                Stream *strm = accept();
                if (strm) originHostId = strm->remoteHostId();
                return strm; }

        inline QString serviceName() { return svname; }

        inline QString serviceDescription() { return svdesc; }

        inline QString protocolName() { return prname; }

        inline QString protocolDescription() { return prdesc; }

        inline QString errorString() { return err; }

signals:
        void newConnection();

protected:
        inline void setErrorString(const QString &err) { this->err = err; }
};


// Private helper class,
// to register with Socket layer to receive key exchange packets.
// Only one instance ever created per host.
class StreamResponder : public KeyResponder, public StreamProtocol
{
        friend class StreamPeer;
        friend class StreamServer;
        friend class StreamHostState;
        Q_OBJECT

        StreamResponder(Host *h);

        // Set of RegClients we've connected to so far
        QPointerSet<RegClient> connrcs;


        void conncli(RegClient *cli);

        virtual Flow *newFlow(const SocketEndpoint &epi, const QByteArray &idi,
                                const QByteArray &ulpi, QByteArray &ulpr);

private slots:
        void clientCreate(RegClient *rc);
        void clientStateChanged();
        void lookupNotify(const QByteArray &id, const Endpoint &loc,
                        const RegInfo &info);
};


// Per-host state for the Stream module.
class StreamHostState : public QObject
{
        friend class BaseStream;
        friend class StreamServer;
        friend class StreamPeer;
        friend class StreamResponder;
        Q_OBJECT

private:
        StreamResponder *rpndr;
        QHash<StreamProtocol::ServicePair,StreamServer*> listeners;
        QHash<QByteArray,StreamPeer*> peers;


        StreamResponder *streamResponder();

public:
        inline StreamHostState() : rpndr(NULL) { }
        virtual ~StreamHostState();

        StreamPeer *streamPeer(const QByteArray &id);

        virtual Host *host() = 0;
};

} // namespace SST

#endif  // SST_STREAM_H

---