XrdCl::Stream Class Reference

Stream. More...

#include <XrdClStream.hh>

Collaboration diagram for XrdCl::Stream:

Public Types
enum	StreamStatus {   Disconnected = 0 ,   Connected = 1 ,   Connecting = 2 ,   Error = 3 }
	Status of the stream. More...

Public Member Functions
	Stream (const URL *url, const URL &prefer=URL())
	Constructor. More...
	~Stream ()
	Destructor. More...
bool	CanCollapse (const URL &url)
void	DisableIfEmpty (uint16_t subStream)
	Disables respective uplink if empty. More...
XRootDStatus	EnableLink (PathID &path)
void	Finalize ()
void	ForceConnect ()
	Force connection. More...
void	ForceError (XRootDStatus status, const bool hush, const uint64_t sess)
	Force error. More...
std::shared_ptr< Channel >	GetChannel ()
const std::string &	GetName () const
	Return stream name. More...
const URL *	GetURL () const
	Get the URL. More...
XRootDStatus	Initialize ()
	Initializer. More...
uint16_t	InspectStatusRsp (uint16_t stream, MsgHandler *&incHandler)
MsgHandler *	InstallIncHandler (std::shared_ptr< Message > &msg, uint16_t stream)
void	OnConnect (uint16_t subStream)
	Call back when a message has been reconstructed. More...
void	OnConnectError (uint16_t subStream, XRootDStatus status)
	On connect error. More...
void	OnError (uint16_t subStream, XRootDStatus status)
	On error. More...
void	OnIncoming (uint16_t subStream, std::shared_ptr< Message > msg, uint32_t bytesReceived)
	Call back when a message has been reconstructed. More...
void	OnMessageSent (uint16_t subStream, Message *msg, uint32_t bytesSent)
bool	OnReadTimeout (uint16_t subStream) XRD_WARN_UNUSED_RESULT
	On read timeout. More...
std::pair< Message *, MsgHandler * >	OnReadyToWrite (uint16_t subStream)
bool	OnWriteTimeout (uint16_t subStream) XRD_WARN_UNUSED_RESULT
	On write timeout. More...
Status	Query (uint16_t query, AnyObject &result)
	Query the stream. More...
void	RegisterEventHandler (ChannelEventHandler *handler)
	Register channel event handler. More...
void	RemoveEventHandler (ChannelEventHandler *handler)
	Remove a channel event handler. More...
XRootDStatus	Send (Message *msg, MsgHandler *handler, bool stateful, time_t expires)
	Queue the message for sending. More...
void	SetChannel (std::weak_ptr< Channel > &channel)
	Sets a weak_ptr of our owning Channel. More...
void	SetChannelData (AnyObject *channelData)
	Set the channel data. More...
void	SetIncomingQueue (InQueue *incomingQueue)
	Set the incoming queue. More...
void	SetJobManager (JobManager *jobManager)
	Set job manager. More...
void	SetOnDataConnectHandler (std::shared_ptr< Job > &onConnJob)
	Set the on-connect handler for data streams. More...
void	SetPoller (Poller *poller)
	Set the poller. More...
void	SetTaskManager (TaskManager *taskManager)
	Set task manager. More...
void	SetTransport (TransportHandler *transport)
	Set the transport. More...
void	Tick (time_t now)

Detailed Description

Stream.

Definition at line 169 of file XrdClStream.hh.

Member Enumeration Documentation

StreamStatus

enum XrdCl::Stream::StreamStatus

Status of the stream.

Enumerator
Disconnected	Not connected.
Connected	Connected.
Connecting	In the process of being connected.
Error	Broken.

Definition at line 175 of file XrdClStream.hh.

      {
        Disconnected    = 0,    
        Connected       = 1,    
        Connecting      = 2,    
        Error           = 3     
      };

Constructor & Destructor Documentation

Stream()

XrdCl::Stream::Stream	(	const URL *	url,
		const URL &	prefer = URL()
	)

Constructor.

Definition at line 299 of file XrdClStream.cc.

                                                   :
    pUrl( url ),
    pPrefer( prefer ),
    pTransport( 0 ),
    pPoller( 0 ),
    pTaskManager( 0 ),
    pJobManager( 0 ),
    pIncomingQueue( 0 ),
    pChannelData( 0 ),
    pLastStreamError( 0 ),
    pConnectionCount( 0 ),
    pConnectionInitTime( 0 ),
    pAddressType( Utils::IPAll ),
    pSessionId( 0 ),
    pBytesSent( 0 ),
    pBytesReceived( 0 )
  {
    pConnectionStarted.tv_sec = 0; pConnectionStarted.tv_usec = 0;
    pConnectionDone.tv_sec = 0;    pConnectionDone.tv_usec = 0;
 
    std::ostringstream o;
    o << pUrl->GetHostId();
    pStreamName = o.str();
 
    pConnectionWindow  = Utils::GetIntParameter( *url, "ConnectionWindow",
                                                 DefaultConnectionWindow );
    pConnectionRetry   = Utils::GetIntParameter( *url, "ConnectionRetry",
                                                 DefaultConnectionRetry );
    pStreamErrorWindow = Utils::GetIntParameter( *url, "StreamErrorWindow",
                                                 DefaultStreamErrorWindow );
 
    std::string netStack = Utils::GetStringParameter( *url, "NetworkStack",
                                                      DefaultNetworkStack );
 
    pAddressType = Utils::String2AddressType( netStack );
    if( pAddressType == Utils::AddressType::IPAuto )
    {
      XrdNetUtils::NetProt stacks = XrdNetUtils::NetConfig( XrdNetUtils::NetType::qryINIF );
      if( !( stacks & XrdNetUtils::hasIP64 ) )
      {
        if( stacks & XrdNetUtils::hasIPv4 )
          pAddressType = Utils::AddressType::IPv4;
        else if( stacks & XrdNetUtils::hasIPv6 )
          pAddressType = Utils::AddressType::IPv6;
      }
    }
 
    Log *log = DefaultEnv::GetLog();
    log->Debug( PostMasterMsg, "[%s] Stream parameters: Network Stack: %s, "
                "Connection Window: %d, ConnectionRetry: %d, Stream Error "
                "Window: %d", pStreamName.c_str(), netStack.c_str(),
                pConnectionWindow, pConnectionRetry, pStreamErrorWindow );
  }

References XrdCl::Log::Debug(), XrdCl::DefaultConnectionRetry, XrdCl::DefaultConnectionWindow, XrdCl::DefaultNetworkStack, XrdCl::DefaultStreamErrorWindow, XrdCl::URL::GetHostId(), XrdCl::Utils::GetIntParameter(), XrdCl::DefaultEnv::GetLog(), XrdCl::Utils::GetStringParameter(), XrdNetUtils::hasIP64, XrdNetUtils::hasIPv4, XrdNetUtils::hasIPv6, XrdNetUtils::NetConfig(), XrdCl::PostMasterMsg, and XrdCl::Utils::String2AddressType().

Here is the call graph for this function:

~Stream()

XrdCl::Stream::~Stream

(

)

Destructor.

Definition at line 356 of file XrdClStream.cc.

  {
    // Used to disconnect substreams here, but since we are refernce counted
    // and connected substream hold a count, if we're here they're closed.
 
    Log *log = DefaultEnv::GetLog();
    log->Debug( PostMasterMsg, "[%s] Destroying stream",
                pStreamName.c_str() );
 
    MonitorDisconnection( XRootDStatus() );
 
    SubStreamList::iterator it;
    for( it = pSubStreams.begin(); it != pSubStreams.end(); ++it )
      delete *it;
  }

References XrdCl::Log::Debug(), XrdCl::DefaultEnv::GetLog(), and XrdCl::PostMasterMsg.

Here is the call graph for this function:

Member Function Documentation

CanCollapse()

bool XrdCl::Stream::CanCollapse

(

const URL &

url

)

Returns

: true is this channel can be collapsed using this URL, false otherwise

Definition at line 1395 of file XrdClStream.cc.

  {
    Log *log = DefaultEnv::GetLog();
 
    //--------------------------------------------------------------------------
    // Resolve all the addresses of the host we're supposed to connect to
    //--------------------------------------------------------------------------
    std::vector<XrdNetAddr> prefaddrs;
    XRootDStatus st = Utils::GetHostAddresses( prefaddrs, url, pAddressType );
    if( !st.IsOK() )
    {
      log->Error( PostMasterMsg, "[%s] Unable to resolve IP address for %s."
                  , pStreamName.c_str(), url.GetHostName().c_str() );
      return false;
    }
 
    //--------------------------------------------------------------------------
    // Resolve all the addresses of the alias
    //--------------------------------------------------------------------------
    std::vector<XrdNetAddr> aliasaddrs;
    st = Utils::GetHostAddresses( aliasaddrs, *pUrl, pAddressType );
    if( !st.IsOK() )
    {
      log->Error( PostMasterMsg, "[%s] Unable to resolve IP address for %s."
                  , pStreamName.c_str(), pUrl->GetHostName().c_str() );
      return false;
    }
 
    //--------------------------------------------------------------------------
    // Now check if the preferred host is part of the alias
    //--------------------------------------------------------------------------
    auto itr = prefaddrs.begin();
    for( ; itr != prefaddrs.end() ; ++itr )
    {
      auto itr2 = aliasaddrs.begin();
      for( ; itr2 != aliasaddrs.end() ; ++itr2 )
        if( itr->Same( &*itr2 ) ) return true;
    }
 
    return false;
  }

References XrdCl::Log::Error(), XrdCl::Utils::GetHostAddresses(), XrdCl::URL::GetHostName(), XrdCl::DefaultEnv::GetLog(), XrdCl::Status::IsOK(), and XrdCl::PostMasterMsg.

Here is the call graph for this function:

DisableIfEmpty()

void XrdCl::Stream::DisableIfEmpty

(

uint16_t

subStream

)

Disables respective uplink if empty.

Definition at line 794 of file XrdClStream.cc.

  {
    bool closing;
    StreamMutexHelper scopedLock( pMutex, subStream, closing );
    if( closing ) return;
    Log *log = DefaultEnv::GetLog();
 
    if( pSubStreams[subStream]->outQueue->IsEmpty() )
    {
      log->Dump( PostMasterMsg, "[%s] All messages consumed, disable uplink",
                 pSubStreams[subStream]->socket->GetStreamName().c_str() );
      pSubStreams[subStream]->socket->DisableUplink();
    }
  }

References XrdCl::Log::Dump(), XrdCl::DefaultEnv::GetLog(), and XrdCl::PostMasterMsg.

Referenced by XrdCl::AsyncSocketHandler::OnWrite().

Here is the call graph for this function:

Here is the caller graph for this function:

EnableLink()

XRootDStatus XrdCl::Stream::EnableLink

(

PathID &

path

)

Connect if needed, otherwise make sure that the underlying socket handler gets write readiness events, it will update the path with what it has actually enabled

Definition at line 391 of file XrdClStream.cc.

  {
    StreamMutexHelper scopedLock( pMutex );
 
    //--------------------------------------------------------------------------
    // We are in the process of connecting the main stream, so we do nothing
    // because when the main stream connection is established it will connect
    // all the other streams
    //--------------------------------------------------------------------------
    if( pSubStreams[0]->status == Socket::Connecting )
      return XRootDStatus();
 
    //--------------------------------------------------------------------------
    // The main stream is connected, so we can verify whether we have
    // the up and the down stream connected and ready to handle data.
    // If anything is not right we fall back to stream 0.
    //--------------------------------------------------------------------------
    if( pSubStreams[0]->status == Socket::Connected )
    {
      if( pSubStreams[path.down]->status != Socket::Connected )
        path.down = 0;
 
      if( pSubStreams[path.up]->status == Socket::Disconnected )
      {
        path.up = 0;
        return pSubStreams[0]->socket->EnableUplink();
      }
 
      if( pSubStreams[path.up]->status == Socket::Connected )
        return pSubStreams[path.up]->socket->EnableUplink();
 
      return XRootDStatus();
    }
 
    //--------------------------------------------------------------------------
    // The main stream is not connected, we need to check whether enough time
    // has passed since we last encountered an error (if any) so that we could
    // re-attempt the connection
    //--------------------------------------------------------------------------
    Log *log = DefaultEnv::GetLog();
    time_t now = ::time(0);
 
    if( now-pLastStreamError < pStreamErrorWindow )
      return pLastFatalError;
 
    gettimeofday( &pConnectionStarted, 0 );
    ++pConnectionCount;
 
    //--------------------------------------------------------------------------
    // Resolve all the addresses of the host we're supposed to connect to
    //--------------------------------------------------------------------------
    XRootDStatus st = Utils::GetHostAddresses( pAddresses, *pUrl, pAddressType );
    if( !st.IsOK() )
    {
      log->Error( PostMasterMsg, "[%s] Unable to resolve IP address for "
                  "the host", pStreamName.c_str() );
      pLastStreamError = now;
      st.status        = stFatal;
      pLastFatalError  = st;
      return st;
    }
 
    if( pPrefer.IsValid() )
    {
      std::vector<XrdNetAddr> addrresses;
      XRootDStatus st = Utils::GetHostAddresses( addrresses, pPrefer, pAddressType );
      if( !st.IsOK() )
      {
        log->Error( PostMasterMsg, "[%s] Unable to resolve IP address for %s",
                    pStreamName.c_str(), pPrefer.GetHostName().c_str() );
      }
      else
      {
        std::vector<XrdNetAddr> tmp;
        tmp.reserve( pAddresses.size() );
        // first add all remaining addresses
        auto itr = pAddresses.begin();
        for( ; itr != pAddresses.end() ; ++itr )
        {
          if( !HasNetAddr( *itr, addrresses ) )
            tmp.push_back( *itr );
        }
        // then copy all 'preferred' addresses
        std::copy( addrresses.begin(), addrresses.end(), std::back_inserter( tmp ) );
        // and keep the result
        pAddresses.swap( tmp );
      }
    }
 
    Utils::LogHostAddresses( log, PostMasterMsg, pUrl->GetHostId(),
                             pAddresses );
 
    while( !pAddresses.empty() )
    {
      pSubStreams[0]->socket->SetAddress( pAddresses.back() );
      pAddresses.pop_back();
      pConnectionInitTime = ::time( 0 );
      st = pSubStreams[0]->socket->Connect( pConnectionWindow );
      if( st.IsOK() )
      {
        pSubStreams[0]->status = Socket::Connecting;
        break;
      }
    }
    return st;
  }

References XrdCl::Socket::Connected, XrdCl::Socket::Connecting, XrdCl::Socket::Disconnected, XrdCl::PathID::down, XrdCl::Log::Error(), XrdCl::Utils::GetHostAddresses(), XrdCl::URL::GetHostId(), XrdCl::URL::GetHostName(), XrdCl::DefaultEnv::GetLog(), XrdCl::Status::IsOK(), XrdCl::URL::IsValid(), XrdCl::Utils::LogHostAddresses(), XrdCl::PostMasterMsg, XrdCl::Status::status, XrdCl::stFatal, and XrdCl::PathID::up.

Referenced by ForceConnect(), OnConnectError(), OnError(), and Send().

Here is the call graph for this function:

Here is the caller graph for this function:

Finalize()

void XrdCl::Stream::Finalize

(

)

Used at finalize time, disconnects the stream. Assumes poller and jobmanager are not running.

Definition at line 568 of file XrdClStream.cc.

  {
    auto channel = GetChannel();
    StreamMutexHelper scopedLock( pMutex );
    SubStreamList::iterator it;
    for( it = pSubStreams.begin(); it != pSubStreams.end(); ++it )
    {
      (*it)->socket->Close();
      (*it)->status = Socket::Disconnected;
    }
    pSessionId = 0;
  }

References XrdCl::Socket::Disconnected, and GetChannel().

Here is the call graph for this function:

ForceConnect()

void XrdCl::Stream::ForceConnect

(

)

Force connection.

Definition at line 552 of file XrdClStream.cc.

  {
    StreamMutexHelper scopedLock( pMutex );
    if( pSubStreams[0]->status == Socket::Connecting )
    {
      pSubStreams[0]->status = Socket::Disconnected;
      XrdCl::PathID path( 0, 0 );
      XrdCl::XRootDStatus st = EnableLink( path );
      if( !st.IsOK() )
        OnConnectError( 0, st );
    }
  }

References XrdCl::Socket::Connecting, XrdCl::Socket::Disconnected, EnableLink(), XrdCl::Status::IsOK(), and OnConnectError().

Here is the call graph for this function:

ForceError()

void XrdCl::Stream::ForceError	(	XRootDStatus	status,
		const bool	hush,
		const uint64_t	sess
	)

Force error.

Definition at line 1141 of file XrdClStream.cc.

  {
    auto channel = GetChannel();
    bool closing;
    StreamMutexHelper scopedLock( pMutex,
      [this, channel, status, hush, sess]()
      {
        this->ForceError(status, hush, sess);
      }, closing );
    if( closing ) return;
    if( sess && sess != pSessionId ) return;
 
    Log    *log = DefaultEnv::GetLog();
    for( size_t substream = 0; substream < pSubStreams.size(); ++substream )
    {
      if( pSubStreams[substream]->status != Socket::Connected ) continue;
      SockHandlerClose( substream );
 
      if( !hush )
        log->Debug( PostMasterMsg, "[%s] Forcing error on disconnect: %s.",
                    pStreamName.c_str(), status.ToString().c_str() );
 
      //--------------------------------------------------------------------
      // Reinsert the stuff that we have failed to sent
      //--------------------------------------------------------------------
      Reinsert( substream );
    }
 
    pConnectionCount = 0;
    pSessionId = 0;
 
    //------------------------------------------------------------------------
    // We're done here, unlock the stream mutex to avoid deadlocks and
    // report the disconnection event to the handlers
    //------------------------------------------------------------------------
    log->Debug( PostMasterMsg, "[%s] Reporting disconnection to queued "
                "message handlers.", pStreamName.c_str() );
 
    SubStreamList::iterator it;
    OutQueue q;
    for( it = pSubStreams.begin(); it != pSubStreams.end(); ++it )
      q.GrabItems( *(*it)->outQueue );
    scopedLock.UnLock();
 
    q.Report( status );
 
    pIncomingQueue->ReportStreamEvent( MsgHandler::Broken, status );
    pChannelEvHandlers.ReportEvent( ChannelEventHandler::StreamBroken, status );
  }

References XrdCl::MsgHandler::Broken, XrdCl::Socket::Connected, XrdCl::Log::Debug(), GetChannel(), XrdCl::DefaultEnv::GetLog(), XrdCl::OutQueue::GrabItems(), XrdCl::PostMasterMsg, XrdCl::OutQueue::Report(), XrdCl::ChannelHandlerList::ReportEvent(), XrdCl::InQueue::ReportStreamEvent(), XrdCl::ChannelEventHandler::StreamBroken, and XrdCl::Status::ToString().

Referenced by XrdCl::AsyncSocketHandler::OnHeaderCorruption().

Here is the call graph for this function:

Here is the caller graph for this function:

GetChannel()

std::shared_ptr<Channel> XrdCl::Stream::GetChannel ( )

inline

Gets a shared_ptr of our owning Channel. Used to by our AsyncSocketHandlers to obtain a ref count, and internally to ensure Channel (and thus ourselves) remains alive during error-event handlers that close sockets.

Definition at line 414 of file XrdClStream.hh.

      {
        return pChannel.lock();
      }

Referenced by XrdCl::AsyncSocketHandler::Connect(), Finalize(), ForceError(), OnConnect(), OnConnectError(), OnError(), and OnReadTimeout().

Here is the caller graph for this function:

GetName()

const std::string& XrdCl::Stream::GetName ( ) const

inline

Return stream name.

Definition at line 297 of file XrdClStream.hh.

      {
        return pStreamName;
      }

GetURL()

const URL* XrdCl::Stream::GetURL ( ) const

inline

Get the URL.

Definition at line 284 of file XrdClStream.hh.

      {
        return pUrl;
      }

Referenced by XrdCl::AsyncSocketHandler::OnConnectionReturn().

Here is the caller graph for this function:

Initialize()

XRootDStatus XrdCl::Stream::Initialize

(

)

Initializer.

Definition at line 375 of file XrdClStream.cc.

  {
    if( !pTransport || !pPoller || !pChannelData )
      return XRootDStatus( stError, errUninitialized );
 
    AsyncSocketHandler *s = new AsyncSocketHandler( *pUrl, pPoller, pTransport,
                                                    pChannelData, 0, this );
    pSubStreams.push_back( new SubStreamData() );
    pSubStreams[0]->socket = s;
    return XRootDStatus();
  }

References XrdCl::errUninitialized, and XrdCl::stError.

InspectStatusRsp()

uint16_t XrdCl::Stream::InspectStatusRsp	(	uint16_t	stream,
		MsgHandler *&	incHandler
	)

In case the message is a kXR_status response it needs further attention

Returns

: a MsgHandler in case we need to read out raw data

Definition at line 1364 of file XrdClStream.cc.

  {
    InMessageHelper &mh = pSubStreams[stream]->inMsgHelper;
    if( !mh.handler )
      return MsgHandler::RemoveHandler;
 
    uint16_t action = mh.handler->InspectStatusRsp();
    mh.action |= action;
 
    if( action & MsgHandler::RemoveHandler )
      pIncomingQueue->RemoveMessageHandler( mh.handler );
 
    if( action & MsgHandler::Raw )
    {
      incHandler = mh.handler;
      return MsgHandler::Raw;
    }
 
    if( action & MsgHandler::Corrupted )
      return MsgHandler::Corrupted;
 
    if( action & MsgHandler::More )
      return MsgHandler::More;
 
    return MsgHandler::None;
  }

References XrdCl::InMessageHelper::action, XrdCl::MsgHandler::Corrupted, XrdCl::InMessageHelper::handler, XrdCl::MsgHandler::InspectStatusRsp(), XrdCl::MsgHandler::More, XrdCl::MsgHandler::None, XrdCl::MsgHandler::Raw, XrdCl::MsgHandler::RemoveHandler, and XrdCl::InQueue::RemoveMessageHandler().

Referenced by XrdCl::AsyncMsgReader::Read().

Here is the call graph for this function:

Here is the caller graph for this function:

InstallIncHandler()

MsgHandler * XrdCl::Stream::InstallIncHandler	(	std::shared_ptr< Message > &	msg,
		uint16_t	stream
	)

Install a message handler for the given message if there is one available, if the handler want's to be called in the raw mode it will be returned, the message ownership flag is returned in any case

Parameters

msg	message header
stream	stream concerned

Returns

a pair containing the handler and ownership flag

Definition at line 1343 of file XrdClStream.cc.

  {
    InMessageHelper &mh = pSubStreams[stream]->inMsgHelper;
    if( !mh.handler )
      mh.handler = pIncomingQueue->GetHandlerForMessage( msg,
                                                         mh.expires,
                                                         mh.action );
 
    if( !mh.handler )
      return nullptr;
 
    if( mh.action & MsgHandler::Raw )
      return mh.handler;
    return nullptr;
  }

References XrdCl::InMessageHelper::action, XrdCl::InMessageHelper::expires, XrdCl::InQueue::GetHandlerForMessage(), XrdCl::InMessageHelper::handler, and XrdCl::MsgHandler::Raw.

Referenced by XrdCl::AsyncMsgReader::Read().

Here is the call graph for this function:

Here is the caller graph for this function:

OnConnect()

void XrdCl::Stream::OnConnect

(

uint16_t

subStream

)

Call back when a message has been reconstructed.

Definition at line 834 of file XrdClStream.cc.

  {
    auto channel = GetChannel();
    bool closing;
    StreamMutexHelper scopedLock( pMutex, subStream, closing );
    if( closing ) return;
    pSubStreams[subStream]->status = Socket::Connected;
 
    std::string ipstack( pSubStreams[0]->socket->GetIpStack() );
    Log *log = DefaultEnv::GetLog();
    log->Debug( PostMasterMsg, "[%s] Stream %d connected (%s).", pStreamName.c_str(),
                subStream, ipstack.c_str() );
 
    if( subStream == 0 )
    {
      pLastStreamError = 0;
      pLastFatalError  = XRootDStatus();
      pConnectionCount = 0;
      uint16_t numSub = pTransport->SubStreamNumber( *pChannelData );
      pSessionId = ++sSessCntGen;
 
      //------------------------------------------------------------------------
      // Create the streams if they don't exist yet
      //------------------------------------------------------------------------
      if( pSubStreams.size() == 1 && numSub > 1 )
      {
        for( uint16_t i = 1; i < numSub; ++i )
        {
          URL url = pTransport->GetBindPreference( *pUrl, *pChannelData );
          AsyncSocketHandler *s = new AsyncSocketHandler( url, pPoller, pTransport,
                                                          pChannelData, i, this );
          pSubStreams.push_back( new SubStreamData() );
          pSubStreams[i]->socket = s;
        }
      }
 
      //------------------------------------------------------------------------
      // Connect the extra streams, if we fail we move all the outgoing items
      // to stream 0, we don't need to enable the uplink here, because it
      // should be already enabled after the handshaking process is completed.
      //------------------------------------------------------------------------
      if( pSubStreams.size() > 1 )
      {
        log->Debug( PostMasterMsg, "[%s] Attempting to connect %zu additional streams.",
                    pStreamName.c_str(), pSubStreams.size() - 1 );
        for( size_t i = 1; i < pSubStreams.size(); ++i )
        {
          if( pSubStreams[i]->status != Socket::Disconnected )
          {
            pSubStreams[0]->outQueue->GrabItems( *pSubStreams[i]->outQueue );
            SockHandlerClose( i );
          }
          pSubStreams[i]->socket->SetAddress( pSubStreams[0]->socket->GetAddress() );
          XRootDStatus st = pSubStreams[i]->socket->Connect( pConnectionWindow );
          if( !st.IsOK() )
          {
            pSubStreams[0]->outQueue->GrabItems( *pSubStreams[i]->outQueue );
            SockHandlerClose( i );
          }
          else
          {
            pSubStreams[i]->status = Socket::Connecting;
          }
        }
      }
 
      //------------------------------------------------------------------------
      // Inform monitoring
      //------------------------------------------------------------------------
      pBytesSent     = 0;
      pBytesReceived = 0;
      gettimeofday( &pConnectionDone, 0 );
      Monitor *mon = DefaultEnv::GetMonitor();
      if( mon )
      {
        Monitor::ConnectInfo i;
        i.server  = pUrl->GetHostId();
        i.sTOD    = pConnectionStarted;
        i.eTOD    = pConnectionDone;
        i.streams = pSubStreams.size();
 
        AnyObject    qryResult;
        std::string *qryResponse = nullptr;
        pTransport->Query( TransportQuery::Auth, qryResult, *pChannelData );
        qryResult.Get( qryResponse );
 
        if (qryResponse) {
          i.auth = *qryResponse;
          delete qryResponse;
        } else {
          i.auth = "";
        }
 
        mon->Event( Monitor::EvConnect, &i );
      }
 
      //------------------------------------------------------------------------
      // For every connected control-stream call the global on-connect handler
      //------------------------------------------------------------------------
      XrdCl::DefaultEnv::GetPostMaster()->NotifyConnectHandler( *pUrl );
    }
    else if( pOnDataConnJob )
    {
      //------------------------------------------------------------------------
      // For every connected data-stream call the on-connect handler
      //------------------------------------------------------------------------
      pJobManager->QueueJob( pOnDataConnJob.get(), 0 );
    }
  }

References XrdCl::Monitor::ConnectInfo::auth, XrdCl::TransportQuery::Auth, XrdCl::Socket::Connected, XrdCl::Socket::Connecting, XrdCl::Log::Debug(), XrdCl::Socket::Disconnected, XrdCl::Monitor::ConnectInfo::eTOD, XrdCl::Monitor::EvConnect, XrdCl::Monitor::Event(), XrdCl::AnyObject::Get(), XrdCl::TransportHandler::GetBindPreference(), GetChannel(), XrdCl::URL::GetHostId(), XrdCl::DefaultEnv::GetLog(), XrdCl::DefaultEnv::GetMonitor(), XrdCl::DefaultEnv::GetPostMaster(), XrdCl::Status::IsOK(), XrdCl::PostMaster::NotifyConnectHandler(), XrdCl::PostMasterMsg, XrdCl::TransportHandler::Query(), XrdCl::JobManager::QueueJob(), XrdCl::Monitor::ConnectInfo::server, XrdCl::Monitor::ConnectInfo::sTOD, XrdCl::Monitor::ConnectInfo::streams, and XrdCl::TransportHandler::SubStreamNumber().

Referenced by XrdCl::AsyncSocketHandler::HandShakeNextStep().

Here is the call graph for this function:

Here is the caller graph for this function:

OnConnectError()

void XrdCl::Stream::OnConnectError	(	uint16_t	subStream,
		XRootDStatus	status
	)

On connect error.

Definition at line 947 of file XrdClStream.cc.

  {
    auto channel = GetChannel();
    bool closing;
    StreamMutexHelper scopedLock( pMutex, subStream, closing );
    if( closing ) return;
    Log *log = DefaultEnv::GetLog();
    SockHandlerClose( subStream );
    time_t now = ::time(0);
 
    //--------------------------------------------------------------------------
    // For every connection error call the global connection error handler
    //--------------------------------------------------------------------------
    XrdCl::DefaultEnv::GetPostMaster()->NotifyConnErrHandler( *pUrl, status );
 
    //--------------------------------------------------------------------------
    // If we connected subStream == 0 and cannot connect >0 then we just give
    // up and move the outgoing messages to another queue
    //--------------------------------------------------------------------------
    if( subStream > 0 )
    {
      pSubStreams[0]->outQueue->GrabItems( *pSubStreams[subStream]->outQueue );
      if( pSubStreams[0]->status == Socket::Connected )
      {
        XRootDStatus st = pSubStreams[0]->socket->EnableUplink();
        if( !st.IsOK() )
          OnFatalError( 0, st, scopedLock );
        return;
      }
 
      if( pSubStreams[0]->status == Socket::Connecting )
        return;
 
      OnFatalError( subStream, status, scopedLock );
      return;
    }
 
    //--------------------------------------------------------------------------
    // Check if we still have time to try and do something in the current window
    //--------------------------------------------------------------------------
    time_t elapsed = now-pConnectionInitTime;
    log->Error( PostMasterMsg, "[%s] elapsed = %lld, pConnectionWindow = %d seconds.",
                pStreamName.c_str(), (long long) elapsed, pConnectionWindow );
 
    //------------------------------------------------------------------------
    // If we have some IP addresses left we try them
    //------------------------------------------------------------------------
    if( !pAddresses.empty() )
    {
      XRootDStatus st;
      do
      {
        pSubStreams[0]->socket->SetAddress( pAddresses.back() );
        pAddresses.pop_back();
        pConnectionInitTime = ::time( 0 );
        st = pSubStreams[0]->socket->Connect( pConnectionWindow );
      }
      while( !pAddresses.empty() && !st.IsOK() );
 
      if( !st.IsOK() )
        OnFatalError( subStream, st, scopedLock );
      else
        pSubStreams[0]->status = Socket::Connecting;
 
      return;
    }
    //------------------------------------------------------------------------
    // If we still can retry with the same host name, we sleep until the end
    // of the connection window and try
    //------------------------------------------------------------------------
    else if( elapsed < pConnectionWindow && pConnectionCount < pConnectionRetry
             && !status.IsFatal() )
    {
      log->Info( PostMasterMsg, "[%s] Attempting reconnection in %lld seconds.",
                 pStreamName.c_str(), (long long) (pConnectionWindow - elapsed) );
 
      pSubStreams[0]->status = Socket::Connecting;
      Task *task = new ::StreamConnectorTask( *pUrl, pStreamName );
      pTaskManager->RegisterTask( task, pConnectionInitTime+pConnectionWindow );
      return;
    }
    //--------------------------------------------------------------------------
    // We are out of the connection window, the only thing we can do here
    // is re-resolving the host name and retrying if we still can
    //--------------------------------------------------------------------------
    else if( pConnectionCount < pConnectionRetry && !status.IsFatal() )
    {
      pAddresses.clear();
      pSubStreams[0]->status = Socket::Disconnected;
      PathID path( 0, 0 );
      XRootDStatus st = EnableLink( path );
      if( !st.IsOK() )
        OnFatalError( subStream, st, scopedLock );
      return;
    }
 
    //--------------------------------------------------------------------------
    // Else, we fail
    //--------------------------------------------------------------------------
    OnFatalError( subStream, status, scopedLock );
  }

References XrdCl::Socket::Connected, XrdCl::Socket::Connecting, XrdCl::Socket::Disconnected, EnableLink(), XrdCl::Log::Error(), GetChannel(), XrdCl::DefaultEnv::GetLog(), XrdCl::DefaultEnv::GetPostMaster(), XrdCl::Log::Info(), XrdCl::Status::IsFatal(), XrdCl::Status::IsOK(), XrdCl::PostMaster::NotifyConnErrHandler(), XrdCl::PostMasterMsg, and XrdCl::TaskManager::RegisterTask().

Referenced by ForceConnect(), XrdCl::AsyncSocketHandler::OnConnectionReturn(), and XrdCl::AsyncSocketHandler::OnFaultWhileHandshaking().

Here is the call graph for this function:

Here is the caller graph for this function:

OnError()

void XrdCl::Stream::OnError	(	uint16_t	subStream,
		XRootDStatus	status
	)

On error.

Definition at line 1052 of file XrdClStream.cc.

  {
    auto channel = GetChannel();
    bool closing;
    StreamMutexHelper scopedLock( pMutex, subStream, closing );
    if( closing ) return;
    Log *log = DefaultEnv::GetLog();
    SockHandlerClose( subStream );
 
    log->Debug( PostMasterMsg, "[%s] Recovering error for stream #%d: %s.",
                pStreamName.c_str(), subStream, status.ToString().c_str() );
 
    //--------------------------------------------------------------------------
    // Reinsert the stuff that we have failed to sent
    //--------------------------------------------------------------------------
    Reinsert( subStream );
 
    //--------------------------------------------------------------------------
    // We are dealing with an error of a peripheral stream. If we don't have
    // anything to send don't bother recovering. Otherwise move the requests
    // to stream 0 if possible.
    //--------------------------------------------------------------------------
    if( subStream > 0 )
    {
      if( pSubStreams[subStream]->outQueue->IsEmpty() )
        return;
 
      if( pSubStreams[0]->status != Socket::Disconnected )
      {
        pSubStreams[0]->outQueue->GrabItems( *pSubStreams[subStream]->outQueue );
        if( pSubStreams[0]->status == Socket::Connected )
        {
          XRootDStatus st = pSubStreams[0]->socket->EnableUplink();
          if( !st.IsOK() )
            OnFatalError( 0, st, scopedLock );
          return;
        }
      }
      OnFatalError( subStream, status, scopedLock );
      return;
    }
 
    //--------------------------------------------------------------------------
    // If we lost the stream 0 we have lost the session, we re-enable the
    // stream if we still have things in one of the outgoing queues, otherwise
    // there is not point to recover at this point.
    //--------------------------------------------------------------------------
    if( subStream == 0 )
    {
      MonitorDisconnection( status );
      pSessionId = 0;
 
      SubStreamList::iterator it;
      size_t outstanding = 0;
      for( it = pSubStreams.begin(); it != pSubStreams.end(); ++it )
        outstanding += (*it)->outQueue->GetSizeStateless();
 
      if( outstanding )
      {
        PathID path( 0, 0 );
        XRootDStatus st = EnableLink( path );
        if( !st.IsOK() )
        {
          OnFatalError( 0, st, scopedLock );
          return;
        }
      }
 
      //------------------------------------------------------------------------
      // We're done here, unlock the stream mutex to avoid deadlocks and
      // report the disconnection event to the handlers
      //------------------------------------------------------------------------
      log->Debug( PostMasterMsg, "[%s] Reporting disconnection to queued "
                  "message handlers.", pStreamName.c_str() );
      OutQueue q;
      for( it = pSubStreams.begin(); it != pSubStreams.end(); ++it )
        q.GrabStateful( *(*it)->outQueue );
      scopedLock.UnLock();
 
      q.Report( status );
      pIncomingQueue->ReportStreamEvent( MsgHandler::Broken, status );
      pChannelEvHandlers.ReportEvent( ChannelEventHandler::StreamBroken, status );
      return;
    }
  }

References XrdCl::MsgHandler::Broken, XrdCl::Socket::Connected, XrdCl::Log::Debug(), XrdCl::Socket::Disconnected, EnableLink(), GetChannel(), XrdCl::DefaultEnv::GetLog(), XrdCl::OutQueue::GrabStateful(), XrdCl::Status::IsOK(), XrdCl::PostMasterMsg, XrdCl::OutQueue::Report(), XrdCl::ChannelHandlerList::ReportEvent(), XrdCl::InQueue::ReportStreamEvent(), XrdCl::ChannelEventHandler::StreamBroken, XrdCl::Status::ToString(), and XrdCl::StreamMutexHelper::UnLock().

Referenced by XrdCl::AsyncSocketHandler::OnFault(), and OnReadTimeout().

Here is the call graph for this function:

Here is the caller graph for this function:

OnIncoming()

void XrdCl::Stream::OnIncoming	(	uint16_t	subStream,
		std::shared_ptr< Message >	msg,
		uint32_t	bytesReceived
	)

Call back when a message has been reconstructed.

Definition at line 678 of file XrdClStream.cc.

  {
    msg->SetSessionId( pSessionId );
    pBytesReceived += bytesReceived;
 
    MsgHandler *handler = nullptr;
    uint16_t action = 0;
    {
      InMessageHelper &mh = pSubStreams[subStream]->inMsgHelper;
      handler = mh.handler;
      action = mh.action;
      mh.Reset();
    }
 
    if( !IsPartial( *msg ) )
    {
      uint32_t streamAction = pTransport->MessageReceived( *msg, subStream,
                                                           *pChannelData );
      if( streamAction & TransportHandler::DigestMsg )
        return;
 
      if( streamAction & TransportHandler::RequestClose )
      {
        RequestClose( *msg );
        return;
      }
    }
 
    Log *log = DefaultEnv::GetLog();
 
    //--------------------------------------------------------------------------
    // No handler, we discard the message ...
    //--------------------------------------------------------------------------
    if( !handler )
    {
      ServerResponse *rsp = (ServerResponse*)msg->GetBuffer();
      log->Warning( PostMasterMsg, "[%s] Discarding received message: %p "
                    "(status=%d, SID=[%d,%d]), no MsgHandler found.",
                    pStreamName.c_str(), (void*)msg.get(), rsp->hdr.status,
                    rsp->hdr.streamid[0], rsp->hdr.streamid[1] );
      return;
    }
 
    //--------------------------------------------------------------------------
    // We have a handler, so we call the callback
    //--------------------------------------------------------------------------
    log->Dump( PostMasterMsg, "[%s] Handling received message: %p.",
               pStreamName.c_str(), (void*)msg.get() );
 
    if( action & (MsgHandler::NoProcess|MsgHandler::Ignore) )
    {
      log->Dump( PostMasterMsg, "[%s] Ignoring the processing handler for: %s.",
                 pStreamName.c_str(), msg->GetObfuscatedDescription().c_str() );
 
      // if we are handling partial response we have to take down the timeout fence
      if( IsPartial( *msg ) )
      {
        XRootDMsgHandler *xrdHandler = dynamic_cast<XRootDMsgHandler*>( handler );
        if( xrdHandler ) xrdHandler->PartialReceived();
      }
 
      return;
    }
 
    Job *job = new HandleIncMsgJob( handler );
    pJobManager->QueueJob( job );
  }

References XrdCl::InMessageHelper::action, XrdCl::TransportHandler::DigestMsg, XrdCl::Log::Dump(), XrdCl::DefaultEnv::GetLog(), XrdCl::InMessageHelper::handler, ServerResponse::hdr, XrdCl::MsgHandler::Ignore, XrdCl::TransportHandler::MessageReceived(), XrdCl::MsgHandler::NoProcess, XrdCl::XRootDMsgHandler::PartialReceived(), XrdCl::PostMasterMsg, XrdCl::JobManager::QueueJob(), XrdCl::TransportHandler::RequestClose, XrdCl::InMessageHelper::Reset(), ServerResponseHeader::status, ServerResponseHeader::streamid, and XrdCl::Log::Warning().

Referenced by XrdCl::AsyncMsgReader::Read().

Here is the call graph for this function:

Here is the caller graph for this function:

OnMessageSent()

void XrdCl::Stream::OnMessageSent	(	uint16_t	subStream,
		Message *	msg,
		uint32_t	bytesSent
	)

Definition at line 812 of file XrdClStream.cc.

  {
    pTransport->MessageSent( msg, subStream, bytesSent,
                             *pChannelData );
    OutQueue::MsgHelper &h = pSubStreams[subStream]->outMsgHelper;
    pBytesSent += bytesSent;
    if( h.handler )
    {
      // ensure expiration time is assigned if still in queue
      pIncomingQueue->AssignTimeout( h.handler );
      // OnStatusReady may cause the handler to delete itself, in
      // which case the handler or the user callback may also delete msg
      h.handler->OnStatusReady( msg, XRootDStatus() );
    }
    pSubStreams[subStream]->outMsgHelper.Reset();
  }

References XrdCl::InQueue::AssignTimeout(), XrdCl::OutQueue::MsgHelper::handler, XrdCl::TransportHandler::MessageSent(), and XrdCl::MsgHandler::OnStatusReady().

Referenced by XrdCl::AsyncMsgWriter::Write().

Here is the call graph for this function:

Here is the caller graph for this function:

OnReadTimeout()

bool XrdCl::Stream::OnReadTimeout

(

uint16_t

subStream

)

On read timeout.

Definition at line 1248 of file XrdClStream.cc.

  {
    //--------------------------------------------------------------------------
    // We only take the main stream into account
    //--------------------------------------------------------------------------
    if( substream != 0 )
      return true;
 
    //--------------------------------------------------------------------------
    // Check if there is no outgoing messages and if the stream TTL is elapesed.
    // It is assumed that the underlying transport makes sure that there is no
    // pending requests that are not answered, ie. all possible virtual streams
    // are de-allocated
    //--------------------------------------------------------------------------
    Log *log = DefaultEnv::GetLog();
    SubStreamList::iterator it;
    time_t                  now = time(0);
 
    bool closing;
    StreamMutexHelper scopedLock( pMutex, substream, closing );
    if( closing ) return false;
    uint32_t outgoingMessages = 0;
    time_t   lastActivity     = 0;
    for( it = pSubStreams.begin(); it != pSubStreams.end(); ++it )
    {
      outgoingMessages += (*it)->outQueue->GetSize();
      time_t sockLastActivity = (*it)->socket->GetLastActivity();
      if( lastActivity < sockLastActivity )
        lastActivity = sockLastActivity;
    }
 
    if( !outgoingMessages )
    {
      bool disconnect = pTransport->IsStreamTTLElapsed( now-lastActivity,
                                                        *pChannelData );
      if( disconnect )
      {
        log->Debug( PostMasterMsg, "[%s] Stream TTL elapsed, disconnecting...",
                    pStreamName.c_str() );
        const uint64_t sess = pSessionId;
        scopedLock.UnLock();
        //----------------------------------------------------------------------
        // Important note!
        //
        // This destroys the Stream object itself, the underlined
        // AsyncSocketHandler object (that called this method) and the Channel
        // object that aggregates this Stream.
        //----------------------------------------------------------------------
        DefaultEnv::GetPostMaster()->ForceDisconnect( GetChannel(), sess );
        return false;
      }
    }
 
    //--------------------------------------------------------------------------
    // Check if the stream is broken
    //--------------------------------------------------------------------------
    XRootDStatus st = pTransport->IsStreamBroken( now-lastActivity,
                                            *pChannelData );
    if( !st.IsOK() )
    {
      scopedLock.UnLock();
      OnError( substream, st );
      return false;
    }
    return true;
  }

References XrdCl::Log::Debug(), XrdCl::PostMaster::ForceDisconnect(), GetChannel(), XrdCl::DefaultEnv::GetLog(), XrdCl::DefaultEnv::GetPostMaster(), XrdCl::Status::IsOK(), XrdCl::TransportHandler::IsStreamBroken(), XrdCl::TransportHandler::IsStreamTTLElapsed(), OnError(), XrdCl::PostMasterMsg, and XrdCl::StreamMutexHelper::UnLock().

Referenced by XrdCl::AsyncSocketHandler::OnReadTimeout().

Here is the call graph for this function:

Here is the caller graph for this function:

OnReadyToWrite()

std::pair< Message *, MsgHandler * > XrdCl::Stream::OnReadyToWrite

(

uint16_t

subStream

)

Definition at line 752 of file XrdClStream.cc.

  {
    bool closing;
    StreamMutexHelper scopedLock( pMutex, subStream, closing );
    if( closing ) return std::make_pair( (Message *)0, (MsgHandler *)0 );
    Log *log = DefaultEnv::GetLog();
    if( pSubStreams[subStream]->outQueue->IsEmpty() )
    {
      log->Dump( PostMasterMsg, "[%s] Nothing to write, disable uplink",
                 pSubStreams[subStream]->socket->GetStreamName().c_str() );
 
      pSubStreams[subStream]->socket->DisableUplink();
      return std::make_pair( (Message *)0, (MsgHandler *)0 );
    }
 
    OutQueue::MsgHelper &h = pSubStreams[subStream]->outMsgHelper;
    h.msg = pSubStreams[subStream]->outQueue->PopMessage( h.handler,
                                                          h.expires,
                                                          h.stateful );
 
    log->Debug( PostMasterMsg, "[%s] Duplicating MsgHandler: %p (message: %s) "
                "from out-queue to in-queue, starting to send outgoing.",
                pUrl->GetHostId().c_str(), (void*)h.handler,
                h.msg->GetObfuscatedDescription().c_str() );
 
    scopedLock.UnLock();
 
    if( h.handler )
    {
      bool rmMsg = false;
      pIncomingQueue->AddMessageHandler( h.handler, rmMsg );
      if( rmMsg )
      {
        Log *log = DefaultEnv::GetLog();
        log->Warning( PostMasterMsg, "[%s] Removed a leftover msg from the in-queue.",
                      pStreamName.c_str() );
      }
      h.handler->OnReadyToSend( h.msg );
    }
    return std::make_pair( h.msg, h.handler );
  }

References XrdCl::InQueue::AddMessageHandler(), XrdCl::Log::Debug(), XrdCl::Log::Dump(), XrdCl::OutQueue::MsgHelper::expires, XrdCl::URL::GetHostId(), XrdCl::DefaultEnv::GetLog(), XrdCl::Message::GetObfuscatedDescription(), XrdCl::OutQueue::MsgHelper::handler, XrdCl::OutQueue::MsgHelper::msg, XrdCl::MsgHandler::OnReadyToSend(), XrdCl::PostMasterMsg, XrdCl::OutQueue::MsgHelper::stateful, XrdCl::StreamMutexHelper::UnLock(), and XrdCl::Log::Warning().

Referenced by XrdCl::AsyncMsgWriter::Write().

Here is the call graph for this function:

Here is the caller graph for this function:

OnWriteTimeout()

bool XrdCl::Stream::OnWriteTimeout

(

uint16_t

subStream

)

On write timeout.

Definition at line 1318 of file XrdClStream.cc.

  {
    return true;
  }

Referenced by XrdCl::AsyncSocketHandler::OnWriteTimeout().

Here is the caller graph for this function:

Query()

Status XrdCl::Stream::Query	(	uint16_t	query,
		AnyObject &	result
	)

Query the stream.

Definition at line 1440 of file XrdClStream.cc.

  {
    switch( query )
    {
      case StreamQuery::IpAddr:
      {
        result.Set( new std::string( pSubStreams[0]->socket->GetIpAddr() ), false );
        return Status();
      }
 
      case StreamQuery::IpStack:
      {
        result.Set( new std::string( pSubStreams[0]->socket->GetIpStack() ), false );
        return Status();
      }
 
      case StreamQuery::HostName:
      {
        result.Set( new std::string( pSubStreams[0]->socket->GetHostName() ), false );
        return Status();
      }
 
      default:
        return Status( stError, errQueryNotSupported );
    }
  }

References XrdCl::errQueryNotSupported, XrdCl::StreamQuery::HostName, XrdCl::StreamQuery::IpAddr, XrdCl::StreamQuery::IpStack, XrdCl::AnyObject::Set(), and XrdCl::stError.

Here is the call graph for this function:

RegisterEventHandler()

void XrdCl::Stream::RegisterEventHandler

(

ChannelEventHandler *

handler

)

Register channel event handler.

Definition at line 1326 of file XrdClStream.cc.

  {
    pChannelEvHandlers.AddHandler( handler );
  }

References XrdCl::ChannelHandlerList::AddHandler().

Here is the call graph for this function:

RemoveEventHandler()

void XrdCl::Stream::RemoveEventHandler

(

ChannelEventHandler *

handler

)

Remove a channel event handler.

Definition at line 1334 of file XrdClStream.cc.

  {
    pChannelEvHandlers.RemoveHandler( handler );
  }

References XrdCl::ChannelHandlerList::RemoveHandler().

Here is the call graph for this function:

Send()

XRootDStatus XrdCl::Stream::Send	(	Message *	msg,
		MsgHandler *	handler,
		bool	stateful,
		time_t	expires
	)

Queue the message for sending.

Definition at line 501 of file XrdClStream.cc.

  {
    StreamMutexHelper scopedLock( pMutex );
    Log *log = DefaultEnv::GetLog();
 
    //--------------------------------------------------------------------------
    // Check the session ID and bounce if needed
    //--------------------------------------------------------------------------
    if( msg->GetSessionId() &&
        (pSubStreams[0]->status != Socket::Connected ||
        pSessionId != msg->GetSessionId()) )
      return XRootDStatus( stError, errInvalidSession );
 
    //--------------------------------------------------------------------------
    // Decide on the path to send the message
    //--------------------------------------------------------------------------
    PathID path = pTransport->MultiplexSubStream( msg, *pChannelData );
    if( pSubStreams.size() <= path.up )
    {
      log->Warning( PostMasterMsg, "[%s] Unable to send message %s through "
                    "substream %d, using 0 instead", pStreamName.c_str(),
                    msg->GetObfuscatedDescription().c_str(), path.up );
      path.up = 0;
    }
 
    log->Dump( PostMasterMsg, "[%s] Sending message %s (%p) through "
               "substream %d expecting answer at %d", pStreamName.c_str(),
               msg->GetObfuscatedDescription().c_str(), (void*)msg, path.up, path.down );
 
    //--------------------------------------------------------------------------
    // Enable *a* path and insert the message to the right queue
    //--------------------------------------------------------------------------
    XRootDStatus st = EnableLink( path );
    if( st.IsOK() )
    {
      pTransport->MultiplexSubStream( msg, *pChannelData, &path );
      handler->OnWaitingToSend( msg );
      pSubStreams[path.up]->outQueue->PushBack( msg, handler,
                                                expires, stateful );
    }
    else
      st.status = stFatal;
    return st;
  }

References XrdCl::Socket::Connected, XrdCl::PathID::down, XrdCl::Log::Dump(), EnableLink(), XrdCl::errInvalidSession, XrdCl::DefaultEnv::GetLog(), XrdCl::Message::GetObfuscatedDescription(), XrdCl::Message::GetSessionId(), XrdCl::Status::IsOK(), XrdCl::TransportHandler::MultiplexSubStream(), XrdCl::MsgHandler::OnWaitingToSend(), XrdCl::PostMasterMsg, XrdCl::Status::status, XrdCl::stError, XrdCl::stFatal, XrdCl::PathID::up, and XrdCl::Log::Warning().

Here is the call graph for this function:

SetChannel()

void XrdCl::Stream::SetChannel ( std::weak_ptr< Channel > &  channel )

inline

Sets a weak_ptr of our owning Channel.

Definition at line 233 of file XrdClStream.hh.

      {
        pChannel = channel;
      }

SetChannelData()

void XrdCl::Stream::SetChannelData ( AnyObject *  channelData )

inline

Set the channel data.

Definition at line 241 of file XrdClStream.hh.

      {
        pChannelData = channelData;
      }

SetIncomingQueue()

void XrdCl::Stream::SetIncomingQueue ( InQueue *  incomingQueue )

inline

Set the incoming queue.

Definition at line 225 of file XrdClStream.hh.

      {
        pIncomingQueue = incomingQueue;
      }

SetJobManager()

void XrdCl::Stream::SetJobManager ( JobManager *  jobManager )

inline

Set job manager.

Definition at line 257 of file XrdClStream.hh.

      {
        pJobManager = jobManager;
      }

SetOnDataConnectHandler()

void XrdCl::Stream::SetOnDataConnectHandler ( std::shared_ptr< Job > &  onConnJob )

inline

Set the on-connect handler for data streams.

Definition at line 391 of file XrdClStream.hh.

      {
        StreamMutexHelper scopedLock( pMutex );
        pOnDataConnJob = onConnJob;
      }

SetPoller()

void XrdCl::Stream::SetPoller ( Poller *  poller )

inline

Set the poller.

Definition at line 217 of file XrdClStream.hh.

      {
        pPoller = poller;
      }

SetTaskManager()

void XrdCl::Stream::SetTaskManager ( TaskManager *  taskManager )

inline

Set task manager.

Definition at line 249 of file XrdClStream.hh.

      {
        pTaskManager = taskManager;
      }

SetTransport()

void XrdCl::Stream::SetTransport ( TransportHandler *  transport )

inline

Set the transport.

Definition at line 209 of file XrdClStream.hh.

      {
        pTransport = transport;
      }

Tick()

void XrdCl::Stream::Tick

(

time_t

now

)

Handle a clock event generated either by socket timeout, or by the task manager event

Definition at line 584 of file XrdClStream.cc.

  {
    //--------------------------------------------------------------------------
    // Check for timed-out requests and incoming handlers
    //--------------------------------------------------------------------------
    StreamMutexHelper scopedLock( pMutex );
    OutQueue q;
    SubStreamList::iterator it;
    for( it = pSubStreams.begin(); it != pSubStreams.end(); ++it )
      q.GrabExpired( *(*it)->outQueue, now );
    scopedLock.UnLock();
 
    q.Report( XRootDStatus( stError, errOperationExpired ) );
    pIncomingQueue->ReportTimeout( now );
  }

References XrdCl::errOperationExpired, XrdCl::OutQueue::GrabExpired(), XrdCl::OutQueue::Report(), XrdCl::InQueue::ReportTimeout(), XrdCl::stError, and XrdCl::StreamMutexHelper::UnLock().

Here is the call graph for this function:

---

The documentation for this class was generated from the following files:

• XrdClStream.hh
• XrdClStream.cc