xref: /dports/net/yate/yate-6.4.0-1/libs/yjabber/jbstream.cpp

/**
 * jbstream.cpp
 * Yet Another Jabber Component Protocol Stack
 * This file is part of the YATE Project http://YATE.null.ro
 *
 * Yet Another Telephony Engine - a fully featured software PBX and IVR
 * Copyright (C) 2004-2014 Null Team
 *
 * This software is distributed under multiple licenses;
 * see the COPYING file in the main directory for licensing
 * information for this specific distribution.
 *
 * This use of this software may be subject to additional restrictions.
 * See the LEGAL file in the main directory for details.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
 */

#include <yatejabber.h>
#include <stdlib.h>

using namespace TelEngine;

#ifdef XDEBUG
  #define JBSTREAM_DEBUG_COMPRESS
  #define JBSTREAM_DEBUG_SOCKET
#else
//  #define JBSTREAM_DEBUG_COMPRESS                 // Show (de)compress debug
//  #define JBSTREAM_DEBUG_SOCKET                   // Show socket read/write debug
#endif

static const String s_dbVerify = "verify";
static const String s_dbResult = "result";

static inline bool isDbVerify(XmlElement& xml)
{
    const String* tag = 0;
    const String* ns = 0;
    return xml.getTag(tag,ns) && *tag == s_dbVerify &&
	ns && *ns == XMPPUtils::s_ns[XMPPNamespace::Dialback];
}

static inline bool isDbResult(XmlElement& xml)
{
    const String* tag = 0;
    const String* ns = 0;
    return xml.getTag(tag,ns) && *tag == s_dbResult &&
	ns && *ns == XMPPUtils::s_ns[XMPPNamespace::Dialback];
}

// Decode a Base64 string to a block
static inline bool decodeBase64(DataBlock& buf, const String& str)
{
    Base64 b((void*)str.c_str(),str.length(),false);
    bool ok = b.decode(buf,false);
    b.clear(false);
    return ok;
}

// Decode a Base64 string to another string
// Check if decoded data has valid UTF8 characters
static bool decodeBase64(String& buf, const String& str, JBStream* stream)
{
    DataBlock d;
    if (!decodeBase64(d,str))
	return false;
    buf.assign((const char*)d.data(),d.length());
    if (-1 != buf.lenUtf8())
	return true;
    Debug(stream,DebugNote,"Received Base64 encoded invalid UTF8 characters [%p]",stream);
    return false;
}

#ifdef DEBUG
static bool checkPing(JBStream* stream, const XmlElement* xml, const String& pingId)
{
    if (!(stream && xml && pingId))
	return false;
    if (pingId != xml->getAttribute(YSTRING("id")))
	return false;
    const char* it = xml->attribute(YSTRING("type"));
    XMPPUtils::IqType iqType = XMPPUtils::iqType(it);
    bool ok = (iqType == XMPPUtils::IqResult || iqType == XMPPUtils::IqError);
    if (ok)
	Debug(stream,DebugAll,"Ping with id=%s confirmed by '%s' [%p]",pingId.c_str(),it,stream);
    return ok;
}
#else
static inline bool checkPing(JBStream* stream, const XmlElement* xml, const String& pingId)
{
    return false;
}
#endif

static const TokenDict s_location[] = {
    {"internal",     0},
    {"remote",       1},
    {"local",       -1},
    {0,0},
};

const TokenDict JBStream::s_stateName[] = {
    {"Running",          Running},
    {"Idle",             Idle},
    {"Connecting",       Connecting},
    {"WaitStart",        WaitStart},
    {"Starting",         Starting},
    {"Features",         Features},
    {"WaitTlsRsp",       WaitTlsRsp},
    {"Auth",             Auth},
    {"Challenge",        Challenge},
    {"Securing",         Securing},
    {"Compressing",      Compressing},
    {"Register",         Register},
    {"Destroy",          Destroy},
    {0,0},
};

const TokenDict JBStream::s_flagName[] = {
    {"noautorestart",    NoAutoRestart},
    {"tlsrequired",      TlsRequired},
    {"dialback",         DialbackOnly},
    {"allowplainauth",   AllowPlainAuth},
    {"register",         RegisterUser},
    {"compress",         Compress},
    {"error",            InError},
    // Internal flags
    {"roster_requested", RosterRequested},
    {"online",           AvailableResource},
    {"secured",          StreamTls | StreamSecured},
    {"encrypted",        StreamTls},
    {"authenticated",    StreamAuthenticated},
    {"waitbindrsp",      StreamWaitBindRsp},
    {"waitsessrsp",      StreamWaitSessRsp},
    {"waitchallenge",    StreamWaitChallenge},
    {"waitchallengersp", StreamWaitChgRsp},
    {"version1",         StreamRemoteVer1},
    {"compressed",       StreamCompressed},
    {"cancompress",      StreamCanCompress},
    {0,0}
};

const TokenDict JBStream::s_typeName[] = {
    {"c2s",      c2s},
    {"s2s",      s2s},
    {"comp",     comp},
    {"cluster",  cluster},
    {0,0}
};

// Retrieve the multiplier for non client stream timers
static inline unsigned int timerMultiplier(JBStream* stream)
{
    return stream->type() == JBStream::c2s ? 1 : 2;
}


/*
 * JBStream
 */
// Incoming
JBStream::JBStream(JBEngine* engine, Socket* socket, Type t, bool ssl)
    : Mutex(true,"JBStream"),
    m_sasl(0),
    m_state(Idle), m_flags(0), m_xmlns(XMPPNamespace::Count), m_lastEvent(0),
    m_setupTimeout(0), m_startTimeout(0),
    m_pingTimeout(0), m_pingInterval(0), m_nextPing(0),
    m_idleTimeout(0), m_connectTimeout(0),
    m_restart(0), m_timeToFillRestart(0),
    m_engine(engine), m_type(t),
    m_incoming(true), m_terminateEvent(0), m_ppTerminate(0), m_ppTerminateTimeout(0),
    m_xmlDom(0), m_socket(0), m_socketFlags(0), m_socketMutex(true,"JBStream::Socket"),
    m_connectPort(0), m_compress(0), m_connectStatus(JBConnect::Start),
    m_redirectMax(0), m_redirectCount(0), m_redirectPort(0)
{
    if (ssl)
	setFlags(StreamSecured | StreamTls);
    m_engine->buildStreamName(m_name,this);
    debugName(m_name);
    debugChain(m_engine);
    Debug(this,DebugAll,"JBStream::JBStream(%p,%p,%s,%s) incoming [%p]",
	engine,socket,typeName(),String::boolText(ssl),this);
    setXmlns();
    // Don't restart incoming streams
    setFlags(NoAutoRestart);
    resetConnection(socket);
    changeState(WaitStart);
}

// Outgoing
JBStream::JBStream(JBEngine* engine, Type t, const JabberID& local, const JabberID& remote,
    const char* name, const NamedList* params, const char* serverHost)
    : Mutex(true,"JBStream"),
    m_sasl(0),
    m_state(Idle), m_local(local), m_remote(remote), m_serverHost(serverHost),
    m_flags(0), m_xmlns(XMPPNamespace::Count), m_lastEvent(0), m_stanzaIndex(0),
    m_setupTimeout(0), m_startTimeout(0),
    m_pingTimeout(0), m_nextPing(0),
    m_idleTimeout(0), m_connectTimeout(0),
    m_restart(1), m_timeToFillRestart(0),
    m_engine(engine), m_type(t),
    m_incoming(false), m_name(name),
    m_terminateEvent(0), m_ppTerminate(0), m_ppTerminateTimeout(0),
    m_xmlDom(0), m_socket(0), m_socketFlags(0), m_socketMutex(true,"JBStream::Socket"),
    m_connectPort(0), m_compress(0), m_connectStatus(JBConnect::Start),
    m_redirectMax(engine->redirectMax()), m_redirectCount(0), m_redirectPort(0)
{
    if (!m_name)
	m_engine->buildStreamName(m_name,this);
    debugName(m_name);
    debugChain(m_engine);
    if (params) {
	int flgs = XMPPUtils::decodeFlags(params->getValue("options"),s_flagName);
	setFlags(flgs & StreamFlags);
	m_connectAddr = params->getValue("server",params->getValue("address"));
	m_connectPort = params->getIntValue("port");
	m_localIp = params->getValue("localip");
    }
    else
	updateFromRemoteDef();
    // Compress always defaults to true if not explicitly disabled
    if (!flag(Compress) && !(params && params->getBoolValue("nocompression")))
	setFlags(Compress);
    Debug(this,DebugAll,"JBStream::JBStream(%p,%s,%s,%s,%s) outgoing [%p]",
	engine,typeName(),local.c_str(),remote.c_str(),m_serverHost.safe(),this);
    setXmlns();
    changeState(Idle);
}

// Destructor
JBStream::~JBStream()
{
    DDebug(this,DebugAll,"JBStream::~JBStream() id=%s [%p]",m_name.c_str(),this);
    TelEngine::destruct(m_sasl);
}

// Outgoing stream connect terminated notification.
void JBStream::connectTerminated(Socket*& sock)
{
    Lock lock(this);
    if (m_state == Connecting) {
	if (sock) {
	    resetConnection(sock);
	    sock = 0;
	    changeState(Starting);
	    XmlElement* s = buildStreamStart();
	    sendStreamXml(WaitStart,s);
	}
	else {
	    DDebug(this,DebugNote,"Connect failed [%p]",this);
	    resetConnectStatus();
	    setRedirect();
	    m_redirectCount = 0;
	    terminate(0,false,0,XMPPError::NoRemote);
	}
	return;
    }
    DDebug(this,DebugInfo,"Connect terminated notification in non %s state [%p]",
	lookup(Connecting,s_stateName),this);
    if (sock) {
	delete sock;
	sock = 0;
    }
}

// Connecting notification. Start connect timer for synchronous connect
bool JBStream::connecting(bool sync, int stat, ObjList& srvs)
{
    if (incoming() || !m_engine || state() != Connecting)
	return false;
    Lock lock(this);
    if (state() != Connecting)
	return false;
    m_connectStatus = stat;
    SrvRecord::copy(m_connectSrvs,srvs);
    if (sync) {
	if (stat != JBConnect::Srv)
	    m_connectTimeout = Time::msecNow() + m_engine->m_connectTimeout;
	else
	    m_connectTimeout = Time::msecNow() + m_engine->m_srvTimeout;
    }
    else
	m_connectTimeout = 0;
    DDebug(this,DebugAll,"Connecting sync=%u stat=%s [%p]",
	sync,lookup(m_connectStatus,JBConnect::s_statusName),this);
    return true;
}

// Get an object from this stream
void* JBStream::getObject(const String& name) const
{
    if (name == "Socket*")
	return state() == Securing ? (void*)&m_socket : 0;
    if (name == "Compressor*")
	return (void*)&m_compress;
    if (name == "JBStream")
	return (void*)this;
    return RefObject::getObject(name);
}

// Get the string representation of this stream
const String& JBStream::toString() const
{
    return m_name;
}

// Check if the stream has valid pending data
bool JBStream::haveData()
{
    Lock2 lck(this,&m_socketMutex);
    // Pending data with socket available for writing
    if (m_pending.skipNull() && socketCanWrite())
	return true;
    // Pending events
    if (m_events.skipNull())
	return true;
    // Pending incoming XML
    XmlDocument* doc = m_xmlDom ? m_xmlDom->document() : 0;
    XmlElement* root = doc ? doc->root(false) : 0;
    XmlElement* first = root ? root->findFirstChild() : 0;
    return first && first->completed();
}

// Retrieve connection address(es), port and status
void JBStream::connectAddr(String& addr, int& port, String& localip, int& stat,
    ObjList& srvs, bool* isRedirect) const
{
    if (m_redirectAddr) {
	addr = m_redirectAddr;
	port = m_redirectPort;
    }
    else {
	addr = m_connectAddr;
	port = m_connectPort;
    }
    if (isRedirect)
	*isRedirect = !m_redirectAddr.null();
    localip = m_localIp;
    stat = m_connectStatus;
    SrvRecord::copy(srvs,m_connectSrvs);
}

// Set/reset RosterRequested flag
void JBStream::setRosterRequested(bool ok)
{
    Lock lock(this);
    if (ok == flag(RosterRequested))
	return;
    if (ok)
	setFlags(RosterRequested);
    else
	resetFlags(RosterRequested);
    XDebug(this,DebugAll,"%s roster requested flag [%p]",ok ? "Set" : "Reset",this);
}

// Set/reset AvailableResource/PositivePriority flags
bool JBStream::setAvailableResource(bool ok, bool positive)
{
    Lock lock(this);
    if (ok && positive)
	setFlags(PositivePriority);
    else
	resetFlags(PositivePriority);
    if (ok == flag(AvailableResource))
	return false;
    if (ok)
	setFlags(AvailableResource);
    else
	resetFlags(AvailableResource);
    XDebug(this,DebugAll,"%s available resource flag [%p]",ok ? "Set" : "Reset",this);
    return true;
}

// Read data from socket. Send it to the parser
bool JBStream::readSocket(char* buf, unsigned int len)
{
    if (!(buf && len > 1))
	return false;
    if (!socketCanRead())
	return false;
    Lock2 lock(*this,m_socketMutex);
    if (!socketCanRead() || state() == Destroy || state() == Idle || state() == Connecting)
	return false;
    socketSetReading(true);
    if (state() != WaitTlsRsp)
	len--;
    else
	len = 1;
    lock.drop();
    // Check stream state
    XMPPError::Type error = XMPPError::NoError;
    int read = m_socket->readData(buf,len);
    Lock lck(m_socketMutex);
    // Check if the connection is waiting to be reset
    if (socketWaitReset()) {
	socketSetReading(false);
	return false;
    }
    // Check if something changed
    if (!(m_socket && socketReading())) {
	Debug(this,DebugAll,"Socket deleted while reading [%p]",this);
	return false;
    }
    if (read && read != Socket::socketError()) {
	if (!flag(StreamCompressed)) {
	    buf[read] = 0;
#ifdef JBSTREAM_DEBUG_SOCKET
	    Debug(this,DebugInfo,"Received %s [%p]",buf,this);
#endif
	    if (!m_xmlDom->parse(buf)) {
		if (m_xmlDom->error() != XmlSaxParser::Incomplete)
		    error = XMPPError::Xml;
		else if (m_xmlDom->buffer().length() > m_engine->m_maxIncompleteXml)
		    error = XMPPError::Policy;
	    }
	}
	else if (m_compress) {
#ifdef JBSTREAM_DEBUG_SOCKET
	    Debug(this,DebugInfo,"Received %d compressed bytes [%p]",read,this);
#endif
	    DataBlock d;
	    int res = m_compress->decompress(buf,read,d);
	    if (res == read) {
#ifdef JBSTREAM_DEBUG_COMPRESS
		Debug(this,DebugInfo,"Decompressed %d --> %u [%p]",read,d.length(),this);
#endif
		if (d.length()) {
		    char c = 0;
		    d.append(&c,1);
		    buf = (char*)d.data();
#ifdef JBSTREAM_DEBUG_SOCKET
		    Debug(this,DebugInfo,"Received compressed %s [%p]",buf,this);
#endif
		    if (!m_xmlDom->parse(buf)) {
			if (m_xmlDom->error() != XmlSaxParser::Incomplete)
			    error = XMPPError::Xml;
			else if (m_xmlDom->buffer().length() > m_engine->m_maxIncompleteXml)
			    error = XMPPError::Policy;
		    }
		}
	    }
	    else
		error = XMPPError::UndefinedCondition;
	}
	else
	    error = XMPPError::Internal;
    }
    socketSetReading(false);
    if (read) {
	if (read == Socket::socketError()) {
	    if (m_socket->canRetry()) {
		read = 0;
#ifdef XDEBUG
		String tmp;
		Thread::errorString(tmp,m_socket->error());
		Debug(this,DebugAll,"Socket temporary unavailable for read. %d: '%s' [%p]",
		    m_socket->error(),tmp.c_str(),this);
#endif
	    }
	    else
		error = XMPPError::SocketError;
	}
    }
    else
	error = XMPPError::SocketError;
    if (error == XMPPError::NoError) {
	// Stop reading if waiting for TLS start and received a complete element
	// We'll wait for the stream processor to handle the received element
	if (read && state() == WaitTlsRsp && !m_xmlDom->buffer().length() &&
	    m_xmlDom->unparsed() == XmlSaxParser::None) {
	    XmlDocument* doc = m_xmlDom->document();
	    // If received a complete element, the parser's current element is
	    // the document's root
	    if (doc && m_xmlDom->isCurrent(doc->root())) {
		DDebug(this,DebugAll,"Received complete element in state=%s. Stop reading [%p]",
		    stateName(),this);
		socketSetCanRead(false);
	    }
	}
	return read > 0;
    }
    // Error
    int location = 0;
    String reason;
    if (error != XMPPError::SocketError) {
	if (error == XMPPError::Xml) {
	    reason << "Parser error '" << m_xmlDom->getError() << "'";
	    Debug(this,DebugNote,"%s buffer='%s' [%p]",
		reason.c_str(),m_xmlDom->buffer().c_str(),this);
	}
	else if (error == XMPPError::UndefinedCondition) {
	    reason = "Decompression failure";
	    Debug(this,DebugNote,"Decompressor failure [%p]",this);
	}
	else if (error == XMPPError::Internal) {
	    reason = "Decompression failure";
	    Debug(this,DebugNote,"No decompressor [%p]",this);
	}
	else {
	    reason = "Parser error 'XML element too long'";
	    Debug(this,DebugNote,"Parser overflow len=%u max= %u [%p]",
		m_xmlDom->buffer().length(),m_engine->m_maxIncompleteXml,this);
	}
    }
    else if (read) {
	String tmp;
	Thread::errorString(tmp,m_socket->error());
	reason << "Socket read error: " << tmp << " (" << m_socket->error() << ")";
	Debug(this,DebugWarn,"%s [%p]",reason.c_str(),this);
    }
    else {
	reason = "Stream EOF";
	Debug(this,DebugInfo,"%s [%p]",reason.c_str(),this);
	location = 1;
    }
    socketSetCanRead(false);
    lck.drop();
    postponeTerminate(location,m_incoming,error,reason);
    return read > 0;
}

// Stream state processor
JBEvent* JBStream::getEvent(u_int64_t time)
{
    if (m_lastEvent)
	return 0;
    Lock lock(this);
    if (m_lastEvent)
	return 0;
    XDebug(this,DebugAll,"JBStream::getEvent() [%p]",this);
    checkPendingEvent();
    if (!m_lastEvent) {
	if (canProcess(time)) {
	    process(time);
	    checkPendingEvent();
	    if (!m_lastEvent)
		checkTimeouts(time);
	}
	else
	    checkPendingEvent();
    }
#ifdef XDEBUG
    if (m_lastEvent)
	Debug(this,DebugAll,"Generating event (%p,%s) in state '%s' [%p]",
	    m_lastEvent,m_lastEvent->name(),stateName(),this);
#endif
    return m_lastEvent;
}

// Send a stanza ('iq', 'message' or 'presence') or dialback elements in Running state.
bool JBStream::sendStanza(XmlElement*& xml)
{
    if (!xml)
	return false;
    DDebug(this,DebugAll,"sendStanza(%p) '%s' [%p]",xml,xml->tag(),this);
    if (!(XMPPUtils::isStanza(*xml) ||
	(m_type == s2s && XMPPUtils::hasXmlns(*xml,XMPPNamespace::Dialback)))) {
	Debug(this,DebugNote,"Request to send non stanza xml='%s' [%p]",xml->tag(),this);
	TelEngine::destruct(xml);
	return false;
    }
    XmlElementOut* xo = new XmlElementOut(xml);
    xml = 0;
    xo->prepareToSend();
    Lock lock(this);
    m_pending.append(xo);
    sendPending();
    return true;
}

// Send stream related XML when negotiating the stream
//  or some other stanza in non Running state
bool JBStream::sendStreamXml(State newState, XmlElement* first, XmlElement* second,
    XmlElement* third)
{
    DDebug(this,DebugAll,"sendStreamXml(%s,%p,%p,%p) [%p]",
	stateName(),first,second,third,this);
    Lock lock(this);
    bool ok = false;
    XmlFragment frag;
    // Use a do while() to break to the end: safe cleanup
    do {
	if (m_state == Idle || m_state == Destroy)
	    break;
	// Check if we have unsent stream xml
	if (m_outStreamXml)
	    sendPending(true);
	if (m_outStreamXml)
	    break;
	if (!first)
	    break;
	// Add stream declaration before stream start
	if (first->getTag() == XMPPUtils::s_tag[XmlTag::Stream] &&
	    first->tag()[0] != '/') {
	    XmlDeclaration* decl = new XmlDeclaration;
	    decl->toString(m_outStreamXml,true);
	    frag.addChild(decl);
	}
	first->toString(m_outStreamXml,true,String::empty(),String::empty(),false);
	frag.addChild(first);
	if (second) {
	    second->toString(m_outStreamXml,true,String::empty(),String::empty(),false);
	    frag.addChild(second);
	    if (third) {
		third->toString(m_outStreamXml,true,String::empty(),String::empty(),false);
		frag.addChild(third);
	    }
	}
	first = second = third = 0;
	if (flag(StreamCompressed) && !compress()) {
	    ok = false;
	    break;
	}
	m_engine->printXml(this,true,frag);
	ok = sendPending(true);
    } while (false);
    TelEngine::destruct(first);
    TelEngine::destruct(second);
    TelEngine::destruct(third);
    if (ok)
	changeState(newState);
    return ok;
}

// Start the stream. This method should be called by the upper layer
//  when processing an incoming stream Start event
void JBStream::start(XMPPFeatureList* features, XmlElement* caps, bool useVer1)
{
    Lock lock(this);
    if (m_state != Starting)
	return;
    if (outgoing()) {
	TelEngine::destruct(features);
	TelEngine::destruct(caps);
	if (m_type == c2s) {
	    // c2s: just wait for stream features
	    changeState(Features);
	}
	else if (m_type == s2s) {
	    // Wait features ?
	    if (flag(StreamRemoteVer1)) {
		changeState(Features);
		return;
	    }
	    // Stream not secured
	    if (!flag(StreamSecured)) {
		// Accept dialback auth stream
		// The namspace presence was already checked in checkStreamStart()
		if (flag(TlsRequired)) {
		    terminate(0,false,0,XMPPError::EncryptionRequired);
		    return;
		}
	    }
	    setFlags(StreamSecured);
	    serverStream()->sendDialback();
	}
	else if (m_type == cluster)
	    changeState(Features);
	else if (m_type == comp)
	    serverStream()->startComp();
	else
	    DDebug(this,DebugStub,"JBStream::start() not handled for type=%s",typeName());
	return;
    }
    m_features.clear();
    if (features)
	m_features.add(*features);
    if (useVer1 && flag(StreamRemoteVer1))
	setFlags(StreamLocalVer1);
    if (flag(StreamRemoteVer1) && flag(StreamLocalVer1)) {
	// Set secured flag if we don't advertise TLS
	if (!(flag(StreamSecured) || m_features.get(XMPPNamespace::Tls)))
	    setSecured();
	// Set authenticated flag if we don't advertise authentication mechanisms
	if (flag(StreamSecured)) {
	    if (flag(StreamAuthenticated))
	        m_features.remove(XMPPNamespace::Sasl);
	    else if (!m_features.get(XMPPNamespace::Sasl))
		setFlags(StreamAuthenticated);
	}
    }
    else
	// c2s using non-sasl auth or s2s not using TLS
	setSecured();
    // Send start and features
    XmlElement* s = buildStreamStart();
    XmlElement* f = 0;
    if (flag(StreamRemoteVer1) && flag(StreamLocalVer1))
	f = m_features.buildStreamFeatures();
    if (f && caps)
	f->addChild(caps);
    else
	TelEngine::destruct(caps);
    State newState = Features;
    if (m_type == c2s) {
	// Change stream state to Running if authenticated and there is no required
	// feature to negotiate
	if (flag(StreamAuthenticated) && !firstRequiredFeature())
	    newState = Running;
    }
    else if (m_type == s2s) {
	// Change stream state to Running if authenticated and features list is empty
	if (flag(StreamAuthenticated) && !m_features.skipNull())
	    newState = Running;
    }
    else if (m_type == cluster) {
	// Change stream state to Running if authenticated and features list is empty
	if (flag(StreamAuthenticated) && !m_features.skipNull())
	    newState = Running;
    }
    sendStreamXml(newState,s,f);
}

// Authenticate an incoming stream
bool JBStream::authenticated(bool ok, const String& rsp, XMPPError::Type error,
    const char* username, const char* id, const char* resource)
{
    Lock lock(this);
    if (m_state != Auth || !incoming())
	return false;
    DDebug(this,DebugAll,"authenticated(%s,'%s',%s) local=%s [%p]",
	String::boolText(ok),rsp.safe(),XMPPUtils::s_error[error].c_str(),
	m_local.c_str(),this);
    if (ok) {
	if (m_type == c2s) {
	    if (m_sasl) {
		// Set remote party node if provided
		if (!TelEngine::null(username)) {
		    m_remote.set(username,m_local.domain(),"");
		    Debug(this,DebugAll,"Remote party set to '%s' [%p]",m_remote.c_str(),this);
		}
		String text;
		m_sasl->buildAuthRspReply(text,rsp);
		XmlElement* s = XMPPUtils::createElement(XmlTag::Success,
		    XMPPNamespace::Sasl,text);
		ok = sendStreamXml(WaitStart,s);
	    }
	    else if (m_features.get(XMPPNamespace::IqAuth)) {
		// Set remote party if not provided
		if (!TelEngine::null(username))
		    m_remote.set(username,m_local.domain(),resource);
		else
		    m_remote.resource(resource);
		if (m_remote.isFull()) {
		    Debug(this,DebugAll,"Remote party set to '%s' [%p]",m_remote.c_str(),this);
		    XmlElement* rsp = XMPPUtils::createIqResult(0,0,id,
			XMPPUtils::createElement(XmlTag::Query,XMPPNamespace::IqAuth));
		    ok = sendStreamXml(Running,rsp);
		    if (!ok)
			m_remote.set(m_local.domain());
		}
		else
		    terminate(0,true,0,XMPPError::Internal);
	    }
	    else
		terminate(0,true,0,XMPPError::Internal);
	}
	else if (m_type == s2s)
	    ok = false;
	else if (m_type == comp) {
	    XmlElement* rsp = XMPPUtils::createElement(XmlTag::Handshake);
	    ok = sendStreamXml(Running,rsp);
	}
	if (ok) {
	    m_features.remove(XMPPNamespace::Sasl);
	    m_features.remove(XMPPNamespace::IqAuth);
	    setFlags(StreamAuthenticated);
	}
    }
    else {
	if (m_type == c2s) {
	    XmlElement* rsp = 0;
	    if (m_sasl)
		rsp = XMPPUtils::createFailure(XMPPNamespace::Sasl,error);
	    else {
		rsp = XMPPUtils::createIq(XMPPUtils::IqError,0,0,id);
		if (TelEngine::null(id))
		    rsp->addChild(XMPPUtils::createElement(XmlTag::Query,XMPPNamespace::IqAuth));
		rsp->addChild(XMPPUtils::createError(XMPPError::TypeAuth,error));
	    }
	    ok = sendStreamXml(Features,rsp);
	}
	else if (m_type == s2s)
	    ok = false;
	else if (m_type == comp)
	    terminate(0,true,0,XMPPError::NotAuthorized);
    }
    TelEngine::destruct(m_sasl);
    return ok;
}

// Terminate the stream. Send stream end tag or error.
// Reset the stream. Deref stream if destroying.
void JBStream::terminate(int location, bool destroy, XmlElement* xml, int error,
    const char* reason, bool final, bool genEvent, const char* content)
{
    XDebug(this,DebugAll,"terminate(%d,%u,%p,%u,%s,%u) state=%s [%p]",
	location,destroy,xml,error,reason,final,stateName(),this);
    Lock lock(this);
    m_pending.clear();
    m_outXmlCompress.clear();
    resetPostponedTerminate();
    // Already in destroy
    if (state() == Destroy) {
	TelEngine::destruct(xml);
	return;
    }
    bool sendEndTag = true;
    destroy = destroy || final || flag(NoAutoRestart);
    // Set error flag
    if (state() == Running) {
	if (error != XMPPError::NoError)
	    setFlags(InError);
	else
	    resetFlags(InError);
    }
    else
	setFlags(InError);
    if (flag(InError)) {
	// Reset re-connect counter if not internal policy error
	if (location || error != XMPPError::Policy)
	    m_restart = 0;
    }
    if (error == XMPPError::NoError && m_engine->exiting())
	error = XMPPError::Shutdown;
    // Last check for sendEndTag
    if (sendEndTag) {
	// Prohibitted states or socket read/write error
	if (m_state == Destroy || m_state == Securing || m_state == Connecting)
	    sendEndTag = false;
	else if (error == XMPPError::SocketError) {
	    sendEndTag = false;
	    reason = "I/O error";
	}
    }
    Debug(this,DebugAll,
	"Terminate by '%s' state=%s destroy=%u error=%s reason='%s' final=%u [%p]",
	lookup(location,s_location),stateName(),destroy,
	XMPPUtils::s_error[error].c_str(),reason,final,this);
    if (sendEndTag) {
	XmlElement* start = 0;
	if (m_state == Starting && incoming())
	    start = buildStreamStart();
	XmlElement* end = new XmlElement(String("/stream:stream"),false);
	if (error != XMPPError::NoError && location < 1) {
	    XmlElement* e = XMPPUtils::createStreamError(error,reason,content);
	    if (!start)
		sendStreamXml(m_state,e,end);
	    else
		sendStreamXml(m_state,start,e,end);
	}
	else {
	    if (!start)
		sendStreamXml(m_state,end);
	    else
		sendStreamXml(m_state,start,end);
	}
    }
    resetConnection();
    m_outStreamXml.clear();
    m_outStreamXmlCompress.clear();

    // Always set termination event, except when called from destructor
    if (genEvent && !(final || m_terminateEvent)) {
	// TODO: Cancel all outgoing elements without id
	m_terminateEvent = new JBEvent(destroy ? JBEvent::Destroy : JBEvent::Terminated,
	    this,xml);
	xml = 0;
	if (!m_terminateEvent->m_text) {
	    if (TelEngine::null(reason))
		m_terminateEvent->m_text = XMPPUtils::s_error[error];
	    else
		m_terminateEvent->m_text = reason;
	}
    }
    TelEngine::destruct(xml);

    changeState(destroy ? Destroy : Idle);
}

// Close the stream. Release memory
void JBStream::destroyed()
{
    terminate(0,true,0,XMPPError::NoError,"",true);
    resetConnection();
    if (m_engine)
	m_engine->removeStream(this,false);
    TelEngine::destruct(m_terminateEvent);
    DDebug(this,DebugAll,"Destroyed local=%s remote=%s [%p]",
	m_local.safe(),m_remote.safe(),this);
    RefObject::destroyed();
}

// Check if stream state processor can continue
// This method is called from getEvent() with the stream locked
bool JBStream::canProcess(u_int64_t time)
{
    if (outgoing()) {
	// Increase stream restart counter if it's time to and should auto restart
	if (!flag(NoAutoRestart) && m_timeToFillRestart < time) {
	    m_timeToFillRestart = time + m_engine->m_restartUpdInterval;
	    if (m_restart < m_engine->m_restartMax) {
		m_restart++;
		DDebug(this,DebugAll,"Restart count set to %u max=%u [%p]",
		    m_restart,m_engine->m_restartMax,this);
	    }
	}
	if (state() == Idle) {
	    // Re-connect
	    bool conn = (m_connectStatus > JBConnect::Start);
	    if (!conn && m_restart) {
		// Don't connect non client/component or cluster if we are in error and
		//  have nothing to send
		if (m_type != c2s && m_type != comp && m_type != cluster &&
		    flag(InError) && !m_pending.skipNull())
		    return false;
		conn = true;
		m_restart--;
	    }
	    if (conn) {
		resetFlags(InError);
		changeState(Connecting);
		m_engine->connectStream(this);
		return false;
	    }
	    // Destroy if not auto-restarting
	    if (flag(NoAutoRestart)) {
		terminate(0,true,0);
		return false;
	    }
	}
    }
    else if (state() == Idle && flag(NoAutoRestart)) {
	terminate(0,true,0);
	return false;
    }
    return true;
}

// Process stream state. Get XML from parser's queue and process it
// This method is called from getEvent() with the stream locked
void JBStream::process(u_int64_t time)
{
    if (!m_xmlDom)
	return;
    XDebug(this,DebugAll,"JBStream::process() [%p]",this);
    while (true) {
	sendPending();
	if (m_terminateEvent)
	    break;

	// Lock the parser to obtain the root and/or child
	// Unlock it before processing received element
	Lock lockDoc(m_socketMutex);
	XmlDocument* doc = m_xmlDom ? m_xmlDom->document() : 0;
	XmlElement* root = doc ? doc->root(false) : 0;
	if (!root)
	    break;

	if (m_state == WaitStart) {
	    // Print the declaration
	    XmlDeclaration* dec = doc->declaration();
	    if (dec)
		m_engine->printXml(this,false,*dec);
	    XmlElement xml(*root);
	    lockDoc.drop();
	    // Print the root. Make sure we don't print its children
	    xml.clearChildren();
	    m_engine->printXml(this,false,xml);
	    // Check if valid
	    if (!XMPPUtils::isTag(xml,XmlTag::Stream,XMPPNamespace::Stream)) {
		String* ns = xml.xmlns();
		Debug(this,DebugMild,"Received invalid stream root '%s' namespace='%s' [%p]",
		    xml.tag(),TelEngine::c_safe(ns),this);
		terminate(0,true,0);
		break;
	    }
	    // Check 'from' and 'to'
	    JabberID from;
	    JabberID to;
	    if (!getJids(&xml,from,to))
		break;
	    DDebug(this,DebugAll,"Processing root '%s' in state %s [%p]",
		xml.tag(),stateName(),this);
	    processStart(&xml,from,to);
	    break;
	}

	XmlElement* xml = root->pop();
	if (!xml) {
	    if (root->completed())
		socketSetCanRead(false);
	    if (m_events.skipNull())
	        break;
	    if (!root->completed()) {
		if (m_ppTerminate && !(m_pending.skipNull() && socketCanWrite())) {
		    lockDoc.drop();
		    postponedTerminate();
		}
		break;
	    }
	    DDebug(this,DebugAll,"Remote closed the stream in state %s [%p]",
		stateName(),this);
	    lockDoc.drop();
	    resetPostponedTerminate();
	    terminate(1,false,0);
	    break;
	}
	lockDoc.drop();

	// Process received element
	// Print it
	m_engine->printXml(this,false,*xml);
	// Check stream termination
	if (streamError(xml))
	    break;
	// Check 'from' and 'to'
	JabberID from;
	JabberID to;
	if (!getJids(xml,from,to))
	    break;
	// Restart the idle timer
	setIdleTimer(time);
	// Check if a received stanza is valid and allowed in current state
	if (!checkStanzaRecv(xml,from,to))
	    break;

	DDebug(this,DebugAll,"Processing (%p,%s) in state %s [%p]",
	    xml,xml->tag(),stateName(),this);

	// Process here dialback verify
	if (m_type == s2s && isDbVerify(*xml)) {
	    switch (state()) {
		case Running:
		case Features:
		case Starting:
		case Challenge:
		case Auth:
		    m_events.append(new JBEvent(JBEvent::DbVerify,this,xml,from,to));
		    break;
		default:
		    dropXml(xml,"dialback verify in unsupported state");
	    }
	    continue;
	}

	switch (m_state) {
	    case Running:
		processRunning(xml,from,to);
		// Reset ping
		setNextPing(true);
		m_pingId = "";
		break;
	    case Features:
		if (m_incoming)
		    processFeaturesIn(xml,from,to);
		else
		    processFeaturesOut(xml,from,to);
		break;
	    case WaitStart:
	    case Starting:
		processStart(xml,from,to);
		TelEngine::destruct(xml);
		break;
	    case Challenge:
		processChallenge(xml,from,to);
		break;
	    case Auth:
		processAuth(xml,from,to);
		break;
	    case WaitTlsRsp:
		processWaitTlsRsp(xml,from,to);
		break;
	    case Register:
		processRegister(xml,from,to);
		break;
	    case Compressing:
		processCompressing(xml,from,to);
		break;
	    default:
		dropXml(xml,"unhandled stream state in process()");
	}
	break;
    }
    XDebug(this,DebugAll,"JBStream::process() exiting [%p]",this);
}

// Process elements in Running state
bool JBStream::processRunning(XmlElement* xml, const JabberID& from, const JabberID& to)
{
    if (!xml)
	return true;
    int t, ns;
    if (!XMPPUtils::getTag(*xml,t,ns))
	return dropXml(xml,"failed to retrieve element tag");
    switch (t) {
	case XmlTag::Message:
	    if (ns != m_xmlns)
		break;
	    m_events.append(new JBEvent(JBEvent::Message,this,xml,from,to));
	    return true;
	case XmlTag::Presence:
	    if (ns != m_xmlns)
		break;
	    m_events.append(new JBEvent(JBEvent::Presence,this,xml,from,to));
	    return true;
	case XmlTag::Iq:
	    if (ns != m_xmlns)
		break;
	    checkPing(this,xml,m_pingId);
	    m_events.append(new JBEvent(JBEvent::Iq,this,xml,from,to,xml->findFirstChild()));
	    return true;
	default:
	    m_events.append(new JBEvent(JBEvent::Unknown,this,xml,from,to));
	    return true;
    }
    // Invalid stanza namespace
    XmlElement* rsp = XMPPUtils::createError(xml,XMPPError::TypeModify,
	XMPPError::InvalidNamespace,"Only stanzas in default namespace are allowed");
    sendStanza(rsp);
    return true;
}

// Check stream timeouts
// This method is called from getEvent() with the stream locked, after
void JBStream::checkTimeouts(u_int64_t time)
{
    if (m_ppTerminateTimeout && m_ppTerminateTimeout <= time) {
	m_ppTerminateTimeout = 0;
	Debug(this,DebugAll,"Postponed termination timed out [%p]",this);
	if (postponedTerminate())
	    return;
    }
    // Running: check ping and idle timers
    if (m_state == Running) {
	const char* reason = 0;
	if (m_pingTimeout) {
	    if (m_pingTimeout < time) {
		Debug(this,DebugNote,"Ping stanza with id '%s' timed out [%p]",m_pingId.c_str(),this);
		reason = "Ping timeout";
	    }
	}
	else if (m_nextPing && time >= m_nextPing) {
	    XmlElement* ping = setNextPing(false);
	    if (ping) {
		DDebug(this,DebugAll,"Sending ping with id=%s [%p]",m_pingId.c_str(),this);
		if (!sendStanza(ping))
		    m_pingId = "";
	    }
	    else {
		resetPing();
		m_pingId = "";
	    }
	}
	if (m_idleTimeout && m_idleTimeout < time && !reason)
	    reason = "Stream idle";
	if (reason)
	    terminate(0,m_incoming,0,XMPPError::ConnTimeout,reason);
	return;
    }
    // Stream setup timer
    if (m_setupTimeout && m_setupTimeout < time) {
	terminate(0,m_incoming,0,XMPPError::Policy,"Stream setup timeout");
	return;
    }
    // Stream start timer
    if (m_startTimeout && m_startTimeout < time) {
	terminate(0,m_incoming,0,XMPPError::Policy,"Stream start timeout");
	return;
    }
    // Stream connect timer
    if (m_connectTimeout && m_connectTimeout < time) {
	DDebug(this,DebugNote,"Connect timed out stat=%s [%p]",
	    lookup(m_connectStatus,JBConnect::s_statusName),this);
	// Don't terminate if there are more connect options
	if (state() == Connecting && m_connectStatus > JBConnect::Start) {
	    m_engine->stopConnect(toString());
	    m_engine->connectStream(this);
	}
	else
	    terminate(0,m_incoming,0,XMPPError::ConnTimeout,"Stream connect timeout");
	return;
    }
}

// Reset the stream's connection. Build a new XML parser if the socket is valid
void JBStream::resetConnection(Socket* sock)
{
    DDebug(this,DebugAll,"JBStream::resetConnection(%p) current=%p [%p]",
	sock,m_socket,this);
    // Release the old one
    if (m_socket) {
	m_socketMutex.lock();
	m_socketFlags |= SocketWaitReset;
	m_socketMutex.unlock();
	// Wait for the socket to become available (not reading or writing)
	Socket* tmp = 0;
	while (true) {
	    Lock lock(m_socketMutex);
	    if (!(m_socket && (socketReading() || socketWriting()))) {
		tmp = m_socket;
		m_socket = 0;
		m_socketFlags = 0;
		if (m_xmlDom) {
		    delete m_xmlDom;
		    m_xmlDom = 0;
		}
		TelEngine::destruct(m_compress);
		break;
	    }
	    lock.drop();
	    Thread::yield(false);
	}
	if (tmp) {
	    tmp->setLinger(-1);
	    tmp->terminate();
	    delete tmp;
	}
    }
    resetPostponedTerminate();
    if (sock) {
	Lock lock(m_socketMutex);
	if (m_socket) {
	    Debug(this,DebugWarn,"Duplicate attempt to set socket! [%p]",this);
	    delete sock;
	    return;
	}
	m_xmlDom = new XmlDomParser(debugName());
	m_xmlDom->debugChain(this);
	m_socket = sock;
	if (debugAt(DebugAll)) {
	    SocketAddr l, r;
	    localAddr(l);
	    remoteAddr(r);
	    Debug(this,DebugAll,"Connection set local=%s:%d remote=%s:%d sock=%p [%p]",
		l.host().c_str(),l.port(),r.host().c_str(),r.port(),m_socket,this);
	}
	m_socket->setReuse(true);
	m_socket->setBlocking(false);
	socketSetCanRead(true);
	socketSetCanWrite(true);
    }
}

// Build a ping iq stanza
XmlElement* JBStream::buildPing(const String& stanzaId)
{
    return 0;
}

// Build a stream start XML element
XmlElement* JBStream::buildStreamStart()
{
    XmlElement* start = new XmlElement(XMPPUtils::s_tag[XmlTag::Stream],false);
    if (incoming())
	start->setAttribute("id",m_id);
    XMPPUtils::setStreamXmlns(*start);
    start->setAttribute(XmlElement::s_ns,XMPPUtils::s_ns[m_xmlns]);
    start->setAttributeValid("from",m_local.bare());
    start->setAttributeValid("to",m_remote.bare());
    if (outgoing() || flag(StreamRemoteVer1))
	start->setAttribute("version","1.0");
    start->setAttribute("xml:lang","en");
    return start;
}

// Process received elements in WaitStart state
// WaitStart: Incoming: waiting for stream start
//            Outgoing: idem (our stream start was already sent)
// Return false if stream termination was initiated
bool JBStream::processStart(const XmlElement* xml, const JabberID& from,
    const JabberID& to)
{
    Debug(this,DebugStub,"JBStream::processStart(%s) [%p]",xml->tag(),this);
    return true;
}

// Process elements in Compressing state
// Return false if stream termination was initiated
bool JBStream::processCompressing(XmlElement* xml, const JabberID& from,
    const JabberID& to)
{
    XDebug(this,DebugAll,"JBStream::processCompressing() [%p]",this);
    int t = XmlTag::Count;
    int n = XMPPNamespace::Count;
    XMPPUtils::getTag(*xml,t,n);
    if (outgoing()) {
	if (n != XMPPNamespace::Compress)
	    return dropXml(xml,"expecting compression namespace");
	// Expecting: compressed/failure
	bool ok = (t == XmlTag::Compressed);
	if (!ok && t != XmlTag::Failure)
	    return dropXml(xml,"expecting compress response (compressed/failure)");
	if (ok) {
	    if (m_compress)
		setFlags(StreamCompressed);
	    else
		return destroyDropXml(xml,XMPPError::Internal,"no compressor");
	}
	else {
	    XmlElement* ch = xml->findFirstChild();
	    Debug(this,DebugInfo,"Compress failed at remote party error=%s [%p]",
		ch ? ch->tag() : "",this);
	    TelEngine::destruct(m_compress);
	}
	TelEngine::destruct(xml);
	// Restart the stream on success
	if (ok) {
	    XmlElement* s = buildStreamStart();
	    return sendStreamXml(WaitStart,s);
	}
	// Compress failed: continue
	JBServerStream* server = serverStream();
	if (server)
	    return server->sendDialback();
	JBClientStream* client = clientStream();
	if (client)
	    return client->bind();
	Debug(this,DebugNote,"Unhandled stream type in %s state [%p]",stateName(),this);
	terminate(0,true,0,XMPPError::Internal);
	return true;
    }
    // Authenticated incoming s2s waiting for compression or any other element
    if (type() == s2s && m_features.get(XMPPNamespace::CompressFeature)) {
	if (t == XmlTag::Compress && n == XMPPNamespace::Compress)
	    return handleCompressReq(xml);
	// Change state to Running
	changeState(Running);
	return processRunning(xml,from,to);
    }

    return dropXml(xml,"not implemented");
}

// Process elements in Register state
bool JBStream::processRegister(XmlElement* xml, const JabberID& from,
    const JabberID& to)
{
    dropXml(xml,"can't process in this state");
    terminate(0,true,0,XMPPError::Internal);
    return false;
}

// Process elements in Auth state
bool JBStream::processAuth(XmlElement* xml, const JabberID& from,
    const JabberID& to)
{
    return dropXml(xml,"can't process in this state");
}

// Check if a received start start element's namespaces are correct.
bool JBStream::processStreamStart(const XmlElement* xml)
{
    XDebug(this,DebugAll,"JBStream::processStreamStart() [%p]",this);
    if (m_state == Starting)
	return true;
    changeState(Starting);
    if (!XMPPUtils::hasDefaultXmlns(*xml,m_xmlns)) {
	Debug(this,DebugNote,"Received '%s' with invalid xmlns='%s' [%p]",
	    xml->tag(),TelEngine::c_safe(xml->xmlns()),this);
	terminate(0,m_incoming,0,XMPPError::InvalidNamespace);
	return false;
    }
    XMPPError::Type error = XMPPError::NoError;
    const char* reason = 0;
    while (true) {
	if (m_type != c2s && m_type != s2s && m_type != comp && m_type != cluster) {
	    Debug(this,DebugStub,"processStreamStart() type %u not handled!",m_type);
	    error = XMPPError::Internal;
	    break;
	}
	// Check xmlns:stream
	String* nsStr = xml->getAttribute("xmlns:stream");
	if (!nsStr || *nsStr != XMPPUtils::s_ns[XMPPNamespace::Stream]) {
	    Debug(this,DebugNote,"Received '%s' with invalid xmlns:stream='%s' [%p]",
		xml->tag(),TelEngine::c_safe(nsStr),this);
	    error = XMPPError::InvalidNamespace;
	    break;
	}
	// Check version
	String ver(xml->getAttribute("version"));
	int remoteVersion = -1;
	if (ver) {
	    int pos = ver.find('.');
	    if (pos > 0)
		remoteVersion = ver.substr(0,pos).toInteger(-1);
	}
	if (remoteVersion == 1)
	    setFlags(StreamRemoteVer1);
	else if (remoteVersion < 1) {
	    if (m_type == c2s)
		XDebug(this,DebugAll,"c2s stream start with version < 1 [%p]",this);
	    else if (m_type == s2s) {
		// Accept invalid/unsupported version only if TLS is not required
		if (!flag(TlsRequired)) {
		    // Check dialback
		    if (!xml->hasAttribute("xmlns:db",XMPPUtils::s_ns[XMPPNamespace::Dialback]))
			error = XMPPError::InvalidNamespace;
		}
		else
		    error = XMPPError::EncryptionRequired;
	    }
	    else if (m_type != comp)
		error = XMPPError::Internal;
	}
	else if (remoteVersion > 1)
	    error = XMPPError::UnsupportedVersion;
	if (error != XMPPError::NoError) {
	    Debug(this,DebugNote,"Unacceptable '%s' version='%s' error=%s [%p]",
		xml->tag(),ver.c_str(),XMPPUtils::s_error[error].c_str(),this);
	    break;
	}
	// Set stream id: generate one for incoming, get it from xml if outgoing
	if (incoming()) {
	    // Generate a random, variable length stream id
	    MD5 md5(String((int)(int64_t)this));
	    md5 << m_name << String((int)Time::msecNow());
	    m_id = md5.hexDigest();
	    m_id << "_" << String((int)Random::random());
	}
	else {
	    m_id = xml->getAttribute("id");
	    if (!m_id) {
		Debug(this,DebugNote,"Received '%s' with empty stream id [%p]",
		    xml->tag(),this);
		reason = "Missing stream id";
		error = XMPPError::InvalidId;
		break;
	    }
	}
	XDebug(this,DebugAll,"Stream id set to '%s' [%p]",m_id.c_str(),this);
	break;
    }
    if (error == XMPPError::NoError)
	return true;
    terminate(0,m_incoming,0,error,reason);
    return false;
}

// Handle an already checked (tag and namespace) compress request
// Respond to it. Change stream state on success
bool JBStream::handleCompressReq(XmlElement* xml)
{
    XMPPError::Type error = XMPPError::UnsupportedMethod;
    State newState = state();
    XmlElement* rsp = 0;
    XmlElement* m = XMPPUtils::findFirstChild(*xml,XmlTag::Method,
	XMPPNamespace::Compress);
    if (m) {
	// Get and check the method
	const String& method = m->getText();
	XMPPFeatureCompress* c = m_features.getCompress();
	if (method && c && c->hasMethod(method)) {
	    // Build the (de)compressor
	    Lock lock(m_socketMutex);
	    m_engine->compressStream(this,method);
	    if (m_compress) {
		newState = WaitStart;
		setFlags(SetCompressed);
		m_features.remove(XMPPNamespace::CompressFeature);
		rsp = XMPPUtils::createElement(XmlTag::Compressed,XMPPNamespace::Compress);
	    }
	    else
		error = XMPPError::SetupFailed;
	}
    }
    TelEngine::destruct(xml);
    if (!rsp)
	rsp = XMPPUtils::createFailure(XMPPNamespace::Compress,error);
    return sendStreamXml(newState,rsp);
}

// Check if a received element is a stream error one
bool JBStream::streamError(XmlElement* xml)
{
    if (!(xml && XMPPUtils::isTag(*xml,XmlTag::Error,XMPPNamespace::Stream)))
	return false;
    String text;
    String error;
    String content;
    XMPPUtils::decodeError(xml,XMPPNamespace::StreamError,&error,&text,&content);
    Debug(this,DebugAll,"Received stream error '%s' content='%s' text='%s' in state %s [%p]",
	error.c_str(),content.c_str(),text.c_str(),stateName(),this);
    int err = XMPPUtils::s_error[error];
    if (err >= XMPPError::Count)
	err = XMPPError::NoError;
    String rAddr;
    int rPort = 0;
    if (err == XMPPError::SeeOther && content) {
	if (m_redirectCount < m_redirectMax) {
	    int pos = content.rfind(':');
	    if (pos >= 0) {
		rAddr = content.substr(0,pos);
		if (rAddr) {
		    rPort = content.substr(pos + 1).toInteger(0);
		    if (rPort < 0)
			rPort = 0;
		}
	    }
	    else
		rAddr = content;
	    if (rAddr) {
		// Check if the connect destination is different
		SocketAddr remoteIp;
		remoteAddr(remoteIp);
		bool sameDest = (rAddr == serverHost()) || (rAddr == m_connectAddr) || (rAddr == remoteIp.host());
		if (sameDest) {
		    sameDest = ((rPort > 0 ? rPort : XMPP_C2S_PORT) == remoteIp.port());
		    if (sameDest) {
			Debug(this,DebugNote,"Ignoring redirect to same destination [%p]",this);
			rAddr = "";
		    }
		}
	    }
	}
    }
    terminate(1,false,xml,err,text,false,rAddr.null());
    setRedirect(rAddr,rPort);
    if (rAddr) {
	resetFlags(InError);
	resetConnectStatus();
	changeState(Connecting);
	m_engine->connectStream(this);
	setRedirect();
    }
    return true;
}

// Check if a received element has valid 'from' and 'to' jid attributes
bool JBStream::getJids(XmlElement* xml, JabberID& from, JabberID& to)
{
    if (!xml)
	return true;
    from = xml->getAttribute("from");
    to = xml->getAttribute("to");
    XDebug(this,DebugAll,"Got jids xml='%s' from='%s' to='%s' [%p]",
	xml->tag(),from.c_str(),to.c_str(),this);
    if (to.valid() && from.valid())
	return true;
    Debug(this,DebugNote,"Received '%s' with bad from='%s' or to='%s' [%p]",
	xml->tag(),from.c_str(),to.c_str(),this);
    terminate(0,m_incoming,xml,XMPPError::BadAddressing);
    return false;
}

// Check if a received element is a presence, message or iq qualified by the stream
//   namespace and the stream is not authenticated
// Validate 'from' for c2s streams
// Validate s2s 'to' domain and 'from' jid
bool JBStream::checkStanzaRecv(XmlElement* xml, JabberID& from, JabberID& to)
{
    if (!XMPPUtils::isStanza(*xml))
	return true;

    // RFC 3920bis 5.2: Accept stanzas only if the stream was authenticated
    // Accept IQs in jabber:iq:register namespace
    // Accept IQs in jabber:iq:auth namespace
    // They might be received on a non authenticated stream)
    if (!flag(StreamAuthenticated)) {
	bool isIq = XMPPUtils::isTag(*xml,XmlTag::Iq,m_xmlns);
	bool valid = isIq && XMPPUtils::findFirstChild(*xml,XmlTag::Count,
	    XMPPNamespace::IqRegister);
	JBClientStream* c2s = clientStream();
	if (!valid && c2s) {
	    // Outgoing client stream: check register responses
	    // Incoming client stream: check auth stanzas
	    if (outgoing())
		valid = c2s->isRegisterId(*xml);
	    else
		valid = isIq && XMPPUtils::findFirstChild(*xml,XmlTag::Count,
		    XMPPNamespace::IqAuth);
	}
	if (!valid) {
	    terminate(0,false,xml,XMPPError::NotAuthorized,
		"Can't accept stanza on non authorized stream");
	    return false;
	}
    }

    switch (m_type) {
	case c2s:
	    if (m_incoming) {
		// Check for valid from
		if (from && !m_remote.match(from)) {
		    XmlElement* e = XMPPUtils::createError(xml,
			XMPPError::TypeModify,XMPPError::BadAddressing);
		    sendStanza(e);
		    return false;
		}
		// Make sure the upper layer always has the full jid
		if (!from)
		    from = m_remote;
		else if (!from.resource())
		    from.resource(m_remote.resource());
	    }
	    else {
		XDebug(this,DebugStub,
		    "Possible checkStanzaRecv() unhandled outgoing c2s stream [%p]",this);
	    }
	    break;
	case comp:
	case s2s:
	    // RFC 3920bis 9.1.1.2 and 9.1.2.1:
	    // Validate 'to' and 'from'
	    // Accept anything for component streams
	    if (!(to && from)) {
		terminate(0,m_incoming,xml,XMPPError::BadAddressing);
		return false;
	    }
	    // TODO: Find an outgoing stream and send stanza error to the remote server
	    //  instead of terminating the stream
	    if (m_type == s2s) {
		if (incoming()) {
		    // Accept stanzas only for validated domains
		    if (!serverStream()->hasRemoteDomain(from.domain())) {
			terminate(0,m_incoming,xml,XMPPError::BadAddressing);
			return false;
		    }
		}
		else {
		    // We should not receive any stanza on outgoing s2s
		    terminate(0,m_incoming,xml,XMPPError::NotAuthorized);
		    return false;
		}
		if (m_local != to.domain()) {
		    terminate(0,m_incoming,xml,XMPPError::BadAddressing);
		    return false;
		}
	    }
	    else if (from.domain() != m_remote.domain()) {
		terminate(0,m_incoming,xml,XMPPError::InvalidFrom);
		return false;
	    }
	    break;
	case cluster:
	    break;
	default:
	    Debug(this,DebugStub,"checkStanzaRecv() unhandled stream type=%s [%p]",
		typeName(),this);
    }
    return true;
}

// Change stream state. Reset state depending data
void JBStream::changeState(State newState, u_int64_t time)
{
    if (newState == m_state)
	return;
    Debug(this,DebugAll,"Changing state from '%s' to '%s' [%p]",
	stateName(),lookup(newState,s_stateName),this);
    // Set/reset state depending data
    switch (m_state) {
	case Running:
	    resetPing();
	    m_pingId = "";
	    break;
	case WaitStart:
	    // Reset connect status if not timeout
	    if (m_startTimeout && m_startTimeout > time)
		resetConnectStatus();
	    m_startTimeout = 0;
	    break;
	case Securing:
	    setFlags(StreamSecured);
	    socketSetCanRead(true);
	    break;
	case Connecting:
	    m_connectTimeout = 0;
	    m_engine->stopConnect(toString());
	    break;
	case Register:
	    if (type() == c2s)
		clientStream()->m_registerReq = 0;
	    break;
	default: ;
    }
    switch (newState) {
	case WaitStart:
	    if (m_engine->m_setupTimeout && m_type != cluster)
		m_setupTimeout = time + timerMultiplier(this) * m_engine->m_setupTimeout;
	    else
		m_setupTimeout = 0;
	    m_startTimeout = time + timerMultiplier(this) * m_engine->m_startTimeout;
	    DDebug(this,DebugAll,"Set timeouts start=" FMT64 " setup=" FMT64 " [%p]",
		m_startTimeout,m_setupTimeout,this);
	    if (m_xmlDom) {
		Lock lck(m_socketMutex);
		if (m_xmlDom) {
		    m_xmlDom->reset();
		    DDebug(this,DebugAll,"XML parser reset [%p]",this);
		}
	    }
	    break;
	case Idle:
	    m_events.clear();
	case Destroy:
	    m_id = "";
	    m_setupTimeout = 0;
	    m_startTimeout = 0;
	    // Reset all internal flags
	    resetFlags(InternalFlags);
	    if (type() == c2s)
		clientStream()->m_registerReq = 0;
	    break;
	case Running:
	    resetConnectStatus();
	    setRedirect();
	    m_redirectCount = 0;
	    m_pingInterval = m_engine->m_pingInterval;
	    setNextPing(true);
	    setFlags(StreamSecured | StreamAuthenticated);
	    resetFlags(InError);
	    m_setupTimeout = 0;
	    m_startTimeout = 0;
	    if (m_state != Running)
		m_events.append(new JBEvent(JBEvent::Running,this,0));
	    break;
	case Securing:
	    socketSetCanRead(false);
	    break;
	default: ;
    }
    m_state = newState;
    if (m_state == Running)
	setIdleTimer(time);
}

// Check if the stream compress flag is set and compression was offered by remote party
XmlElement* JBStream::checkCompress()
{
    if (flag(StreamCompressed) || !flag(Compress))
	return 0;
    XMPPFeatureCompress* c = m_features.getCompress();
    if (!c)
	return 0;
    if (!(c && c->methods()))
	return 0;
    XmlElement* x = 0;
    Lock lock(m_socketMutex);
    m_engine->compressStream(this,c->methods());
    if (m_compress && m_compress->format()) {
	x = XMPPUtils::createElement(XmlTag::Compress,XMPPNamespace::Compress);
	x->addChild(XMPPUtils::createElement(XmlTag::Method,m_compress->format()));
    }
    else
	TelEngine::destruct(m_compress);
    return x;
}

// Check for pending events. Set the last event
void JBStream::checkPendingEvent()
{
    if (m_lastEvent)
	return;
    if (!m_terminateEvent) {
	GenObject* gen = m_events.remove(false);
	if (gen)
	    m_lastEvent = static_cast<JBEvent*>(gen);
	return;
    }
    // Check for register events and raise them before the terminate event
    for (ObjList* o = m_events.skipNull(); o; o = o->skipNext()) {
	JBEvent* ev = static_cast<JBEvent*>(o->get());
	if (ev->type() == JBEvent::RegisterOk || ev->type() == JBEvent::RegisterFailed) {
	    m_lastEvent = ev;
	    m_events.remove(ev,false);
	    return;
	}
    }
    m_lastEvent = m_terminateEvent;
    m_terminateEvent = 0;
}

// Send pending stream XML or stanzas
bool JBStream::sendPending(bool streamOnly)
{
    if (!m_socket)
	return false;
    XDebug(this,DebugAll,"JBStream::sendPending() [%p]",this);
    bool noComp = !flag(StreamCompressed);
    // Always try to send pending stream XML first
    if (m_outStreamXml) {
	const void* buf = 0;
	unsigned int len = 0;
	if (noComp) {
	    buf = m_outStreamXml.c_str();
	    len = m_outStreamXml.length();
	}
	else {
	    buf = m_outStreamXmlCompress.data();
	    len = m_outStreamXmlCompress.length();
	}
	if (!writeSocket(buf,len))
	    return false;
	bool all = false;
	if (noComp) {
	    all = (len == m_outStreamXml.length());
	    if (all)
		m_outStreamXml.clear();
	    else
		m_outStreamXml = m_outStreamXml.substr(len);
	}
	else {
	    all = (len == m_outStreamXmlCompress.length());
	    if (all) {
		m_outStreamXml.clear();
		m_outStreamXmlCompress.clear();
	    }
	    else
		m_outStreamXmlCompress.cut(-(int)len);
	}
	// Start TLS now for incoming streams
	if (m_incoming && m_state == Securing) {
	    if (all) {
		m_engine->encryptStream(this);
		setFlags(StreamTls);
		socketSetCanRead(true);
	    }
	    return true;
	}
	// Check set StreamCompressed flag if all data sent
	if (all && flag(SetCompressed))
	    setFlags(StreamCompressed);
	if (streamOnly || !all)
	    return true;
    }

    // Send pending stanzas
    if (m_state != Running || streamOnly)
	return true;
    ObjList* obj = m_pending.skipNull();
    if (!obj)
	return true;
    XmlElementOut* eout = static_cast<XmlElementOut*>(obj->get());
    XmlElement* xml = eout->element();
    if (!xml) {
	m_pending.remove(eout,true);
	return true;
    }
    bool sent = eout->sent();
    const void* buf = 0;
    unsigned int len = 0;
    if (noComp)
	buf = (const void*)eout->getData(len);
    else {
	if (!sent) {
	    // Make sure the buffer is prepared for sending
	    eout->getData(len);
	    m_outXmlCompress.clear();
	    if (!compress(eout))
		return false;
	}
	buf = m_outXmlCompress.data();
	len = m_outXmlCompress.length();
    }
    // Print the element only if it's the first time we try to send it
    if (!sent)
	m_engine->printXml(this,true,*xml);
    if (writeSocket(buf,len)) {
	if (!len)
	    return true;
	setIdleTimer();
	// Adjust element's buffer. Remove it from list on completion
	unsigned int rest = 0;
	if (noComp) {
	    eout->dataSent(len);
	    rest = eout->dataCount();
	}
	else {
	    m_outXmlCompress.cut(-(int)len);
	    rest = m_outXmlCompress.length();
	}
	if (!rest) {
	    DDebug(this,DebugAll,"Sent element (%p,%s) [%p]",xml,xml->tag(),this);
	    m_pending.remove(eout,true);
	}
	else
	    DDebug(this,DebugAll,"Partially sent element (%p,%s) sent=%u rest=%u [%p]",
		xml,xml->tag(),len,rest,this);
	return true;
    }
    // Error
    Debug(this,DebugNote,"Failed to send (%p,%s) [%p]",xml,xml->tag(),this);
    return false;
}

// Write data to socket
bool JBStream::writeSocket(const void* data, unsigned int& len)
{
    if (!(data && len)) {
	len = 0;
	return true;
    }
    Lock lock(m_socketMutex);
    if (!socketCanWrite()) {
	len = 0;
	if (0 != (m_socketFlags & SocketCanWrite)) {
	    socketSetCanWrite(false);
	    postponeTerminate(0,m_incoming,XMPPError::SocketError,"No socket");
	}
	return false;
    }
    socketSetWriting(true);
    lock.drop();
#ifdef JBSTREAM_DEBUG_SOCKET
    if (!flag(StreamCompressed))
	Debug(this,DebugInfo,"Sending %s [%p]",(const char*)data,this);
    else
	Debug(this,DebugInfo,"Sending %u compressed bytes [%p]",len,this);
#endif
    int w = m_socket->writeData(data,len);
    if (w != Socket::socketError())
	len = w;
    else
	len = 0;
#ifdef JBSTREAM_DEBUG_SOCKET
    if (!flag(StreamCompressed)) {
	String sent((const char*)data,len);
	Debug(this,DebugInfo,"Sent %s [%p]",sent.c_str(),this);
    }
    else
	Debug(this,DebugInfo,"Sent %u compressed bytes [%p]",len,this);
#endif
    Lock lck(m_socketMutex);
    // Check if the connection is waiting to be reset
    if (socketWaitReset()) {
	socketSetWriting(false);
	return true;
    }
    // Check if something changed
    if (!(m_socket && socketWriting())) {
	Debug(this,DebugAll,"Socket deleted while writing [%p]",this);
	return true;
    }
    socketSetWriting(false);
    if (w != Socket::socketError() || m_socket->canRetry())
	return true;
    socketSetCanWrite(false);
    String tmp;
    Thread::errorString(tmp,m_socket->error());
    String reason;
    reason << "Socket send error: " << tmp << " (" << m_socket->error() << ")";
    Debug(this,DebugWarn,"%s [%p]",reason.c_str(),this);
    lck.drop();
    postponeTerminate(0,m_incoming,XMPPError::SocketError,reason);
    return false;
}

// Update stream flags and remote connection data from engine
void JBStream::updateFromRemoteDef()
{
    m_engine->lock();
    JBRemoteDomainDef* domain = m_engine->remoteDomainDef(m_remote.domain());
    // Update flags
    setFlags(domain->m_flags & StreamFlags);
    // Update connection data
    if (outgoing() && state() == Idle) {
	m_connectAddr = domain->m_address;
	m_connectPort = domain->m_port;
    }
    m_engine->unlock();
}

// Retrieve the first required feature in the list
XMPPFeature* JBStream::firstRequiredFeature()
{
    for (ObjList* o = m_features.skipNull(); o; o = o->skipNext()) {
	XMPPFeature* f = static_cast<XMPPFeature*>(o->get());
	if (f->required())
	    return f;
    }
    return 0;
}

// Drop (delete) received XML element
bool JBStream::dropXml(XmlElement*& xml, const char* reason)
{
    if (!xml)
	return true;
    Debug(this,DebugNote,"Dropping xml=(%p,%s) ns=%s in state=%s reason='%s' [%p]",
	xml,xml->tag(),TelEngine::c_safe(xml->xmlns()),stateName(),reason,this);
    TelEngine::destruct(xml);
    return true;
}

// Set stream flag mask
void JBStream::setFlags(int mask)
{
#ifdef XDEBUG
    String f;
    XMPPUtils::buildFlags(f,mask,s_flagName);
    Debug(this,DebugAll,"Setting flags 0x%X (%s) current=0x%X [%p]",
	mask,f.c_str(),m_flags,this);
#endif
    m_flags |= mask;
#ifdef DEBUG
    if (0 != (mask & StreamCompressed))
	Debug(this,DebugAll,"Stream is using compression [%p]",this);
#endif
}

// Reset stream flag mask
void JBStream::resetFlags(int mask)
{
#ifdef XDEBUG
    String f;
    XMPPUtils::buildFlags(f,mask,s_flagName);
    Debug(this,DebugAll,"Resetting flags 0x%X (%s) current=0x%X [%p]",
	mask,f.c_str(),m_flags,this);
#endif
    m_flags &= ~mask;
}

// Set the idle timer in Running state
void JBStream::setIdleTimer(u_int64_t msecNow)
{
    // Set only for non c2s in Running state
    if (m_type == c2s || m_type == cluster || m_state != Running ||
	!m_engine->m_idleTimeout)
	return;
    m_idleTimeout = msecNow + m_engine->m_idleTimeout;
    XDebug(this,DebugAll,"Idle timeout set to " FMT64 "ms [%p]",m_idleTimeout,this);
}

// Reset ping data
void JBStream::resetPing()
{
    if (!(m_pingTimeout || m_nextPing))
	return;
    XDebug(this,DebugAll,"Reset ping data [%p]",this);
    m_nextPing = 0;
    m_pingTimeout = 0;
}

// Set the time of the next ping if there is any timeout and we don't have a ping in progress
// @return XmlElement containing the ping to send, 0 if no ping is going to be sent or 'force' is true
XmlElement* JBStream::setNextPing(bool force)
{
    if (!m_pingInterval) {
	resetPing();
	return 0;
    }
    if (m_type != c2s && m_type != comp)
	return 0;
    if (force) {
	m_nextPing = Time::msecNow() + m_pingInterval;
	m_pingTimeout = 0;
	XDebug(this,DebugAll,"Next ping " FMT64U " [%p]",m_nextPing,this);
	return 0;
    }
    XmlElement* ping = 0;
    if (m_nextPing) {
	// Ping still active in engine ?
	Time time;
	if (m_nextPing > time.msec())
	    return 0;
	if (m_engine->m_pingTimeout) {
	    generateIdIndex(m_pingId,"_ping_");
	    ping = buildPing(m_pingId);
	    if (ping)
		m_pingTimeout = time.msec() + m_engine->m_pingTimeout;
	    else
		m_pingTimeout = 0;
	}
	else
	    resetPing();
    }
    if (m_pingInterval)
	m_nextPing = Time::msecNow() + m_pingInterval;
    else
	m_nextPing = 0;
    XDebug(this,DebugAll,"Next ping " FMT64U " ping=%p [%p]",m_nextPing,ping,this);
    return ping;
}

// Process incoming elements in Challenge state
// Return false if stream termination was initiated
bool JBStream::processChallenge(XmlElement* xml, const JabberID& from, const JabberID& to)
{
    int t, n;
    if (!XMPPUtils::getTag(*xml,t,n))
	return dropXml(xml,"failed to retrieve element tag");
    if (n != XMPPNamespace::Sasl)
	return dropXml(xml,"expecting sasl namespace");
    if (t == XmlTag::Abort) {
	TelEngine::destruct(xml);
	TelEngine::destruct(m_sasl);
	XmlElement* rsp = XMPPUtils::createFailure(XMPPNamespace::Sasl,XMPPError::Aborted);
	sendStreamXml(Features,rsp);
	return true;
    }
    if (t != XmlTag::Response) {
	dropXml(xml,"expecting sasl response");
	return true;
    }
    XMPPError::Type error = XMPPError::NoError;
    // Use a while() to set error and break to the end
    while (true) {
	// Decode non empty auth data
	const String& text = xml->getText();
	if (text) {
	    String tmp;
	    if (!decodeBase64(tmp,text,this)) {
		error = XMPPError::IncorrectEnc;
		break;
	    }
	    if (m_sasl && !m_sasl->parseMD5ChallengeRsp(tmp)) {
		error = XMPPError::MalformedRequest;
		break;
	    }
	}
	else if (m_sasl)
	    TelEngine::destruct(m_sasl->m_params);
	break;
    }
    if (error == XMPPError::NoError) {
	changeState(Auth);
	m_events.append(new JBEvent(JBEvent::Auth,this,xml,from,to));
    }
    else {
	Debug(this,DebugNote,"Received bad challenge response error='%s' [%p]",
	    XMPPUtils::s_error[error].c_str(),this);
	XmlElement* failure = XMPPUtils::createFailure(XMPPNamespace::Sasl,error);
	sendStreamXml(Features,failure);
	TelEngine::destruct(xml);
    }
    return true;
}

// Process incoming 'auth' elements qualified by SASL namespace
// Return false if stream termination was initiated
bool JBStream::processSaslAuth(XmlElement* xml, const JabberID& from, const JabberID& to)
{
    if (!xml)
	return true;
    if (!XMPPUtils::isTag(*xml,XmlTag::Auth,XMPPNamespace::Sasl))
	return dropXml(xml,"expecting 'auth' in sasl namespace");
    XMPPFeatureSasl* sasl = m_features.getSasl();
    TelEngine::destruct(m_sasl);
    XMPPError::Type error = XMPPError::NoError;
    const char* mName = xml->attribute("mechanism");
    int mech = XMPPUtils::authMeth(mName);
    // Use a while() to set error and break to the end
    while (true) {
	if (!sasl->mechanism(mech)) {
	    error = XMPPError::InvalidMechanism;
	    break;
	}
	if (mech == XMPPUtils::AuthMD5) {
	    // Ignore auth text: we will challenge the client
	    m_sasl = new SASL(false,m_local.domain());
	    String buf;
	    if (m_sasl->buildMD5Challenge(buf)) {
		XDebug(this,DebugAll,"Sending challenge=%s [%p]",buf.c_str(),this);
		Base64 b((void*)buf.c_str(),buf.length());
		b.encode(buf);
		XmlElement* chg = XMPPUtils::createElement(XmlTag::Challenge,
		    XMPPNamespace::Sasl,buf);
		if (!sendStreamXml(Challenge,chg)) {
		    TelEngine::destruct(xml);
		    return false;
		}
	    }
	    else {
		TelEngine::destruct(m_sasl);
		error = XMPPError::TempAuthFailure;
		break;
	    }
	}
	else if (mech == XMPPUtils::AuthPlain) {
	    // Decode non empty auth data
	    DataBlock d;
	    const String& text = xml->getText();
	    if (text && text != "=" && !decodeBase64(d,text)) {
		error = XMPPError::IncorrectEnc;
		break;
	    }
	    m_sasl = new SASL(true);
	    if (!m_sasl->parsePlain(d)) {
		error = XMPPError::MalformedRequest;
		break;
	    }
	}
	else {
	    // This should never happen: we don't handle a mechanism sent
	    // to the remote party!
	    Debug(this,DebugWarn,"Unhandled advertised auth mechanism='%s' [%p]",
		mName,this);
	    error = XMPPError::TempAuthFailure;
	    break;
	}
	break;
    }
    if (error == XMPPError::NoError) {
	// Challenge state: we've challenged the remote party
	// Otherwise: request auth from upper layer
	if (state() == Challenge)
	    TelEngine::destruct(xml);
	else {
	    changeState(Auth);
	    m_events.append(new JBEvent(JBEvent::Auth,this,xml,from,to));
	}
    }
    else {
	Debug(this,DebugNote,"Received auth request mechanism='%s' error='%s' [%p]",
	    mName,XMPPUtils::s_error[error].c_str(),this);
	XmlElement* failure = XMPPUtils::createFailure(XMPPNamespace::Sasl,error);
	sendStreamXml(m_state,failure);
	TelEngine::destruct(xml);
    }
    return true;
}

// Process received elements in Features state (incoming stream)
// Return false if stream termination was initiated
bool JBStream::processFeaturesIn(XmlElement* xml, const JabberID& from, const JabberID& to)
{
    if (!xml)
	return true;
    const String* t = 0;
    const String* nsName = 0;
    if (!xml->getTag(t,nsName))
	return dropXml(xml,"invalid tag namespace prefix");
    int ns = nsName ? XMPPUtils::s_ns[*nsName] : XMPPNamespace::Count;

    // Component: Waiting for handshake in the stream namespace
    if (type() == comp) {
	if (outgoing())
	    return dropXml(xml,"invalid state for incoming stream");
	if (*t != XMPPUtils::s_tag[XmlTag::Handshake] || ns != m_xmlns)
	    return dropXml(xml,"expecting handshake in stream's namespace");
	JBEvent* ev = new JBEvent(JBEvent::Auth,this,xml,from,to);
	ev->m_text = xml->getText();
	m_events.append(ev);
	changeState(Auth);
	return true;
    }

    XMPPFeature* f = 0;
    // Stream compression feature and compression namespace are not the same!
    if (ns != XMPPNamespace::Compress)
	f = m_features.get(ns);
    else
	f = m_features.get(XMPPNamespace::CompressFeature);

    // Check if received unexpected feature
    if (!f) {
	// Check for some features that can be negotiated via 'iq' elements
	if (m_type == c2s && *t == XMPPUtils::s_tag[XmlTag::Iq] && ns == m_xmlns) {
	    int chTag = XmlTag::Count;
	    int chNs = XMPPNamespace::Count;
	    XmlElement* child = xml->findFirstChild();
	    if (child)
		XMPPUtils::getTag(*child,chTag,chNs);
	    // Bind
	    if (chNs == XMPPNamespace::Bind && m_features.get(XMPPNamespace::Bind)) {
		// We've sent bind feature
		// Don't accept it if not authenticated and TLS/SASL must be negotiated
		if (!flag(StreamAuthenticated)) {
		    XMPPFeature* tls = m_features.get(XMPPNamespace::Tls);
		    if (tls && tls->required()) {
			XmlElement* e = XMPPUtils::createError(xml,XMPPError::TypeWait,
			    XMPPError::EncryptionRequired);
			sendStreamXml(m_state,e);
			return true;
		    }
		    XMPPFeature* sasl = m_features.get(XMPPNamespace::Sasl);
		    XMPPFeature* iqAuth = m_features.get(XMPPNamespace::IqAuth);
		    if ((sasl && sasl->required()) || (iqAuth && iqAuth->required())) {
			XmlElement* e = XMPPUtils::createError(xml,XMPPError::TypeAuth,
			    XMPPError::NotAllowed);
			sendStreamXml(m_state,e);
			return true;
		    }
		}
		// Remove TLS/SASL features from list: they can't be negotiated anymore
		setFlags(StreamSecured | StreamAuthenticated);
		m_features.remove(XMPPNamespace::Tls);
		m_features.remove(XMPPNamespace::Sasl);
		m_features.remove(XMPPNamespace::IqAuth);
		changeState(Running);
		return processRunning(xml,from,to);
	    }
	    else if (chNs == XMPPNamespace::IqRegister) {
		// Register
		m_events.append(new JBEvent(JBEvent::Iq,this,xml,xml->findFirstChild()));
		return true;
	    }
	    else if (chNs == XMPPNamespace::IqAuth) {
		XMPPUtils::IqType type = XMPPUtils::iqType(xml->attribute("type"));
		bool req = type == XMPPUtils::IqGet || type == XMPPUtils::IqSet;
		// Stream non SASL auth
		// Check if we support it
		if (!m_features.get(XMPPNamespace::IqAuth)) {
		    if (req) {
			XmlElement* e = XMPPUtils::createError(xml,XMPPError::TypeCancel,
			    XMPPError::NotAllowed);
			return sendStreamXml(m_state,e);
		    }
		    return dropXml(xml,"unexpected jabber:iq:auth element");
		}
		if (flag(StreamRemoteVer1)) {
		    XMPPFeature* tls = m_features.get(XMPPNamespace::Tls);
		    if (tls && tls->required()) {
			XmlElement* e = XMPPUtils::createError(xml,XMPPError::TypeWait,
			    XMPPError::EncryptionRequired);
			sendStreamXml(m_state,e);
			return true;
		    }
		}
		if (chTag != XmlTag::Query) {
		    if (req) {
			XmlElement* e = XMPPUtils::createError(xml,XMPPError::TypeModify,
			    XMPPError::FeatureNotImpl);
			sendStreamXml(m_state,e);
			return true;
		    }
		    return dropXml(xml,"expecting iq auth with 'query' child");
		}
		// Send it to the uppper layer
		if (type == XMPPUtils::IqSet) {
		    m_events.append(new JBEvent(JBEvent::Auth,this,xml,xml->findFirstChild()));
		    changeState(Auth);
		}
		else
		    m_events.append(new JBEvent(JBEvent::Iq,this,xml,xml->findFirstChild()));
		return true;
	    }
	}
	// s2s waiting for dialback
	if (m_type == s2s) {
	    if (isDbResult(*xml))
		return serverStream()->processDbResult(xml,from,to);
	    // Drop the element if not authenticated
	    if (!flag(StreamAuthenticated))
		return dropXml(xml,"expecting dialback result");
	}
	// Check if all remaining features are optional
	XMPPFeature* req = firstRequiredFeature();
	if (req) {
	    Debug(this,DebugInfo,
		"Received '%s' while having '%s' required feature not negotiated [%p]",
		xml->tag(),req->c_str(),this);
	    // TODO: terminate the stream?
	    return dropXml(xml,"required feature negotiation not completed");
	}
	// No more required features: change state to Running
	// Remove TLS/SASL features from list: they can't be negotiated anymore
	setFlags(StreamSecured | StreamAuthenticated);
	m_features.remove(XMPPNamespace::Tls);
	m_features.remove(XMPPNamespace::Sasl);
	changeState(Running);
	return processRunning(xml,from,to);
    }
    // Stream enchryption
    if (ns == XMPPNamespace::Tls) {
	if (*t != XMPPUtils::s_tag[XmlTag::Starttls])
	    return dropXml(xml,"expecting tls 'starttls' element");
	if (!flag(StreamSecured)) {
	    // Change state before trying to send the element
	    // to signal to sendPending() to enchrypt the stream after sending it
	    changeState(Securing);
	    sendStreamXml(WaitStart,
		XMPPUtils::createElement(XmlTag::Proceed,XMPPNamespace::Tls));
	}
	else {
	    Debug(this,DebugNote,"Received '%s' element while already secured [%p]",
		xml->tag(),this);
	    // We shouldn't have Starttls in features list
	    // Something went wrong: terminate the stream
	    terminate(0,true,xml,XMPPError::Internal,"Stream already secured");
	    return false;
	}
	TelEngine::destruct(xml);
	return true;
    }
    // Stream SASL auth
    if (ns == XMPPNamespace::Sasl) {
	if (*t != XMPPUtils::s_tag[XmlTag::Auth])
	    return dropXml(xml,"expecting sasl 'auth' element");
	if (!flag(StreamAuthenticated)) {
	    // Check if we must negotiate TLS before authentication
	    XMPPFeature* tls = m_features.get(XMPPNamespace::Tls);
	    if (tls) {
		if (!flag(StreamSecured) && tls->required()) {
		    TelEngine::destruct(xml);
		    XmlElement* failure = XMPPUtils::createFailure(XMPPNamespace::Sasl,
			XMPPError::EncryptionRequired);
		    sendStreamXml(m_state,failure);
		    return true;
		}
		setSecured();
	    }
	}
	else {
	    // Remote party requested authentication while already done:
	    // Reset our flag and let it authenticate again
	    Debug(this,DebugNote,
		"Received auth request while already authenticated [%p]",
		this);
	    resetFlags(StreamAuthenticated);
	}
	return processSaslAuth(xml,from,to);
    }
    // Stream compression
    if (ns == XMPPNamespace::Compress) {
	if (*t != XMPPUtils::s_tag[XmlTag::Compress])
	    return dropXml(xml,"expecting stream compression 'compress' element");
	return handleCompressReq(xml);
    }
    return dropXml(xml,"unhandled stream feature");
}

// Process received elements in Features state (outgoing stream)
// Return false if stream termination was initiated
bool JBStream::processFeaturesOut(XmlElement* xml, const JabberID& from,
    const JabberID& to)
{
    if (!xml)
	return true;
    if (!XMPPUtils::isTag(*xml,XmlTag::Features,XMPPNamespace::Stream))
	return dropXml(xml,"expecting stream features");
    m_features.fromStreamFeatures(*xml);
    // Check TLS
    if (!flag(StreamSecured)) {
	XMPPFeature* tls = m_features.get(XMPPNamespace::Tls);
	if (tls) {
	    if (m_engine->hasClientTls()) {
		TelEngine::destruct(xml);
		XmlElement* x = XMPPUtils::createElement(XmlTag::Starttls,
		    XMPPNamespace::Tls);
		return sendStreamXml(WaitTlsRsp,x);
	    }
	    if (tls->required() || flag(TlsRequired))
		return destroyDropXml(xml,XMPPError::Internal,
		    "required encryption not available");
	}
	else if (flag(TlsRequired))
	    return destroyDropXml(xml,XMPPError::EncryptionRequired,
		"required encryption not supported by remote");
	setFlags(StreamSecured);
    }
    // Check auth
    if (!flag(StreamAuthenticated)) {
	JBServerStream* server = serverStream();
	if (server) {
	    TelEngine::destruct(xml);
	    return server->sendDialback();
	}
	JBClientStream* client = clientStream();
	if (client) {
	    // Start auth or request registration data
	    TelEngine::destruct(xml);
	    if (!flag(RegisterUser))
		return client->startAuth();
	    return client->requestRegister(false);
	}
    }
    // Check compression
    XmlElement* x = checkCompress();
    if (x) {
	TelEngine::destruct(xml);
	return sendStreamXml(Compressing,x);
    }
    JBClientStream* client = clientStream();
    if (client) {
	TelEngine::destruct(xml);
	return client->bind();
    }
    JBServerStream* server = serverStream();
    JBClusterStream* cluster = clusterStream();
    if (server || cluster) {
	TelEngine::destruct(xml);
	changeState(Running);
	return true;
    }
    return dropXml(xml,"incomplete features process for outgoing stream");
}

// Process received elements in WaitTlsRsp state (outgoing stream)
// The element will be consumed
// Return false if stream termination was initiated
bool JBStream::processWaitTlsRsp(XmlElement* xml, const JabberID& from,
    const JabberID& to)
{
    if (!xml)
	return true;
    int t,n;
    const char* reason = 0;
    if (XMPPUtils::getTag(*xml,t,n)) {
	if (n == XMPPNamespace::Tls) {
	    // Accept proceed and failure
	    if (t != XmlTag::Proceed && t != XmlTag::Failure)
		reason = "expecting tls 'proceed' or 'failure'";
	}
	else
	    reason = "expecting tls namespace";
    }
    else
	reason = "failed to retrieve element tag";
    if (reason) {
	// TODO: Unacceptable response to starttls request
	// Restart socket read or terminate the stream ?
	socketSetCanRead(true);
	return dropXml(xml,reason);
    }
    if (t == XmlTag::Proceed) {
	TelEngine::destruct(xml);
	changeState(Securing);
	m_engine->encryptStream(this);
	socketSetCanRead(true);
	setFlags(StreamTls);
	XmlElement* s = buildStreamStart();
	return sendStreamXml(WaitStart,s);
    }
    // TODO: Implement TLS usage reset if the stream is going to re-connect
    terminate(1,false,xml,XMPPError::NoError,"Server can't start TLS");
    return false;
}

// Set stream namespace from type
void JBStream::setXmlns()
{
    switch (m_type) {
	case c2s:
	    m_xmlns = XMPPNamespace::Client;
	    break;
	case s2s:
	    m_xmlns = XMPPNamespace::Server;
	    break;
	case comp:
	    m_xmlns = XMPPNamespace::ComponentAccept;
	    break;
	case cluster:
	    m_xmlns = XMPPNamespace::YateCluster;
	    break;
    }
}

// Event termination notification
void JBStream::eventTerminated(const JBEvent* ev)
{
    if (ev && ev == m_lastEvent) {
	m_lastEvent = 0;
	XDebug(this,DebugAll,"Event (%p,%s) terminated [%p]",ev,ev->name(),this);
    }
}

// Compress data to be sent (the pending stream xml buffer or pending stanza)
// Return false on failure
bool JBStream::compress(XmlElementOut* xml)
{
    DataBlock& buf = xml ? m_outXmlCompress : m_outStreamXmlCompress;
    const String& xmlBuf = xml ? xml->buffer() : m_outStreamXml;
    m_socketMutex.lock();
    int res = m_compress ? m_compress->compress(xmlBuf.c_str(),xmlBuf.length(),buf) : -1000;
    m_socketMutex.unlock();
    const char* s = xml ? "pending" : "stream";
    if (res >= 0) {
	if ((unsigned int)res == xmlBuf.length()) {
#ifdef JBSTREAM_DEBUG_COMPRESS
	    Debug(this,DebugInfo,"Compressed %s xml %u --> %u [%p]",
		s,xmlBuf.length(),buf.length(),this);
#endif
	    return true;
	}
	Debug(this,DebugNote,"Partially compressed %s xml %d/%u [%p]",
	    s,res,xmlBuf.length(),this);
    }
    else
	Debug(this,DebugNote,"Failed to compress %s xml: %d [%p]",s,res,this);
    return false;
}

// Reset connect status data
void JBStream::resetConnectStatus()
{
    DDebug(this,DebugAll,"resetConnectStatus() [%p]",this);
    m_connectStatus = JBConnect::Start;
    m_connectSrvs.clear();
}

// Postpone stream terminate until all parsed elements are processed
// Terminate now if allowed
void JBStream::postponeTerminate(int location, bool destroy, int error, const char* reason)
{
    lock();
    XDebug(this,DebugAll,"postponeTerminate(%d,%u,%s,%s) state=%s [%p]",
	location,destroy,XMPPUtils::s_error[error].c_str(),reason,stateName(),this);
    if (!m_ppTerminate) {
	int interval = 0;
	if (type() == c2s)
	    interval = m_engine->m_pptTimeoutC2s;
	else
	    interval = m_engine->m_pptTimeout;
	if (interval && haveData()) {
	    m_ppTerminate = new NamedList("");
	    m_ppTerminate->addParam("location",String(location));
	    m_ppTerminate->addParam("destroy",String::boolText(destroy));
	    m_ppTerminate->addParam("error",String(error));
	    m_ppTerminate->addParam("reason",reason);
	    m_ppTerminateTimeout = Time::msecNow() + interval;
	    Debug(this,DebugInfo,
		"Postponed termination location=%d destroy=%u error=%s reason=%s interval=%us [%p]",
		location,destroy,XMPPUtils::s_error[error].c_str(),reason,interval,this);
	}
    }
    bool postponed = m_ppTerminate != 0;
    unlock();
    if (!postponed)
	terminate(location,destroy,0,error,reason);
}

// Handle postponed termination. Return true if found
bool JBStream::postponedTerminate()
{
    if (!m_ppTerminate)
	return false;
    int location = m_ppTerminate->getIntValue("location");
    bool destroy = m_ppTerminate->getBoolValue("destroy");
    int error = m_ppTerminate->getIntValue("error");
    String reason = m_ppTerminate->getValue("reason");
    resetPostponedTerminate();
    DDebug(this,DebugAll,"postponedTerminate(%d,%u,%s,%s) state=%s [%p]",
	location,destroy,XMPPUtils::s_error[error].c_str(),reason.c_str(),
	stateName(),this);
    terminate(location,destroy,0,error,reason);
    return true;
}

// Reset redirect data
void JBStream::setRedirect(const String& addr, int port)
{
    if (!addr) {
	if (m_redirectAddr)
	    Debug(this,DebugInfo,"Cleared redirect data [%p]",this);
	m_redirectAddr = "";
	m_redirectPort = 0;
	return;
    }
    if (m_redirectCount >= m_redirectMax) {
	setRedirect();
	return;
    }
    resetConnectStatus();
    m_redirectAddr = addr;
    m_redirectPort = port;
    m_redirectCount++;
    Debug(this,DebugInfo,"Set redirect to '%s:%d' in state %s (counter=%u max=%u) [%p]",
	m_redirectAddr.c_str(),m_redirectPort,stateName(),m_redirectCount,m_redirectMax,this);
}


/*
 * JBClientStream
 */
JBClientStream::JBClientStream(JBEngine* engine, Socket* socket, bool ssl)
    : JBStream(engine,socket,c2s,ssl),
    m_userData(0), m_registerReq(0)
{
}

JBClientStream::JBClientStream(JBEngine* engine, const JabberID& jid, const String& account,
    const NamedList& params, const char* name, const char* serverHost)
    : JBStream(engine,c2s,jid,jid.domain(),TelEngine::null(name) ? account.c_str() : name,
	&params,serverHost),
    m_account(account), m_userData(0), m_registerReq(0)
{
    m_password = params.getValue("password");
}

// Build a ping iq stanza
XmlElement* JBClientStream::buildPing(const String& stanzaId)
{
    return XMPPUtils::createPing(stanzaId);
}

// Bind a resource to an incoming stream
void JBClientStream::bind(const String& resource, const char* id, XMPPError::Type error)
{
    DDebug(this,DebugAll,"bind(%s,'%s') [%p]",resource.c_str(),
	XMPPUtils::s_error[error].c_str(),this);
    Lock lock(this);
    if (!incoming() || m_remote.resource())
	return;
    XmlElement* xml = 0;
    if (resource) {
	m_remote.resource(resource);
	xml = XMPPUtils::createIq(XMPPUtils::IqResult,0,0,id);
	XmlElement* bind = XMPPUtils::createElement(XmlTag::Bind,
	    XMPPNamespace::Bind);
	bind->addChild(XMPPUtils::createElement(XmlTag::Jid,m_remote));
	xml->addChild(bind);
    }
    else {
	if (error == XMPPError::NoError)
	    error = XMPPError::NotAllowed;
	xml = XMPPUtils::createError(XMPPError::TypeModify,error);
    }
    // Remove non-negotiable bind feature on success
    if (sendStanza(xml) && resource)
	m_features.remove(XMPPNamespace::Bind);
}

// Request account setup (or info) on outgoing stream
bool JBClientStream::requestRegister(bool data, bool set, const String& newPass)
{
    if (incoming())
	return true;

    Lock lock(this);
    DDebug(this,DebugAll,"requestRegister(%u,%u) [%p]",data,set,this);
    XmlElement* req = 0;
    if (data) {
	// Register new user, change the account or remove it
	if (set) {
	    // TODO: Allow user account register/change through unsecured streams ?
	    String* pass = 0;
	    if (!flag(StreamAuthenticated))
		pass = &m_password;
	    else if (newPass) {
		m_newPassword = newPass;
		pass = &m_newPassword;
	    }
	    if (!pass)
		return false;
	    m_registerReq = '2';
	    req = XMPPUtils::createRegisterQuery(0,0,String(m_registerReq),
		m_local.node(),*pass);
	}
	else if (flag(StreamAuthenticated)) {
	    m_registerReq = '3';
	    req = XMPPUtils::createRegisterQuery(XMPPUtils::IqSet,0,0,
		String(m_registerReq),XMPPUtils::createElement(XmlTag::Remove));
	}
	else
	    return false;
    }
    else {
	// Request register info
	m_registerReq = '1';
	req = XMPPUtils::createRegisterQuery(XMPPUtils::IqGet,0,0,String(m_registerReq));
    }
    if (!flag(StreamAuthenticated) || state() != Running)
	return sendStreamXml(Register,req);
    return sendStanza(req);
}

// Process elements in Running state
bool JBClientStream::processRunning(XmlElement* xml, const JabberID& from, const JabberID& to)
{
    if (!xml)
	return true;
    // Check if a resource was bound to an incoming stream
    // Accept only 'iq' with bind namespace only if we've sent 'bind' feature
    if (incoming()) {
	if (!m_remote.resource()) {
	    if (XMPPUtils::isTag(*xml,XmlTag::Iq,m_xmlns)) {
		XmlElement* child = XMPPUtils::findFirstChild(*xml,XmlTag::Bind,XMPPNamespace::Bind);
		if (child && m_features.get(XMPPNamespace::Bind)) {
		    m_events.append(new JBEvent(JBEvent::Bind,this,xml,from,to,child));
		    return true;
		}
	    }
	    XmlElement* e = XMPPUtils::createError(xml,XMPPError::TypeCancel,
		XMPPError::NotAllowed,"No resource bound to the stream");
	    sendStanza(e);
	    return true;
	}
    }
    else if (m_registerReq && XMPPUtils::isTag(*xml,XmlTag::Iq,m_xmlns) &&
	isRegisterId(*xml) && XMPPUtils::isResponse(*xml))
	return processRegister(xml,from,to);
    return JBStream::processRunning(xml,from,to);
}

// Process received elements in WaitStart state
// WaitStart: Incoming: waiting for stream start
//            Outgoing: idem (our stream start was already sent)
// Return false if stream termination was initiated
bool JBClientStream::processStart(const XmlElement* xml, const JabberID& from,
    const JabberID& to)
{
    XDebug(this,DebugAll,"JBClientStream::processStart(%s) [%p]",xml->tag(),this);

    // Check element
    if (!processStreamStart(xml))
	return false;

    // RFC3920 5.3.1:
    // The 'from' attribute must be set for response stream start
    if (outgoing()) {
	if (from.null()) {
	    Debug(this,DebugNote,"Received '%s' with empty 'from' [%p]",xml->tag(),this);
	    terminate(0,false,0,XMPPError::BadAddressing,"Missing 'from' attribute");
	    return false;
	}
    }
    else {
	if (!flag(StreamAuthenticated)) {
	    m_remote.set(from);
	    m_local.set(to);
        }
    }
    m_remote.resource("");
    // RFC3920 5.3.1: The 'to' attribute must always be set
    // RFC3920: The 'to' attribute is optional
    bool validTo = !to.null();
    if (validTo) {
	if (outgoing())
	    validTo = (m_local.bare() == to);
	else
	    validTo = engine()->hasDomain(to.domain());
    }
#ifdef RFC3920
    else
	validTo = outgoing();
#endif
    if (!validTo) {
	Debug(this,DebugNote,"Received '%s' with invalid to='%s' [%p]",
	    xml->tag(),to.c_str(),this);
	terminate(0,false,0,
	    to.null() ? XMPPError::BadAddressing : XMPPError::HostUnknown,
	    "Invalid 'to' attribute");
	return false;
    }
    if (incoming() || flag(StreamRemoteVer1)) {
	m_events.append(new JBEvent(JBEvent::Start,this,0,from,to));
	return true;
    }
    Debug(this,DebugNote,"Outgoing client stream: unsupported remote version (expecting 1.x)");
    terminate(0,true,0,XMPPError::Internal,"Unsupported version");
    return false;
}

// Process elements in Auth state
bool JBClientStream::processAuth(XmlElement* xml, const JabberID& from,
    const JabberID& to)
{
    if (!xml)
	return true;
    if (incoming())
	return destroyDropXml(xml,XMPPError::Internal,"invalid state for incoming stream");
    int t,n;
    if (!XMPPUtils::getTag(*xml,t,n))
	return destroyDropXml(xml,XMPPError::Internal,"failed to retrieve element tag");

    // Authenticating
    if (!flag(StreamAuthenticated)) {
	// TODO: The server might challenge us again
	//       Implement support for multiple challenge/response steps
	if (n != XMPPNamespace::Sasl)
	    return destroyDropXml(xml,XMPPError::InvalidNamespace,
		"element with non SASL namespace");
	if (!m_sasl)
	    return destroyDropXml(xml,XMPPError::Internal,"no SASL data");
	if (t == XmlTag::Failure) {
	    terminate(0,true,xml);
	    return false;
	}
	if (!m_sasl->m_plain) {
	    // Digest MD5
	    if (flag(StreamWaitChallenge)) {
		if (t != XmlTag::Challenge)
		    return destroyDropXml(xml,XMPPError::BadRequest,"expecting challenge");
		String tmp;
		if (!decodeBase64(tmp,xml->getText(),this))
		    return destroyDropXml(xml,XMPPError::IncorrectEnc,
			"challenge with incorrect encoding");
		if (!m_sasl->parseMD5Challenge(tmp))
		    return destroyDropXml(xml,XMPPError::MalformedRequest,
			"invalid challenge format");
		TelEngine::destruct(xml);
		m_sasl->setAuthParams(m_local.node(),m_password);
		tmp.clear();
		m_sasl->buildAuthRsp(tmp,"xmpp/" + m_local.domain());
		resetFlags(StreamWaitChallenge);
		setFlags(StreamWaitChgRsp);
		XmlElement* rsp = XMPPUtils::createElement(XmlTag::Response,XMPPNamespace::Sasl,tmp);
		return sendStreamXml(state(),rsp);
	    }
	    // Digest MD5 response reply
	    if (flag(StreamWaitChgRsp)) {
#ifdef RFC3920
		// Expect success or challenge
		// challenge is accepted if not already received one
		if (t != XmlTag::Success && (t != XmlTag::Challenge || flag(StreamRfc3920Chg)))
#else
		// Expect success
		if (t != XmlTag::Success)
#endif
		    return dropXml(xml,"unexpected element");
		if (!flag(StreamRfc3920Chg)) {
		    String rspAuth;
		    if (!decodeBase64(rspAuth,xml->getText(),this))
			return destroyDropXml(xml,XMPPError::IncorrectEnc,
			    "challenge response reply with incorrect encoding");
		    if (!rspAuth.startSkip("rspauth=",false))
			return destroyDropXml(xml,XMPPError::BadFormat,
			    "invalid challenge response reply");
		    if (!m_sasl->validAuthReply(rspAuth))
			return destroyDropXml(xml,XMPPError::InvalidAuth,
			    "incorrect challenge response reply auth");
		}
#ifdef RFC3920
		// Send empty response to challenge
		if (t == XmlTag::Challenge) {
		    setFlags(StreamRfc3920Chg);
		    TelEngine::destruct(xml);
		    XmlElement* rsp = XMPPUtils::createElement(XmlTag::Response,
			XMPPNamespace::Sasl);
		    return sendStreamXml(state(),rsp);
		}
#endif
		resetFlags(StreamWaitChgRsp | StreamRfc3920Chg);
	    }
	    else
		return dropXml(xml,"unhandled sasl digest md5 state");
	}
	else {
	    // Plain
	    if (t != XmlTag::Success)
		return dropXml(xml,"unexpected element");
	}
	// Authenticated. Bind a resource
	Debug(this,DebugAll,"Authenticated [%p]",this);
	TelEngine::destruct(xml);
	TelEngine::destruct(m_sasl);
	setFlags(StreamAuthenticated);
	XmlElement* start = buildStreamStart();
	return sendStreamXml(WaitStart,start);
    }

    XMPPUtils::IqType iq = XMPPUtils::iqType(xml->attribute("type"));
    String* id = xml->getAttribute("id");

    // Waiting for bind response
    if (flag(StreamWaitBindRsp)) {
	// Expecting 'iq' result or error
	if (t != XmlTag::Iq ||
	    (iq != XMPPUtils::IqResult && iq != XMPPUtils::IqError) ||
	    !id || *id != "bind_1")
	    return dropXml(xml,"unexpected element");
	if (iq == XMPPUtils::IqError) {
	    Debug(this,DebugNote,"Resource binding failed [%p]",this);
	    terminate(0,true,xml);
	    return false;
	}
	// Check it
	bool ok = false;
	while (true) {
	    XmlElement* bind = XMPPUtils::findFirstChild(*xml,XmlTag::Bind,XMPPNamespace::Bind);
	    if (!bind)
		break;
	    XmlElement* tmp = bind->findFirstChild(&XMPPUtils::s_tag[XmlTag::Jid]);
	    if (!tmp)
		break;
	    JabberID jid(tmp->getText());
	    if (jid.bare() != m_local.bare())
		break;
	    ok = true;
	    if (m_local.resource() != jid.resource()) {
		m_local.resource(jid.resource());
		Debug(this,DebugAll,"Resource set to '%s' [%p]",
		    local().resource().c_str(),this);
	    }
	    break;
	}
	if (!ok)
	    return destroyDropXml(xml,XMPPError::UndefinedCondition,
		"unacceptable bind response");
	resetFlags(StreamWaitBindRsp);
	TelEngine::destruct(xml);
	if (!m_features.get(XMPPNamespace::Session)) {
	    changeState(Running);
	    return true;
	}
	// Send session
	XmlElement* sess = XMPPUtils::createIq(XMPPUtils::IqSet,0,0,"sess_1");
	sess->addChild(XMPPUtils::createElement(XmlTag::Session,XMPPNamespace::Session));
	setFlags(StreamWaitSessRsp);
	return sendStreamXml(state(),sess);
    }

    // Waiting for session response
    if (flag(StreamWaitSessRsp)) {
	// Expecting 'iq' result or error
	if (t != XmlTag::Iq ||
	    (iq != XMPPUtils::IqResult && iq != XMPPUtils::IqError) ||
	    !id || *id != "sess_1")
	    return dropXml(xml,"unexpected element");
	if (iq == XMPPUtils::IqError) {
	    Debug(this,DebugNote,"Session failed [%p]",this);
	    terminate(0,true,xml);
	    return false;
	}
	TelEngine::destruct(xml);
	resetFlags(StreamWaitSessRsp);
	changeState(Running);
	return true;
    }

    return dropXml(xml,"unhandled");
}

// Process elements in Register state
bool JBClientStream::processRegister(XmlElement* xml, const JabberID& from,
    const JabberID& to)
{
    if (!xml)
	return true;
    int t, ns;
    if (!XMPPUtils::getTag(*xml,t,ns))
	return dropXml(xml,"failed to retrieve element tag");
    if (t != XmlTag::Iq)
	return dropXml(xml,"expecting 'iq'");
    XMPPUtils::IqType iq = XMPPUtils::iqType(xml->attribute("type"));
    if (iq != XMPPUtils::IqResult && iq != XMPPUtils::IqError)
	return dropXml(xml,"expecting 'iq' response");
    if (!isRegisterId(*xml))
	return dropXml(xml,"unexpected response id");
    if (iq == XMPPUtils::IqError) {
	m_events.append(new JBEvent(JBEvent::RegisterFailed,this,xml,from,to));
	// Don't terminate if the user requested account change after authentication
	if (!flag(StreamAuthenticated))
	    terminate(0,true,0,XMPPError::NoError);
	return flag(StreamAuthenticated);
    }
    // Requested registration data
    if (m_registerReq == '1') {
	// XEP-0077: check for username and password children or
	//  instructions
	XmlElement* query = XMPPUtils::findFirstChild(*xml,XmlTag::Query,
	    XMPPNamespace::IqRegister);
	if (query && XMPPUtils::findFirstChild(*query,XmlTag::Username) &&
	    XMPPUtils::findFirstChild(*query,XmlTag::Password)) {
	    TelEngine::destruct(xml);
	    return requestRegister(true);
	}
	m_events.append(new JBEvent(JBEvent::RegisterFailed,this,xml,from,to));
	// Don't terminate if the user requested account change after authentication
	if (!flag(StreamAuthenticated))
	    terminate(0,true,0,XMPPError::NoError);
	return flag(StreamAuthenticated);
    }
    // Requested registration/change
    if (m_registerReq == '2') {
	m_events.append(new JBEvent(JBEvent::RegisterOk,this,xml,from,to));
	// Reset register user flag
	resetFlags(RegisterUser);
	// Done if account changed after authentication
	if (flag(StreamAuthenticated)) {
	    m_password = m_newPassword;
	    return true;
	}
	// Start auth
	changeState(Features);
	return startAuth();
    }
    // Requested account removal
    if (m_registerReq == '3') {
	terminate(0,true,xml,XMPPError::Reg,"Account removed");
	return false;
    }
    return destroyDropXml(xml,XMPPError::Internal,"unhandled state");
}

// Release memory
void JBClientStream::destroyed()
{
    userData(0);
    JBStream::destroyed();
}

// Start outgoing stream authentication
bool JBClientStream::startAuth()
{
    if (incoming() || state() != Features)
	return false;

    TelEngine::destruct(m_sasl);

    XMPPFeatureSasl* sasl = m_features.getSasl();
    if (!sasl) {
	terminate(0,true,0,XMPPError::NoError,"Missing authentication data");
	return false;
    }

    // RFC 3920 SASL auth
    int mech = XMPPUtils::AuthNone;
    if (sasl->mechanism(XMPPUtils::AuthMD5))
	mech = XMPPUtils::AuthMD5;
    else if (sasl->mechanism(XMPPUtils::AuthPlain) && flag(AllowPlainAuth))
	mech = XMPPUtils::AuthPlain;
    else {
	terminate(0,true,0,XMPPError::NoError,"Unsupported authentication mechanism");
	return false;
    }

    m_sasl = new SASL(mech == XMPPUtils::AuthPlain);
    String rsp;
    if (m_sasl->m_plain) {
	m_sasl->setAuthParams(m_local.node(),m_password);
	if (!m_sasl->buildAuthRsp(rsp)) {
	    terminate(0,true,0,XMPPError::NoError,"Invalid auth data length for plain auth");
	    return false;
	}
    }
    else
	setFlags(StreamWaitChallenge);
    // MD5: send auth element, wait challenge
    // Plain auth: send auth element with credentials and wait response (success/failure)
    XmlElement* xml = XMPPUtils::createElement(XmlTag::Auth,XMPPNamespace::Sasl,rsp);
    xml->setAttribute("mechanism",lookup(mech,XMPPUtils::s_authMeth));
    return sendStreamXml(Auth,xml);
}

// Start resource binding on outgoing stream
bool JBClientStream::bind()
{
    Debug(this,DebugAll,"Binding resource [%p]",this);
    XmlElement* bind = XMPPUtils::createElement(XmlTag::Bind,XMPPNamespace::Bind);
    if (m_local.resource())
	bind->addChild(XMPPUtils::createElement(XmlTag::Resource,m_local.resource()));
    XmlElement* b = XMPPUtils::createIq(XMPPUtils::IqSet,0,0,"bind_1");
    b->addChild(bind);
    setFlags(StreamWaitBindRsp);
    return sendStreamXml(Auth,b);
}


/*
 * JBServerStream
 */
// Build an incoming stream from a socket
JBServerStream::JBServerStream(JBEngine* engine, Socket* socket, bool component)
    : JBStream(engine,socket,component ? comp : s2s),
    m_remoteDomains(""), m_dbKey(0)
{
}

// Build an outgoing stream
JBServerStream::JBServerStream(JBEngine* engine, const JabberID& local,
    const JabberID& remote, const char* dbId, const char* dbKey, bool dbOnly,
    const NamedList* params)
    : JBStream(engine,s2s,local,remote,0,params),
    m_remoteDomains(""), m_dbKey(0)
{
    if (!(TelEngine::null(dbId) || TelEngine::null(dbKey)))
	m_dbKey = new NamedString(dbId,dbKey);
    if (dbOnly)
	setFlags(DialbackOnly | NoAutoRestart);
}

// Constructor. Build an outgoing component stream
JBServerStream::JBServerStream(JBEngine* engine, const JabberID& local, const JabberID& remote,
    const String* name, const NamedList* params)
    : JBStream(engine,comp,local,remote,name ? name->c_str() : 0,params),
    m_remoteDomains(""), m_dbKey(0)
{
    if (params)
	m_password = params->getValue("password");
}

// Send a dialback verify response
bool JBServerStream::sendDbVerify(const char* from, const char* to, const char* id,
    XMPPError::Type rsp)
{
    adjustDbRsp(rsp);
    XmlElement* result = XMPPUtils::createDialbackVerifyRsp(from,to,id,rsp);
    DDebug(this,DebugAll,"Sending '%s' db:verify response from %s to %s [%p]",
	result->attribute("type"),from,to,this);
    return state() < Running ? sendStreamXml(state(),result) : sendStanza(result);
}

// Send a dialback key response. Update the remote domains list
// Terminate the stream if there are no more remote domains
bool JBServerStream::sendDbResult(const JabberID& from, const JabberID& to, XMPPError::Type rsp)
{
    Lock lock(this);
    // Check local domain
    if (m_local != from)
	return false;
    // Respond only to received requests
    NamedString* p = m_remoteDomains.getParam(to);
    if (!p)
	return false;
    bool valid = rsp == XMPPError::NoError;
    // Don't deny already authenticated requests
    if (p->null() && !valid)
	return false;
    // Set request state or remove it if not accepted
    if (valid)
	p->clear();
    else
	m_remoteDomains.clearParam(to);
    bool ok = false;
    adjustDbRsp(rsp);
    XmlElement* result = XMPPUtils::createDialbackResult(from,to,rsp);
    DDebug(this,DebugAll,"Sending '%s' db:result response from %s to %s [%p]",
	result->attribute("type"),from.c_str(),to.c_str(),this);
    if (m_state < Running) {
	ok = sendStreamXml(Running,result);
	// Remove features and set the authenticated flag
	if (ok && valid) {
	    m_features.remove(XMPPNamespace::Sasl);
	    m_features.remove(XMPPNamespace::IqAuth);
	    setFlags(StreamAuthenticated);
	    // Compression can still be set
	    if (!flag(StreamCompressed) && m_features.get(XMPPNamespace::CompressFeature))
		setFlags(StreamCanCompress);
	    else
		resetFlags(StreamCanCompress);
	}
    }
    else if (m_state == Running)
	ok = sendStanza(result);
    else
	TelEngine::destruct(result);
    // Terminate the stream if there are no more remote domains
    if (!m_remoteDomains.count())
	terminate(-1,true,0,rsp);
    return ok;
}

// Send dialback data (key/verify)
bool JBServerStream::sendDialback()
{
    State newState = Running;
    XmlElement* result = 0;
    if (!flag(DialbackOnly)) {
	if (flag(StreamAuthenticated))
	    newState = Running;
	else {
	    String key;
	    engine()->buildDialbackKey(id(),m_local,m_remote,key);
	    result = XMPPUtils::createDialbackKey(m_local,m_remote,key);
	    newState = Auth;
	}
    }
    else if (!m_dbKey) {
	// Dialback only with no key?
	Debug(this,DebugNote,"Outgoing dialback stream with no key! [%p]",this);
	terminate(0,true,0,XMPPError::Internal);
	return false;
    }
    if (m_dbKey) {
	XmlElement* db = XMPPUtils::createDialbackVerify(m_local,m_remote,
	    m_dbKey->name(),*m_dbKey);
	if (result)
	    return sendStreamXml(newState,result,db);
	return sendStreamXml(newState,db);
    }
    if (result)
	return sendStreamXml(newState,result);
    changeState(newState);
    return true;
}

// Release memory
void JBServerStream::destroyed()
{
    TelEngine::destruct(m_dbKey);
    JBStream::destroyed();
}

// Process elements in Running state
bool JBServerStream::processRunning(XmlElement* xml, const JabberID& from,
    const JabberID& to)
{
    if (!xml)
	return true;
    // Incoming, authenticated stream which might still request compression
    // Any other element will reset compression offer
    if (flag(StreamCanCompress)) {
	if (incoming() && !flag(StreamCompressed) &&
	    m_features.get(XMPPNamespace::CompressFeature)) {
	    int t = XmlTag::Count;
	    int n = XMPPNamespace::Count;
	    XMPPUtils::getTag(*xml,t,n);
	    if (t == XmlTag::Compress && n == XMPPNamespace::Compress)
		return handleCompressReq(xml);
	}
	resetFlags(StreamCanCompress);
	m_features.remove(XMPPNamespace::CompressFeature);
    }
    // Check the tags of known dialback elements:
    //  there are servers who don't stamp them with the namespace
    // Let other elements stamped with dialback namespace go the upper layer
    if (type() != comp && isDbResult(*xml)) {
	if (outgoing())
	    return dropXml(xml,"dialback result on outgoing stream");
	return processDbResult(xml,from,to);
    }
    // Call default handler
    return JBStream::processRunning(xml,from,to);
}

// Build a stream start XML element
XmlElement* JBServerStream::buildStreamStart()
{
    XmlElement* start = new XmlElement(XMPPUtils::s_tag[XmlTag::Stream],false);
    if (incoming())
	start->setAttribute("id",m_id);
    XMPPUtils::setStreamXmlns(*start);
    start->setAttribute(XmlElement::s_ns,XMPPUtils::s_ns[m_xmlns]);
    if (type() == s2s) {
	start->setAttribute(XmlElement::s_nsPrefix + "db",XMPPUtils::s_ns[XMPPNamespace::Dialback]);
	if (!dialback()) {
	    start->setAttributeValid("from",m_local.bare());
	    start->setAttributeValid("to",m_remote.bare());
	    if (outgoing() || flag(StreamLocalVer1))
		start->setAttribute("version","1.0");
	    start->setAttribute("xml:lang","en");
	}
    }
    else if (type() == comp) {
	if (incoming())
	    start->setAttributeValid("from",m_remote.domain());
	else
	    start->setAttributeValid("to",m_local.domain());
    }
    return start;
}

// Process received elements in WaitStart state
// WaitStart: Incoming: waiting for stream start
//            Outgoing: idem (our stream start was already sent)
// Return false if stream termination was initiated
bool JBServerStream::processStart(const XmlElement* xml, const JabberID& from,
    const JabberID& to)
{
    XDebug(this,DebugAll,"JBServerStream::processStart() [%p]",this);

    if (!processStreamStart(xml))
	return false;

    if (type() == comp) {
	String from = xml->attribute("from");
	if (m_local == from) {
	    changeState(Starting);
	    m_events.append(new JBEvent(JBEvent::Start,this,0,to,JabberID::empty()));
	}
	else
	    terminate(0,false,0,XMPPError::InvalidFrom);
	return false;
    }

    if (outgoing()) {
	m_events.append(new JBEvent(JBEvent::Start,this,0,from,to));
	return true;
    }

    // Incoming stream
    m_local = to;
    if (m_local && !engine()->hasDomain(m_local)) {
	terminate(0,true,0,XMPPError::HostUnknown);
	return false;
    }
    updateFromRemoteDef();
    m_events.append(new JBEvent(JBEvent::Start,this,0,from,to));
    return true;
}

// Process elements in Auth state
bool JBServerStream::processAuth(XmlElement* xml, const JabberID& from,
    const JabberID& to)
{
    if (incoming())
	return dropXml(xml,"invalid state for incoming stream");
    // Component
    if (type() == comp) {
	int t,n;
	if (!XMPPUtils::getTag(*xml,t,n))
	    return destroyDropXml(xml,XMPPError::Internal,"failed to retrieve element tag");
	if (t != XmlTag::Handshake || n != m_xmlns)
	    return dropXml(xml,"expecting handshake in stream's namespace");
	// Stream authenticated
	TelEngine::destruct(xml);
	setFlags(StreamAuthenticated);
	changeState(Running);
	Debug(this,DebugAll,"Authenticated [%p]",this);
	return true;
    }
    // Waiting for db:result
    if (!isDbResult(*xml))
	return dropXml(xml,"expecting dialback result");
    // Result
    // Outgoing stream waiting for dialback key response
    if (outgoing()) {
	if (m_remote != from || m_local != to)
	    return destroyDropXml(xml,XMPPError::BadAddressing,
		"dialback response with invalid 'from'");
	// Expect dialback key response
	int rsp = XMPPUtils::decodeDbRsp(xml);
	if (rsp != XMPPError::NoError) {
	    terminate(1,false,xml,rsp);
	    return false;
	}
	// Stream authenticated
	TelEngine::destruct(xml);
	setFlags(StreamAuthenticated);
	// Check compression
	XmlElement* x = checkCompress();
	if (x)
	    return sendStreamXml(Compressing,x);
	changeState(Running);
	return true;
    }
    return dropXml(xml,"incomplete state process");
}

// Start the stream (reply to received stream start)
bool JBServerStream::startComp(const String& local, const String& remote)
{
    if (state() != Starting || type() != comp)
	return false;
    Lock lock(this);
    XmlElement* s = 0;
    if (incoming()) {
	m_local.set(local);
	m_remote.set(remote);
	s = buildStreamStart();
    }
    else {
	String digest;
	buildSha1Digest(digest,m_password);
	s = XMPPUtils::createElement(XmlTag::Handshake,digest);
    }
    setSecured();
    return sendStreamXml(incoming() ? Features : Auth,s);
}

// Process dialback key (db:result) requests
bool JBServerStream::processDbResult(XmlElement* xml, const JabberID& from,
    const JabberID& to)
{
    // Check TLS when stream:features were sent
    if (m_state == Features) {
	if (flag(TlsRequired) && !flag(StreamSecured))
	    return destroyDropXml(xml,XMPPError::EncryptionRequired,
		"required encryption not supported by remote");
	// TLS can't be negotiated anymore
	setFlags(StreamSecured);
    }
    // Check remote domain
    if (!from)
	return destroyDropXml(xml,XMPPError::BadAddressing,
	    "dialback result with empty 'from' domain");
    // Accept non empty key only
    const char* key = xml->getText();
    if (TelEngine::null(key))
	return destroyDropXml(xml,XMPPError::NotAcceptable,
	    "dialback result with empty key");
    // Check local domain
    if (!(to && engine()->hasDomain(to))) {
	const char* reason = "dialback result with unknown 'to' domain";
	dropXml(xml,reason);
	XmlElement* rsp = XMPPUtils::createDialbackResult(to,from,XMPPError::ItemNotFound);
	if (m_state < Running)
	    sendStreamXml(state(),rsp);
	else
	    sendStanza(rsp);
	return false;
    }
    if (!m_local)
	m_local = to;
    else if (m_local != to)
	return destroyDropXml(xml,XMPPError::NotAcceptable,
	    "dialback result with incorrect 'to' domain");
    // Ignore duplicate requests
    if (m_remoteDomains.getParam(from)) {
	dropXml(xml,"duplicate dialback key request");
	return false;
    }
    m_remoteDomains.addParam(from,key);
    DDebug(this,DebugAll,"Added db:result request from %s [%p]",from.c_str(),this);
    // Notify the upper layer of incoming request
    JBEvent* ev = new JBEvent(JBEvent::DbResult,this,xml,from,to);
    ev->m_text = key;
    m_events.append(ev);
    return true;
}


/*
 * JBClusterStream
 */
// Build an incoming stream from a socket
JBClusterStream::JBClusterStream(JBEngine* engine, Socket* socket)
    : JBStream(engine,socket,cluster)
{
}

// Build an outgoing stream
JBClusterStream::JBClusterStream(JBEngine* engine, const JabberID& local,
    const JabberID& remote, const NamedList* params)
    : JBStream(engine,cluster,local,remote,0,params)
{
}

// Build a stream start XML element
XmlElement* JBClusterStream::buildStreamStart()
{
    XmlElement* start = new XmlElement(XMPPUtils::s_tag[XmlTag::Stream],false);
    if (incoming())
	start->setAttribute("id",m_id);
    XMPPUtils::setStreamXmlns(*start);
    start->setAttribute(XmlElement::s_ns,XMPPUtils::s_ns[m_xmlns]);
    start->setAttributeValid("from",m_local);
    start->setAttributeValid("to",m_remote);
    start->setAttribute("version","1.0");
    start->setAttribute("xml:lang","en");
    return start;
}

// Process received elements in WaitStart state
// WaitStart: Incoming: waiting for stream start
//            Outgoing: idem (our stream start was already sent)
// Return false if stream termination was initiated
bool JBClusterStream::processStart(const XmlElement* xml, const JabberID& from,
    const JabberID& to)
{
    XDebug(this,DebugAll,"JBClusterStream::processStart() [%p]",this);
    if (!processStreamStart(xml))
	return false;
    // Check from/to
    bool ok = true;
    if (outgoing())
	ok = (m_local == to) && (m_remote == from);
    else {
	if (!m_remote) {
	    m_local = to;
	    m_remote = from;
	    ok = from && to;
	}
	else
	    ok = (m_local == to) && (m_remote == from);
    }
    if (!ok) {
	Debug(this,DebugNote,"Got invalid from='%s' or to='%s' in stream start [%p]",
	    from.c_str(),to.c_str(),this);
	terminate(0,true,0,XMPPError::BadAddressing);
	return false;
    }
    m_events.append(new JBEvent(JBEvent::Start,this,0,m_remote,m_local));
    return true;
}

// Process elements in Running state
bool JBClusterStream::processRunning(XmlElement* xml, const JabberID& from, const JabberID& to)
{
    if (!xml)
	return true;
    int t, ns;
    if (!XMPPUtils::getTag(*xml,t,ns))
	return dropXml(xml,"failed to retrieve element tag");
    JBEvent::Type evType = JBEvent::Unknown;
    XmlElement* child = 0;
    switch (t) {
	case XmlTag::Iq:
	    checkPing(this,xml,m_pingId);
	    evType = JBEvent::Iq;
	    child = xml->findFirstChild();
	    break;
	case XmlTag::Message:
	    evType = JBEvent::Message;
	    break;
	case XmlTag::Presence:
	    evType = JBEvent::Presence;
	    break;
	default: ;
    }
    m_events.append(new JBEvent(evType,this,xml,m_remote,m_local,child));
    return true;
}

/* vi: set ts=8 sw=4 sts=4 noet: */