c2s.cpp

/*
  Copyright (c) 2008-2009 by Jakob Schroeter <js@camaya.net>
  This file is part of the glooxd library. http://camaya.net/glooxd

  This software is distributed under a license. The full license
  agreement can be found in the file LICENSE in this distribution.
  This software may not be copied, modified, sold or distributed
  other than expressed in the named license agreement.

  This software is distributed without any warranty.
*/


#include "c2s.h"
#include "configmanager.h"
#include "sm.h"
#include "streambase.h"
#include "streamfeature.h"
#include "router.h"

#include <gloox/connectiontcpserver.h>
#include <gloox/connectionbase.h>
#include <gloox/jid.h>
#include <gloox/logsink.h>
#include <gloox/mutexguard.h>
#include <gloox/util.h>

#ifdef _WIN32
#ifndef WIN32_LEAN_AND_MEAN
#define WIN32_LEAN_AND_MEAN
#endif
#include <windows.h>
#else
#include <unistd.h>
#endif

#include <cstdio>

namespace glooxd
{

  C2S::C2S( Router& _router, ConfigManager& _cm, const gloox::LogSink& _logInstance,
            SM& _sm, bool threading,
            const gloox::JID& _name )
    : ComponentBase( _router, _cm, _name, _logInstance ),
      m_sm( _sm ), m_threading( threading ), m_strictXMPP( true ), m_running( true ),
      m_supportRecipientUnavailable( true )
  {
  }

  C2S::~C2S()
  {
    if( m_running )
      handleShutdown();

    gloox::util::clearList( m_servers );
    gloox::util::clearList( m_clients );
//     gloox::util::clearList( m_obsoleteClients );
    printf( "m_obsoleteClients.size(): %d\n", m_obsoleteClients.size() );
    gloox::util::clearList( m_features );
    gloox::util::clearList( m_queue );
  }

  void C2S::run()
  {
    do
    {
      if( m_queueSemaphore.trywait() )
      {
        gloox::Tag* t = m_queue.front();
        m_queueMutex.lock();
        m_queue.pop_front();
        m_queueMutex.unlock();
        handleTag( t );
      }
      recv( 10 );
    } while( m_threading );
  }

  void C2S::handleShutdown()
  {
    m_serverMutex.lock();
    gloox::util::ForEach( m_servers, &gloox::ConnectionTCPServer::disconnect );
    m_serverMutex.unlock();

    gloox::ConnectionError ce = gloox::ConnNoError;
    gloox::StreamError se = gloox::StreamErrorSystemShutdown;
    m_clientMutex.lock();
    gloox::util::ForEach( m_clients, &StreamBase::disconnect, ce, se, gloox::EmptyString );
    m_clientMutex.unlock();

    m_running = false;
    m_threading = false;
    m_queueMutex.lock();
    m_queue.push_back( 0 );
    m_queueMutex.unlock();
  }

  gloox::ConnectionError C2S::recv( int timeout )
  {
    if( !m_clients.size() && !m_servers.size() )
    {
#ifdef _WIN32
      ::Sleep( timeout / 1000 );
#else
      usleep( timeout );
#endif
      return gloox::ConnNoError;
    }

//     printf( "recv: locking\n" );
    m_serverMutex.lock();
    ServerList::const_iterator its = m_servers.begin();
    for( ; its != m_servers.end(); ++its )
      (*its)->recv( timeout );

    m_serverMutex.unlock();
//     printf( "recv: unlocked\n" );

//     printf( "recv: client locking\n" );
    m_clientMutex.lock();
    StreamBaseList::const_iterator itc = m_clients.begin();
    for( ; itc != m_clients.end(); ++itc )
    {
      (*itc)->recv( timeout );
      (*itc)->run();
    }

//     printf( "recv: obsolete client locking\n" );
    m_obsoleteClientMutex.lock();

    itc = m_obsoleteClients.begin();
    for( ; itc != m_obsoleteClients.end(); ++itc )
      m_clients.remove( (*itc) );

    m_clientMutex.unlock();
//     printf( "recv: client unlocked\n" );

    gloox::util::clearList( m_obsoleteClients );
    m_obsoleteClientMutex.unlock();
//     printf( "recv: obsolete client unlocked\n" );

    return gloox::ConnNoError;
  }

  gloox::ConnectionError C2S::receive()
  {
    gloox::ConnectionError ce = gloox::ConnNoError;
    while( ce == gloox::ConnNoError )
      ce = recv( 10 );

    return ce;
  }

  bool C2S::handleC2SInterfaceAdded( const std::string& interface,
                                     int port )
  {
    if( port < 0 )
      return false;

    ServerList::iterator it = m_servers.begin();
    for( ; it != m_servers.end()
             && ( (*it)->server() != gloox::EmptyString || (*it)->port() != port )
             && ( (*it)->server() != interface || (*it)->port() != port ); ++it )
      ;

    if( it == m_servers.end() )
    {

      gloox::ConnectionTCPServer* s = new gloox::ConnectionTCPServer( this, m_logInstance, interface, port );
      if( s->connect() != gloox::ConnNoError )
      {
        m_logInstance.warn( static_cast<gloox::LogArea>( LogAreaC2S ),
                             "Failed to listen on "
                                 + ( interface.empty() ? "*" : interface ) + ":"
                                 + gloox::util::int2string( port ) );
        delete s;
        return false;
      }
      else
      {
        m_logInstance.dbg( static_cast<gloox::LogArea>( LogAreaC2S ),
                            "Listening on " + ( interface.empty() ? "*" : interface ) + ":"
                                + gloox::util::int2string( port ) );

#ifdef DEBUG
        printf( "handleC2SInterfaceAdded 1: locking\n" );
#endif
        m_serverMutex.lock();
        m_servers.push_back( s );
        m_serverMutex.unlock();
#ifdef DEBUG
        printf( "handleC2SInterfaceAdded 1: unlocked\n" );
#endif
      }
    }
    else
    {
      m_logInstance.dbg( static_cast<gloox::LogArea>( LogAreaC2S ),
                         "Already listening on " + ( interface.empty() ? "*" : interface ) + ":"
                             + gloox::util::int2string( port ) );
    }

    if( m_servers.size() == 1 )
      m_cm.handleC2SListening( true );

    return true;
  }

  void C2S::handleDomainAdded( const std::string& domain )
  {
    if( domain.empty() )
      return;

    m_router.registerDomain( this, domain, true );
  }

  void C2S::handleDomainRemoved( const std::string& domain )
  {
    if( domain.empty() )
      return;

    m_router.removeDomain( this, domain, true );
  }

  bool C2S::handleC2SInterfaceRemoved( const std::string& interface,
                                       int port )
  {
    gloox::util::MutexGuard mg( m_serverMutex );
    ServerList::iterator it = m_servers.begin();
    for( ; it != m_servers.end(); ++it )
    {
      if( (*it)->server() == interface && (*it)->port() == port )
      {
        delete (*it);
        m_servers.erase( it );
        if( !m_servers.size() )
          m_cm.handleC2SListening( false );

        return true;
      }
    }

    return false;
  }

  void C2S::disconnect( StreamBase* client )
  {
    if( !client )
      return;

    m_sm.removeSession( client->jid().full() );

#ifdef DEBUG
    printf( "making client obsolete: %s\n", client->jid().full().c_str() );
    printf( "disconnect: obsolete client locking\n" );
#endif
    gloox::util::MutexGuard mg( m_obsoleteClientMutex );
#ifdef DEBUG
    printf( "disconnect: obsolete client locked\n" );
#endif
    m_obsoleteClients.push_back( client );
  }

  void C2S::handleIncomingConnection( gloox::ConnectionBase* /*server*/,
                                      gloox::ConnectionBase* connection )
  {
    if( !connection )
      return;

    m_logInstance.dbg( static_cast<gloox::LogArea>( LogAreaC2S ),
                        "Incoming connection from "
                            + connection->server() + ":"
                            + gloox::util::int2string( connection->port() ) );

    StreamBase* client = new StreamBase( *this, m_cm, connection, m_logInstance );
    client->setStrictXMPP( m_strictXMPP );
    FeatureList fl;
    FeatureList::const_iterator it = m_features.begin();
    for( ; it != m_features.end(); ++it )
    {
      StreamFeature* f = (*it)->newInstance();
      f->setParent( client );
      fl.push_back( f );
    }
    client->setFeatures( fl );

#ifdef DEBUG
    printf( "handleIncomingConnection: client locking\n" );
#endif
    gloox::util::MutexGuard mg( m_clientMutex );
#ifdef DEBUG
    printf( "handleIncomingConnection: client locked\n" );
#endif
    m_clients.push_back( client );
  }

  void C2S::handleIncomingTag( gloox::Tag* tag )
  {
    if( !tag /*|| !tag->hasAttribute( "from" )*/ )
      return;

#ifdef DEBUG
    printf( "C2S::handleIncomingTag()\n" );
    printf( "passing to router directly\n" );
#endif
    m_router.handleIncomingTag( tag );
  }

  void C2S::handleOutgoingTag( gloox::Tag* tag )
  {
    if( !tag )
      return;

    m_queueMutex.lock();
    m_queue.push_back( tag );
    m_queueMutex.unlock();
    m_queueSemaphore.post();
  }

  void C2S::handleTag( gloox::Tag* tag )
  {
    if( !tag )
      return;

#ifdef DEBUG
    printf( "C2S::handleTag()\n" );
#endif

    const gloox::JID& to( tag->findAttribute( "to" ) );

#ifdef DEBUG
    printf( "handleTag: client locking\n" );
#endif
    gloox::util::MutexGuard mg( m_clientMutex );
#ifdef DEBUG
    printf( "handleTag: client locked\n" );
#endif
    StreamBaseList::const_iterator it = m_clients.begin();
    for( ; it != m_clients.end(); ++it )
    {
      if( (*it)->jid().full() == to.full() )
      {
        (*it)->handleOutgoingTag( tag );
        return;
      }
      else if( (*it)->id() == to.resource() )
      {
        (*it)->handleIncomingTag( tag );
        return;
      }
    }

    m_logInstance.warn( static_cast<gloox::LogArea>( LogAreaC2S ),
                        "C2S::handleTag unable to route stanza: "
                            + tag->xml() );

    if( m_supportRecipientUnavailable )
    {
      // NOTE: We will send back a <recipient-unavailable /> error response
      //  if we make it to here, but only on an IQ stanza (since that is the
      //  only one that is waiting for a resposne).  If we did send it on
      //  a presence stanza, we'd end up in an infinite loop as it tries
      //  to send an unavailable presense to the disconnected client.
      if( tag->name() == "iq" )
      {
        // We also could get in an infinite loop if we try to reply back
        // to a sender that is not available. So check the from.
        const gloox::JID& from( tag->findAttribute( "from" ) );
        if( !from.full().empty() )
        {
          StreamBaseList::const_iterator it2 = m_clients.begin();
          for( ; it2 != m_clients.end(); ++it2 )
          {
            if( (*it2)->jid().full() == from.full() )
            {
              gloox::Tag* t = returnError( tag, gloox::StanzaErrorRecipientUnavailable, gloox::StanzaErrorTypeWait );
              handleOutgoingTag( t );
              break;
            }
          }
        }
      }
    }

    delete tag;
  }

}