yatejabber.h

 
#ifndef __YATEJABBER_H
#define __YATEJABBER_H
 
#include <xmpputils.h>

namespace TelEngine {
 
class SASL;                              // SASL authentication mechanism
class JBEvent;                           // A Jabber event
class JBStream;                          // A Jabber stream
class JBClientStream;                    // A client to server stream
class JBServerStream;                    // A server to server stream
class JBClusterStream;                   // A cluster stream
class JBRemoteDomainDef;                 // Options and connect settings for a remote domain
class JBConnect;                         // A socket connector
class JBEngine;                          // A Jabber engine
class JBServerEngine;                    // A Jabber server engine
class JBClientEngine;                    // A Jabber client engine
class JBStreamSet;                       // A set of streams to be processed in an uniform way
class JBStreamSetProcessor;              // Specialized stream processor
class JBStreamSetReceive;                // Specialized stream data receiver
class JBStreamSetList;                   // A list of stream sets
class JBEntityCaps;                      // Entity capability
class JBEntityCapsList;                  // Entity capability list manager
 

#define XMPP_C2S_PORT 5222

#define XMPP_S2S_PORT 5269

#define XMPP_MAX_INCOMPLETEXML 8192
 

class YJABBER_API SASL : public GenObject
{
    YCLASS(SASL,GenObject)
public:
    SASL(bool plain, const char* realm = 0);

    ~SASL()
        { TelEngine::destruct(m_params); }

    void setAuthParams(const char* user = 0, const char* pwd = 0);

    bool buildAuthRsp(String& buf, const char* digestUri = 0);

    inline void buildAuthRspReply(String& buf, const String& rsp) {
            if (m_plain)
                return;
            String tmp("rspauth=" + rsp);
            Base64 b((void*)tmp.c_str(),tmp.length(),false);
            b.encode(buf);
            b.clear(false);
        }

    inline bool validAuthReply(const String& reply) {
            String tmp;
            if (m_params)
                buildMD5Digest(tmp,m_params->getValue("password"),false);
            return tmp == reply;
        }

    bool buildMD5Challenge(String& buf);

    inline void buildMD5Digest(String& dest, const char* password,
        bool challengeRsp = true) {
            if (m_params)
                buildMD5Digest(dest,*m_params,password,challengeRsp);
        }

    bool parsePlain(const DataBlock& buf);

    bool parseMD5Challenge(const String& buf);

    bool parseMD5ChallengeRsp(const String& buf);

    static bool parsePlain(const DataBlock& buf, String& user, String& pwd,
        String* authzid = 0);

    static void buildMD5Digest(String& dest, const NamedList& params,
        const char* password, bool challengeRsp = true);
 
    bool m_plain;
    NamedList* m_params;
    String m_realm;
    String m_nonce;
    String m_cnonce;
    unsigned int m_nonceCount;
 
private:
    SASL() {}
};
 

class YJABBER_API JBEvent : public RefObject
{
    YCLASS(JBEvent,RefObject)
    friend class JBStream;
    friend class JBClientStream;
    friend class JBServerStream;
public:
    enum Type {
        // Stream terminated. Try to connect or wait to be destroyed
        Terminated,
        // Stream is destroying
        Destroy,
        // Stream start was received: when processing this event, the upper
        // layer must call stream's start() method or terminate the stream
        Start,
        // Incoming stream need auth: when processing this event, the upper
        // layer must call stream's authenticated() method
        // Component: the event's text contains the handshake data
        Auth,
        // The event's element is an 'iq' with a child qualified by bind namespace
        // This event is generated by an incoming client stream without a bound resource
        Bind,
        // Stream is running (can send/recv stanzas)
        Running,
        // The event's element is a 'message'
        Message,
        // The event's element is a 'presence'
        Presence,
        // The event's element is an 'iq'
        Iq,
        // The event's element is a 'db:result' one received by a server-to-server stream
        //  containing the dialback key to verify
        // The event's text is filled with dialback key to verify
        DbResult,
        // The event's element is a 'db:verify' one received by a server-to-server stream
        DbVerify,
        // New user register or user password change succeeded
        RegisterOk,
        // New user register or user password change failed
        // The event's element is the response
        RegisterFailed,
        // Non stanza element received in Running state
        Unknown
    };

    inline JBEvent(Type type, JBStream* stream, XmlElement* element,
        const JabberID& from, const JabberID& to, XmlElement* child = 0)
        : m_type(type), m_stream(0), m_link(true), m_element(element),
        m_child(child)
        { init(stream,element,&from,&to); }

    inline JBEvent(Type type, JBStream* stream, XmlElement* element,
        XmlElement* child = 0)
        : m_type(type), m_stream(0), m_link(true), m_element(element),
        m_child(child)
        { init(stream,element); }

    virtual ~JBEvent();

    inline int type() const
        { return m_type; }

    inline const char* name() const
        { return lookup(type()); }

    inline const String& stanzaType() const
        { return m_stanzaType; }

    inline const JabberID& from() const
        { return m_from; }

    inline const JabberID& to() const
        { return m_to; }

    inline const String& id() const
        { return m_id; }

    inline const String& text() const
        { return m_text; }

    inline JBStream* stream() const
        { return m_stream; }

    JBClientStream* clientStream();

    JBServerStream* serverStream();

    JBClusterStream* clusterStream();

    inline XmlElement* element() const
        { return m_element; }

    inline XmlElement* child() const
        { return m_child; }

    XmlElement* releaseXml(bool del = false);

    XmlElement* buildIqResult(bool addTags, XmlElement* child = 0);

    bool sendIqResult(XmlElement* child = 0);

    XmlElement* buildIqError(bool addTags, XMPPError::Type error, const char* reason = 0,
        XMPPError::ErrorType type = XMPPError::TypeModify);

    bool sendStanzaError(XMPPError::Type error, const char* reason = 0,
        XMPPError::ErrorType type = XMPPError::TypeModify);

    void releaseStream(bool release = false);

    inline static const char* lookup(int type)
        { return TelEngine::lookup(type,s_type); }
 
private:
    static const TokenDict s_type[];     // Event names
    JBEvent() {}                         // Don't use it!
    bool init(JBStream* stream, XmlElement* element,
        const JabberID* from = 0, const JabberID* to = 0);
 
    Type m_type;                         // Type of this event
    JBStream* m_stream;                  // The stream that generated this event
    bool m_link;                         // Stream link state
    XmlElement* m_element;               // Received XML element, if any
    XmlElement* m_child;                 // The first child element for 'iq' elements
    String m_stanzaType;                 // Stanza's 'type' attribute
    JabberID m_from;                     // Stanza's 'from' attribute
    JabberID m_to;                       // Stanza's 'to' attribute
    String m_id;                         // 'id' attribute if the received element has one
    String m_text;                       // The stanza's text or termination reason for
                                         //  Terminated/Destroy events
};
 

class YJABBER_API JBStream : public RefObject, public DebugEnabler, public Mutex
{
    friend class JBEngine;
    friend class JBEvent;
public:
    enum Type {
        c2s = 0,                         // Client to server
        s2s,                             // Server to server
        comp,                            // External component
        cluster,                         // Cluster stream
        TypeCount                        // Unknown
    };

    enum State {
        Idle = 0,                        // Stream is waiting to be connected or destroyed
        Connecting,                      // Outgoing stream is waiting for the socket to connect
        WaitStart,                       // Waiting for remote's stream start
                                         // (outgoing: stream start already sent)
        Starting,                        // Incoming stream is processing a stream start element
        Features,                        // Outgoing: waiting for stream features
                                         // Incoming: stream features sent
        WaitTlsRsp,                      // 'starttls' sent: waiting for response
        Securing,                        // Stream is currently negotiating the TLS
        Auth,                            // Auth element (db:result for s2s streams) sent
                                         // Incoming comp: handshake received
        Challenge,                       // 'challenge' element sent/received
        Compressing,                     // Stream is negotiating compression
                                         // outgoing: compress element sent, wait for response
                                         // incoming: waiting for <compressed> element to be sent
        Register,                        // A new user is currently registering
        // Keep Running state here: we expect all other states
        //  (except for Destroy) to have lower values
        Running,                         // Established. Allow XML stanzas to pass over the stream
        Destroy,                         // Stream is destroying. No more traffic allowed
    };

    enum Flags {
        NoAutoRestart       = 0x00000001,// Don't restart stream when down
        TlsRequired         = 0x00000002,// TLS is mandatory on this stream
        AllowPlainAuth      = 0x00000004,// Allow plain password authentication
                                         //  If not allowed and this is the only method
                                         //  offered by server the stream will be terminated
        DialbackOnly        = 0x00000008,// Outgoing s2s dialback stream
        RegisterUser        = 0x00000010,// Outgoing c2s register new user
        Compress            = 0x00000020,// Offer/handle compression
        InError             = 0x00000080,// The stream was terminated with error
        // Flags to be managed by the upper layer
        RosterRequested     = 0x00000100,// c2s: the roster was already requested
        AvailableResource   = 0x00000200,// c2s: available presence was sent/received
        PositivePriority    = 0x00000400,// c2s: the resource advertised by the client has priority >= 0
        // Internal flags (cleared when the stream is re-started)
        SetCompressed       = 0x00010000,// Set StreamCompressed flag after succesfully sending
                                         //  the current stream xml buffer
        StreamSecured       = 0x00020000,// TLS stage was done (possible without using TLS)
        StreamTls           = 0x00040000,// The stream is using TLS
        StreamAuthenticated = 0x00080000,// Stream already authenticated
        StreamRemoteVer1    = 0x00100000,// Remote party advertised RFC3920 version=1.0
        StreamLocalVer1     = 0x00200000,// Advertise RFC3920 version=1.0 on incoming streams
        StreamWaitBindRsp   = 0x01000000,// Outgoing c2s waiting for bind response
        StreamWaitSessRsp   = 0x02000000,// Outgoing c2s waiting for session response
        StreamWaitChallenge = 0x04000000,// Outgoing waiting for auth challenge
        StreamWaitChgRsp    = 0x08000000,// Outgoing waiting challenge response confirmation
        StreamRfc3920Chg    = 0x10000000,// Outgoing sent empty response to challenge with rspauth (RFC3920)
        StreamCompressed    = 0x20000000,// The stream is using compression
        StreamCanCompress   = 0x40000000,// Incoming s2s may still be compressed
        // Flag masks
        StreamFlags         = 0x000000ff,
        InternalFlags       = 0xffff0000,
    };

    virtual ~JBStream();

    inline int type() const
        { return m_type; }

    inline int xmlns() const
        { return m_xmlns; }

    inline State state() const
        { return m_state; }

    inline bool incoming() const
        { return m_incoming; }

    inline bool outgoing() const
        { return !m_incoming; }

    inline JBEngine* engine() const
        { return m_engine; }

    inline const char* name() const
        { return m_name; }

    inline const String& id() const
        { return m_id; }

    inline bool isId(const String& str) {
            Lock lock(this);
            return str == m_id;
        }

    inline const JabberID& local() const
        { return m_local; }

    inline void local(JabberID& jid) {
            Lock lock(this);
            jid = m_local;
        }

    inline void setLocal(const char* jid)
        { m_local.set(jid); }

    inline const JabberID& remote() const
        { return m_remote; }

    inline void remote(JabberID& jid) {
            Lock lock(this);
            jid = m_remote;
        }

    inline bool remoteAddr(SocketAddr& addr) {
            Lock lock(this);
            return m_socket && m_socket->getPeerName(addr);
        }

    inline bool localAddr(SocketAddr& addr) {
            Lock lock(this);
            return m_socket && m_socket->getSockName(addr);
        }

    inline int flags() const
        { return m_flags; }

    inline bool flag(int mask) const
        { return 0 != (m_flags & mask); }

    inline void setTlsRequired(bool set) {
            Lock lock(this);
            if (set)
                setFlags(TlsRequired);
            else
                resetFlags(TlsRequired);
        }

    bool haveData();

    void connectAddr(String& addr, int& port, String& localip, int& stat,
        ObjList& srvs, bool* isRedirect = 0) const;

    inline const String& serverHost() const
        { return m_serverHost ? m_serverHost : m_remote.domain(); }

    void setRosterRequested(bool ok);

    bool setAvailableResource(bool ok, bool positive = true);

    bool readSocket(char* buf, unsigned int len);

    virtual JBClientStream* clientStream()
        { return 0; }

    virtual JBServerStream* serverStream()
        { return 0; }

    virtual JBClusterStream* clusterStream()
        { return 0; }

    JBEvent* getEvent(u_int64_t time = Time::msecNow());

    bool sendStanza(XmlElement*& xml);

    bool sendStreamXml(State newState, XmlElement* first, XmlElement* second = 0,
        XmlElement* third = 0);

    void start(XMPPFeatureList* features = 0, XmlElement* caps = 0, bool useVer1 = true);

    bool authenticated(bool ok, const String& rsp = String::empty(),
        XMPPError::Type error = XMPPError::NotAuthorized,
        const char* username = 0, const char* id = 0, const char* resource = 0);

    void terminate(int location, bool destroy, XmlElement* xml,
        int error = XMPPError::NoError, const char* reason = "",
        bool final = false, bool genEvent = true, const char* content = 0);

    virtual void connectTerminated(Socket*& sock);

    virtual bool connecting(bool sync, int stat, ObjList& srvs);

    virtual void* getObject(const String& name) const;

    inline const char* stateName() const
        { return lookup(state(),s_stateName); }

    inline const char* typeName() const
        { return lookup(type(),s_typeName); }

    inline void buildSha1Digest(String& buf, const String& secret) {
            SHA1 sha(id() + secret);
            buf = sha.hexDigest();
            buf.toLower();
        }

    virtual const String& toString() const;

    static inline Type lookupType(const char* text, Type defVal = TypeCount)
        { return (Type)lookup(text,s_typeName,defVal); }

    SASL* m_sasl;

    static const TokenDict s_stateName[];

    static const TokenDict s_flagName[];

    static const TokenDict s_typeName[];
 
protected:
    JBStream(JBEngine* engine, Socket* socket, Type t, bool ssl = false);

    JBStream(JBEngine* engine, Type t, const JabberID& local, const JabberID& remote,
        const char* name = 0, const NamedList* params = 0, const char* serverHost = 0);

    virtual void destroyed();

    virtual bool canProcess(u_int64_t time);

    virtual void process(u_int64_t time);

    virtual bool processRunning(XmlElement* xml, const JabberID& from,
        const JabberID& to);

    virtual void checkTimeouts(u_int64_t time);

    virtual void resetConnection(Socket* sock = 0);

    virtual XmlElement* buildPing(const String& stanzaId);

    virtual XmlElement* buildStreamStart();

    virtual bool processStart(const XmlElement* xml, const JabberID& from,
        const JabberID& to);

    virtual bool processAuth(XmlElement* xml, const JabberID& from,
        const JabberID& to);

    virtual bool processCompressing(XmlElement* xml, const JabberID& from,
        const JabberID& to);

    virtual bool processRegister(XmlElement* xml, const JabberID& from,
        const JabberID& to);

    bool processStreamStart(const XmlElement* xml);

    bool handleCompressReq(XmlElement* xml);

    bool streamError(XmlElement* xml);

    bool getJids(XmlElement* xml, JabberID& from, JabberID& to);

    bool checkStanzaRecv(XmlElement* xml, JabberID& from, JabberID& to);

    void changeState(State newState, u_int64_t time = Time::msecNow());

    XmlElement* checkCompress();

    void checkPendingEvent();

    bool sendPending(bool streamOnly = false);

    bool writeSocket(const void* data, unsigned int& len);

    void updateFromRemoteDef();

    XMPPFeature* firstRequiredFeature();

    bool dropXml(XmlElement*& xml, const char* reason);

    inline bool destroyDropXml(XmlElement*& xml, XMPPError::Type error, const char* reason) {
            dropXml(xml,reason);
            terminate(0,true,0,error);
            return false;
        }

    void setFlags(int mask);

    void resetFlags(int mask);

    inline void setSecured() {
            setFlags(StreamSecured);
            m_features.remove(XMPPNamespace::Tls);
        }

    void setIdleTimer(u_int64_t msecNow = Time::msecNow());

    void resetPing();

    XmlElement* setNextPing(bool force);

    inline void generateIdIndex(String& buf, const char* extra = 0)
        { buf = id() + extra + String(++m_stanzaIndex); }
 
    State m_state;                       // Stream state
    String m_id;                         // Stream id
    JabberID m_local;                    // Local peer's jid
    JabberID m_remote;                   // Remote peer's jid
    String m_serverHost;                 // Outgoing: optional server host (replaces remote domain when connecting)
    int m_flags;                         // Stream flags
    XMPPNamespace::Type m_xmlns;         // Stream namespace
    XMPPFeatureList m_features;          // Advertised features
    JBEvent* m_lastEvent;                // Last event generated by this stream
    ObjList m_events;                    // Queued events
    ObjList m_pending;                   // Pending outgoing elements
    unsigned int m_stanzaIndex;          // Index used to generate IDs for stanzas
    // Timers
    u_int64_t m_setupTimeout;            // Overall stream setup timeout
    u_int64_t m_startTimeout;            // Incoming: wait stream start period
    u_int64_t m_pingTimeout;             // Sent ping timeout
    u_int64_t m_pingInterval;            // Ping interval
    u_int64_t m_nextPing;                // Next ping
    u_int64_t m_idleTimeout;             // Stream idle timeout
    u_int64_t m_connectTimeout;          // Stream connect timeout
    //
    unsigned int m_restart;              // Remaining restarts
    u_int64_t m_timeToFillRestart;       // The next time to increase the restart counter
 
    String m_pingId;
 
private:
    // Forbidden default constructor
    inline JBStream() {}
    // Process incoming elements in Challenge state
    // The element will be consumed
    // Return false if stream termination was initiated
    bool processChallenge(XmlElement* xml, const JabberID& from,
        const JabberID& to);
    // Process incoming 'auth' elements qualified by SASL namespace
    // The element will be consumed
    // Return false if stream termination was initiated
    bool processSaslAuth(XmlElement* xml, const JabberID& from,
        const JabberID& to);
    // Process received elements in Features state (incoming stream)
    // The element will be consumed
    // Return false if stream termination was initiated
    bool processFeaturesIn(XmlElement* xml, const JabberID& from,
        const JabberID& to);
    // Process received elements in Features state (outgoing stream)
    // The element will be consumed
    // Return false if stream termination was initiated
    bool processFeaturesOut(XmlElement* xml, const JabberID& from,
        const JabberID& to);
    // Process received elements in WaitTlsRsp state (outgoing stream)
    // The element will be consumed
    // Return false if stream termination was initiated
    bool processWaitTlsRsp(XmlElement* xml, const JabberID& from,
        const JabberID& to);
    // Set stream namespace from type
    void setXmlns();
    // Event termination notification
    // @param event The notifier. Ignored if it's not m_lastEvent
    void eventTerminated(const JBEvent* event);
    // Compress data to be sent (the pending stream xml buffer or pending stanza)
    // Return false on failure
    bool compress(XmlElementOut* xml = 0);
    // Reset connect status data
    void resetConnectStatus();
    // Postpone stream terminate until all parsed elements are processed
    // Terminate now if allowed
    // This method is thread safe
    void postponeTerminate(int location, bool destroy, int error, const char* reason);
    // Handle postponed termination. Return true if found
    // This method is not thread safe
    bool postponedTerminate();
    // Reset redirect data
    void setRedirect(const String& addr = String::empty(), int port = 0);
    // Reset postponed terminate data
    inline void resetPostponedTerminate() {
            m_ppTerminateTimeout = 0;
            TelEngine::destruct(m_ppTerminate);
        }
 
    enum {
        SocketCanRead = 0x01,
        SocketReading = 0x02,
        SocketCanWrite = 0x10,
        SocketWriting = 0x20,
        SocketWaitReset = 0x80,
    };
    inline void socketSetCanRead(bool ok) {
            Lock lock(m_socketMutex);
            if (ok)
                m_socketFlags |= SocketCanRead;
            else
                m_socketFlags &= ~SocketCanRead;
        }
    inline void socketSetReading(bool ok) {
            if (ok)
                m_socketFlags |= SocketReading;
            else
                m_socketFlags &= ~SocketReading;
        }
    inline void socketSetCanWrite(bool ok) {
            Lock lock(m_socketMutex);
            if (ok)
                m_socketFlags |= SocketCanWrite;
            else
                m_socketFlags &= ~SocketCanWrite;
        }
    inline void socketSetWriting(bool ok) {
            if (ok)
                m_socketFlags |= SocketWriting;
            else
                m_socketFlags &= ~SocketWriting;
        }
    inline bool socketCanRead() const {
            return m_socket && (m_socketFlags & SocketCanRead) &&
                !socketWaitReset();
        }
    inline bool socketCanWrite() const {
            return m_socket && (m_socketFlags & SocketCanWrite) &&
                !socketWaitReset();
        }
    inline bool socketReading() const
        { return (m_socketFlags & SocketReading) != 0; }
    inline bool socketWriting() const
        { return (m_socketFlags & SocketWriting) != 0; }
    inline bool socketWaitReset() const
        { return 0 != (m_socketFlags & SocketWaitReset); }
 
    JBEngine* m_engine;                  // The owner of this stream
    int m_type;                          // Stream type
    bool m_incoming;                     // Stream direction
    String m_name;                       // Local (internal) name
    JBEvent* m_terminateEvent;           // Pending terminate event
    NamedList* m_ppTerminate;            // Postponed terminate parameters
    u_int64_t m_ppTerminateTimeout;      // Postponed terminate timeout
    // Pending outgoing XML
    String m_outStreamXml;
    DataBlock m_outStreamXmlCompress;
    DataBlock m_outXmlCompress;
    // Connection related data
    XmlDomParser* m_xmlDom;
    Socket* m_socket;
    char m_socketFlags;                  // Socket flags: 0: unavailable
    Mutex m_socketMutex;                 // Protect the socket and parser
    String m_connectAddr;                // Remote ip to connect to
    int m_connectPort;                   // Remote port to connect to
    String m_localIp;                    // Local ip to bind when connecting
    Compressor* m_compress;
    int m_connectStatus;                 // Current connect stream status
    ObjList m_connectSrvs;               // Current connect stream SRV records
    unsigned int m_redirectMax;
    unsigned int m_redirectCount;
    String m_redirectAddr;
    int m_redirectPort;
};
 

class YJABBER_API JBClientStream : public JBStream
{
    YCLASS(JBClientStream,JBStream)
    friend class JBStream;
public:
    JBClientStream(JBEngine* engine, Socket* socket, bool ssl = false);

    JBClientStream(JBEngine* engine, const JabberID& jid, const String& account,
        const NamedList& params, const char* name = 0, const char* serverHost = 0);

    inline const String& account() const
        { return m_account; }

    inline GenObject* userData()
        { return m_userData; }

    inline void userData(GenObject* data) {
            Lock lock(this);
            TelEngine::destruct(m_userData);
            m_userData = data;
        }

    virtual JBClientStream* clientStream()
        { return this; }

    virtual XmlElement* buildPing(const String& stanzaId);

    void bind(const String& resource, const char* id,
        XMPPError::Type error = XMPPError::NoError);

    bool requestRegister(bool data, bool set = true,
        const String& newPass = String::empty());
 
protected:
    virtual bool processRunning(XmlElement* xml, const JabberID& from,
        const JabberID& to);

    virtual bool processStart(const XmlElement* xml, const JabberID& from,
        const JabberID& to);

    virtual bool processAuth(XmlElement* xml, const JabberID& from,
        const JabberID& to);

    virtual bool processRegister(XmlElement* xml, const JabberID& from,
        const JabberID& to);

    virtual void destroyed();

    bool startAuth();

    bool bind();
 
private:
    inline bool isRegisterId(XmlElement& xml) {
            if (!m_registerReq)
                return false;
            String* id = xml.getAttribute("id");
            return id && id->length() == 1 && (*id)[0] == m_registerReq;
        }
 
    String m_account;                    // Stream account
    GenObject* m_userData;               // User (upper layer) data
    String m_password;                   // The password
    String m_newPassword;                // New password
    char m_registerReq;                  // Register requested. 1(data) 2(register) 3(remove)
};
 

class YJABBER_API JBServerStream : public JBStream
{
    YCLASS(JBServerStream,JBStream)
    friend class JBStream;
public:
    JBServerStream(JBEngine* engine, Socket* socket, bool component = false);

    JBServerStream(JBEngine* engine, const JabberID& local, const JabberID& remote,
        const char* dbId = 0, const char* dbKey = 0, bool dbOnly = false,
        const NamedList* params = 0);

    JBServerStream(JBEngine* engine, const JabberID& local, const JabberID& remote,
        const String* name = 0, const NamedList* params = 0);

    inline bool dialback() const
        { return outgoing() && flag(DialbackOnly); }

    inline const NamedList& remoteDomains() const
        { return m_remoteDomains; }

    inline bool hasRemoteDomain(const String& domain, bool auth = true) {
            NamedString* tmp = m_remoteDomains.getParam(domain);
            return tmp && (!auth || tmp->null());
        }

    inline NamedString* takeDb() {
            Lock lock(this);
            NamedString* tmp = m_dbKey;
            m_dbKey = 0;
            return tmp;
        }

    virtual JBServerStream* serverStream()
        { return this; }

    bool sendDbVerify(const char* from, const char* to, const char* id,
        XMPPError::Type rsp = XMPPError::NoError);

    bool sendDbResult(const JabberID& from, const JabberID& to,
        XMPPError::Type rsp = XMPPError::NoError);

    bool sendDialback();

    bool startComp(const String& local = String::empty(), const String& remote = String::empty());
 
protected:
    virtual void destroyed();

    virtual bool processRunning(XmlElement* xml, const JabberID& from,
        const JabberID& to);

    virtual XmlElement* buildStreamStart();

    virtual bool processStart(const XmlElement* xml, const JabberID& from,
        const JabberID& to);

    virtual bool processAuth(XmlElement* xml, const JabberID& from,
        const JabberID& to);

    bool processDbResult(XmlElement* xml, const JabberID& from, const JabberID& to);

    inline void adjustDbRsp(XMPPError::Type& rsp) {
            Lock lock(this);
            if (!flag(StreamRemoteVer1) && rsp != XMPPError::NoError)
                rsp = XMPPError::NotAuthorized;
        }

    NamedList m_remoteDomains;
 
private:
    NamedString* m_dbKey;                // Outgoing: initial dialback key to check
    String m_password;                   // Outgoing component: password
};
 

class YJABBER_API JBClusterStream : public JBStream
{
    YCLASS(JBClusterStream,JBStream)
    friend class JBStream;
public:
    JBClusterStream(JBEngine* engine, Socket* socket);

    JBClusterStream(JBEngine* engine, const JabberID& local, const JabberID& remote,
        const NamedList* params = 0);

    virtual JBClusterStream* clusterStream()
        { return this; }
 
protected:
    virtual XmlElement* buildStreamStart();

    virtual bool processStart(const XmlElement* xml, const JabberID& from,
        const JabberID& to);

    virtual bool processRunning(XmlElement* xml, const JabberID& from,
        const JabberID& to);
};
 

class YJABBER_API JBRemoteDomainDef : public String
{
    YCLASS(JBRemoteDomainDef,String)
public:
    inline JBRemoteDomainDef(const char* domain = 0)
        : String(domain), m_port(0), m_flags(0)
        {}

    String m_address;

    int m_port;

    int m_flags;
};
 

class YJABBER_API JBConnect : public GenObject
{
    YCLASS(JBConnect,GenObject)
public:
    enum Status {
        Start = 0,
        Address,                         // Use configured address
        Srv,                             // Use SRV records
        Domain                           // Use stream remote domain
    };

    JBConnect(const JBStream& stream);

    virtual ~JBConnect();

    virtual void stopConnect();

    virtual const String& toString() const;

    static const TokenDict s_statusName[];
 
protected:
    void connect();
 
private:
    // No default constructor
    inline JBConnect()
        {}
    // Check if exiting. Release socket if exiting
    bool exiting(Socket*& sock);
    // Create and try to connect a socket. Return it on success
    // Set stop on fatal failure and return 0
    Socket* connect(const char* addr, int port, bool& stop);
    // Notify termination, remove from engine
    void terminated(Socket* sock, bool final);
    // Notify connecting to the stream. Return false if stream vanished
    bool notifyConnecting(bool sync, bool useCurrentStat = false);
    // Delete a socket and zero the pointer
    void deleteSocket(Socket*& sock);
    // Advance connect status
    void advanceStatus();
 
    int m_status;                        // Current status
    String m_domain;                     // Remote domain
    String m_address;                    // Remote ip address
    int m_port;                          // Port to connect to
    JBEngine* m_engine;                  // The engine owning this connector
    String m_stream;                     // Stream name
    JBStream::Type m_streamType;         // Stream type
    String m_localIp;                    // Local ip to bind when connecting
    ObjList m_srvs;                      // SRV records list
};
 

class YJABBER_API JBEngine : public DebugEnabler, public Mutex, public GenObject
{
    YCLASS(JBEngine,GenObject)
    friend class JBStream;
    friend class JBConnect;
    friend class JBStreamSetProcessor;
public:
    JBEngine(const char* name = "jbengine");

    virtual ~JBEngine();

    inline unsigned int streamReadBuffer() const
        { return m_streamReadBuffer; }

    inline bool exiting() const
        { return m_exiting; }

    inline void setExiting() {
            if (m_exiting)
                return;
            m_exiting = true;
            dropAll(JBStream::TypeCount,JabberID::empty(),JabberID::empty(),
                XMPPError::Shutdown);
        }

    inline unsigned int redirectMax() const
        { return m_redirectMax; }

    inline bool hasClientTls() const
        { return m_hasClientTls; }

    inline JBRemoteDomainDef* remoteDomainDef(const String& domain) {
            ObjList* o = m_remoteDomains.find(domain);
            return o ? static_cast<JBRemoteDomainDef*>(o->get()) : &m_remoteDomain;
        }

    virtual void destruct();

    virtual void initialize(const NamedList& params);

    virtual void cleanup(bool final = false, bool waitTerminate = true);

    bool acceptConn(Socket* sock, SocketAddr& remote, JBStream::Type t, bool ssl = false);

    virtual JBStream* findStream(const String& id,
        JBStream::Type hint = JBStream::TypeCount);

    ObjList* findClientStreams(bool in, const JabberID& jid, int flags = 0xffffffff);

    ObjList* findClientStreams(bool in, const JabberID& jid, const ObjList& resources,
        int flags = 0xffffffff);

    JBClientStream* findClientStream(bool in, const JabberID& jid);

    virtual unsigned int dropAll(JBStream::Type type = JBStream::TypeCount,
        const JabberID& local = JabberID::empty(),
        const JabberID& remote = JabberID::empty(),
        XMPPError::Type error = XMPPError::NoError, const char* reason = 0);

    virtual void buildStreamName(String& name, const JBStream* stream)
        {}

    virtual bool hasDomain(const String& domain)
        { return false; }

    virtual void processEvent(JBEvent* ev);

    virtual void returnEvent(JBEvent* ev, XMPPError::Type error = XMPPError::NoError,
        const char* reason = 0);

    virtual void encryptStream(JBStream* stream);

    virtual void connectStream(JBStream* stream);

    virtual void compressStream(JBStream* stream, const String& formats);

    virtual void buildDialbackKey(const String& id, const String& local,
        const String& remote, String& key);

    bool checkDupId(JBStream* stream);

    virtual void printXml(const JBStream* stream, bool send, XmlChild& xml) const;

    virtual void printXml(const JBStream* stream, bool send, XmlFragment& frag) const;
 
protected:
    virtual void addStream(JBStream* stream);

    virtual void removeStream(JBStream* stream, bool delObj = true);

    virtual void stopStreamSets(bool waitTerminate = true)
        {}

    virtual void getStreamList(RefPointer<JBStreamSetList>& list, int type)
        {}

    inline void getStreamLists(RefPointer<JBStreamSetList> list[JBStream::TypeCount],
        int type = JBStream::TypeCount) {
            if (type == JBStream::c2s || type == JBStream::TypeCount)
                getStreamList(list[JBStream::c2s],JBStream::c2s);
            if (type == JBStream::s2s || type == JBStream::TypeCount)
                getStreamList(list[JBStream::s2s],JBStream::s2s);
            if (type == JBStream::comp || type == JBStream::TypeCount)
                getStreamList(list[JBStream::comp],JBStream::comp);
            if (type == JBStream::cluster || type == JBStream::TypeCount)
                getStreamList(list[JBStream::cluster],JBStream::cluster);
        }

    JBStream* findStream(const String& id, JBStreamSetList* list);
 
    bool m_exiting;                      // Engine exiting flag
    JBRemoteDomainDef m_remoteDomain;    // Default remote domain definition
    ObjList m_remoteDomains;             // Remote domain definitions
    unsigned char m_restartMax;          // Maximum value for stream restart counter
    unsigned int m_restartUpdInterval;   // Update interval for stream restart counter
    unsigned int m_setupTimeout;         // Overall stream setup timeout
    unsigned int m_startTimeout;         // Wait stream start period
    unsigned int m_connectTimeout;       // Outgoing: socket connect timeout
    unsigned int m_srvTimeout;           // SRV query timeout
    unsigned int m_pingInterval;         // Stream idle interval (no data received)
    unsigned int m_pingTimeout;          // Sent ping timeout
    unsigned int m_idleTimeout;          // Stream idle timeout (nothing sent or received)
    unsigned int m_pptTimeoutC2s;        // Client streams postpone termination intervals
    unsigned int m_pptTimeout;           // Non client streams postpone stream termination intervals
    unsigned int m_streamReadBuffer;     // Stream read buffer length
    unsigned int m_maxIncompleteXml;     // Maximum length of an incomplete xml
    unsigned int m_redirectMax;          // Max redirect counter for outgoing streams
    bool m_hasClientTls;                 // True if TLS is available for outgoing streams
    int m_printXml;                      // Print XML data to output
    bool m_initialized;                  // True if already initialized
 
private:
    // Add/remove a connect stream thread when started/stopped
    void connectStatus(JBConnect* conn, bool started);
    // Stop a connect stream
    void stopConnect(const String& name);
 
    ObjList m_connect;                   // Connecting streams
};

class YJABBER_API JBServerEngine : public JBEngine
{
    YCLASS(JBServerEngine,JBEngine)
public:
    JBServerEngine(const char* name = "jbserverengine");

    ~JBServerEngine();

    virtual void cleanup(bool final = false, bool waitTerminate = true);

    virtual void buildStreamName(String& name, const JBStream* stream)
        { name << "stream/" << getStreamIndex(); }

    JBServerStream* findServerStream(const String& local, const String& remote, bool out,
        bool auth = true);

    JBServerStream* createServerStream(const String& local, const String& remote,
        const char* dbId = 0, const char* dbKey = 0, bool dbOnly = false,
        const NamedList* params = 0);

    JBServerStream* createCompStream(const String& name, const String& local, const String& remote,
        const NamedList* params = 0);

    JBClusterStream* findClusterStream(const String& remote, JBClusterStream* skip = 0);

    virtual JBClusterStream* createClusterStream(const String& local,
        const String& remote, const NamedList* params = 0);

    unsigned int terminateClientStreams(const JabberID& jid,
        XMPPError::Type error = XMPPError::NoError, const char* reason = 0);
 
protected:
    virtual void addStream(JBStream* stream);

    virtual void removeStream(JBStream* stream, bool delObj = true);

    virtual void stopStreamSets(bool waitTerminate = true);

    virtual void getStreamList(RefPointer<JBStreamSetList>& list, int type);

    virtual void getStreamListsType(int type, RefPointer<JBStreamSetList>& recv,
        RefPointer<JBStreamSetList>& process);

    inline unsigned int getStreamIndex() {
            Lock lock(this);
            return ++m_streamIndex;
        }
 
    unsigned int m_streamIndex;          // Index used to build stream name
    JBStreamSetList* m_c2sReceive;       // c2s streams receive list
    JBStreamSetList* m_c2sProcess;       // c2s streams process list
    JBStreamSetList* m_s2sReceive;       // s2s streams receive list
    JBStreamSetList* m_s2sProcess;       // s2s streams process list
    JBStreamSetList* m_compReceive;      // comp streams receive list
    JBStreamSetList* m_compProcess;      // comp streams process list
    JBStreamSetList* m_clusterReceive;   // cluster streams receive list
    JBStreamSetList* m_clusterProcess;   // cluster streams process list
};

class YJABBER_API JBClientEngine : public JBEngine
{
    YCLASS(JBClientEngine,JBEngine)
public:
    JBClientEngine(const char* name = "jbclientengine");

    ~JBClientEngine();

    virtual void cleanup(bool final = false, bool waitTerminate = true);

    JBClientStream* findAccount(const String& account);

    JBClientStream* create(const String& account, const NamedList& params,
        const String& name = String::empty());

    virtual void getStreamList(RefPointer<JBStreamSetList>& list, int type);
 
protected:
    virtual void addStream(JBStream* stream);

    virtual void removeStream(JBStream* stream, bool delObj = true);

    virtual void stopStreamSets(bool waitTerminate = true);
 
    JBStreamSetList* m_receive;          // Streams receive list
    JBStreamSetList* m_process;          // Streams process list
};

class YJABBER_API JBStreamSet : public GenObject, public Mutex
{
    YCLASS(JBStreamSet,GenObject);
    friend class JBStreamSetList;
public:
    virtual ~JBStreamSet();

    inline ObjList& clients()
        { return m_clients; }

    virtual bool add(JBStream* client);

    virtual bool remove(JBStream* client, bool delObj = true);

    unsigned int dropAll(const JabberID& local = JabberID::empty(),
        const JabberID& remote = JabberID::empty(),
        XMPPError::Type error = XMPPError::NoError, const char* reason = 0);

    void run();

    virtual bool start();

    virtual void stop();
 
protected:
    JBStreamSet(JBStreamSetList* owner);

    virtual bool process(JBStream& stream) = 0;
 
    bool m_changed;                      // List changed flag
    bool m_exiting;                      // The thread is exiting (don't accept clients)
    JBStreamSetList* m_owner;            // The list owning this set
    ObjList m_clients;                   // The streams list
 
private:
    JBStreamSet() {}                     // Private default constructor (forbidden)
};
 

class YJABBER_API JBStreamSetProcessor : public JBStreamSet
{
    YCLASS(JBStreamSetProcessor,JBStreamSet);
protected:
    inline JBStreamSetProcessor(JBStreamSetList* owner)
        : JBStreamSet(owner)
        {}

    virtual bool process(JBStream& stream);
};
 

class YJABBER_API JBStreamSetReceive : public JBStreamSet
{
    YCLASS(JBStreamSetReceive,JBStreamSet);
protected:
    JBStreamSetReceive(JBStreamSetList* owner);

    virtual bool process(JBStream& stream);
 
protected:
    DataBlock m_buffer;                  // Read buffer
};
 

class YJABBER_API JBStreamSetList : public RefObject, public Mutex
{
    YCLASS(JBStreamSetList,RefObject);
    friend class JBStreamSet;
public:
    JBStreamSetList(JBEngine* engine, unsigned int max, unsigned int sleepMs,
        const char* name);

    inline ObjList& sets()
        { return m_sets; }

    virtual ~JBStreamSetList();

    inline unsigned int maxStreams() const
        { return m_max; }

    inline unsigned int streamCount() const
        { return m_streamCount; }

    inline JBEngine* engine() const
        { return m_engine; }

    bool add(JBStream* client);

    void remove(JBStream* client, bool delObj = true);

    void stop(JBStreamSet* set = 0, bool waitTerminate = true);

    virtual const String& toString() const;
 
protected:
    virtual void destroyed();

    void remove(JBStreamSet* set);

    virtual JBStreamSet* build();
 
    JBEngine* m_engine;                  // The engine owning this list
    String m_name;                       // List name
    unsigned int m_max;                  // The maximum number of streams per set
    unsigned int m_sleepMs;              // Time to sleep if nothig processed
    ObjList m_sets;                      // The sets list
 
private:
    JBStreamSetList() {}                 // Private default constructor (forbidden)
 
    unsigned int m_streamCount;          // Current number of streams in this list
};
 

class YJABBER_API JBEntityCaps : public String
{
    YCLASS(JBEntityCaps,String);
public:
    enum {
        Ver1_3 = 1,  // Version lower then 1.4 (m_data is the node version + advertised extensions)
        Ver1_4 = 2,  // Version 1.4 or greater (m_data is the SHA-1 hash of features and identities)
    };

    inline JBEntityCaps(const char* id, char version, const char* node, const char* data)
        : String(id),
        m_version(version), m_node(node), m_data(data)
        {}

    inline bool hasFeature(int ns)
        { return 0 != m_features.get(ns); }

    inline bool hasAudio() {
            return hasFeature(XMPPNamespace::JingleAppsRtpAudio) ||
                hasFeature(XMPPNamespace::JingleAudio) ||
                hasFeature(XMPPNamespace::JingleVoiceV1);
        }

    static inline void buildId(String& buf, char version, const char* node,
        const char* data, String* ext = 0)
        { buf << (int)version << node << data << (ext ? ext->c_str() : ""); }
 
    char m_version;
    String m_node;
    String m_data;
    XMPPFeatureList m_features;
 
private:
    JBEntityCaps() {}
};
 

class YJABBER_API JBEntityCapsList : public ObjList, public Mutex
{
    YCLASS(JBEntityCapsList,ObjList);
public:
    inline JBEntityCapsList()
        : Mutex(true,"JBEntityCapsList"), m_enable(true), m_reqIndex(0)
        { m_reqPrefix << "xep0115" << (unsigned int)Time::msecNow() << "_"; }

    inline JBEntityCaps* findCaps(const String& id) {
            for (ObjList* o = skipNull(); o; o = o->skipNext())
                if (o->get()->toString() == id)
                    return static_cast<JBEntityCaps*>(o->get());
            return 0;
        }

    void expire(u_int64_t msecNow = Time::msecNow());

    bool processRsp(XmlElement* rsp, const String& id, bool ok);

    void requestCaps(JBStream* stream, const char* from, const char* to, const String& id,
        char version, const char* node, const char* data);

    XmlDocument* toDocument(const char* rootName = "entitycaps");

    void fromDocument(XmlDocument& doc, const char* rootName = "entitycaps");

    virtual bool processCaps(String& capsId, XmlElement* xml, JBStream* stream,
        const char* from, const char* to);

    inline void addCaps(NamedList& list, const String& id) {
            Lock lock(this);
            JBEntityCaps* caps = findCaps(id);
            if (caps)
                addCaps(list,*caps);
        }

    virtual void addCaps(NamedList& list, JBEntityCaps& caps);

    bool loadXmlDoc(const char* file, DebugEnabler* enabler = 0);

    bool saveXmlDoc(const char* file, DebugEnabler* enabler = 0);

    static bool decodeCaps(const XmlElement& xml, char& version, String*& node,
        String*& ver, String*& ext);

    bool m_enable;
 
protected:
    virtual void capsAdded(JBEntityCaps* caps)
        {}
 
    unsigned int m_reqIndex;             // Disco info request index
    String m_reqPrefix;                  // Prefix for disco info stanza id
    ObjList m_requests;                  // List of sent disco info requests
};
 
}; // namespace TelEngine
 
#endif /* __YATEJABBER_H */
 
/* vi: set ts=8 sw=4 sts=4 noet: */