Loading...
Searching...
No Matches
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.
| ~Stream ()
| Destructor.
bool | CanCollapse (const URL &url)
void | DisableIfEmpty (uint16_t subStream)
| Disables respective uplink if empty.
void | Disconnect (bool force=false)
| Disconnect the stream.
XRootDStatus | EnableLink (PathID &path)
void | ForceConnect ()
| Force connection.
void | ForceError (XRootDStatus status, bool hush=false)
| Force error.
const std::string & | GetName () const
| Return stream name.
const URL * | GetURL () const
| Get the URL.
XRootDStatus | Initialize ()
| Initializer.
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.
void | OnConnectError (uint16_t subStream, XRootDStatus status)
| On connect error.
void | OnError (uint16_t subStream, XRootDStatus status)
| On error.
void | OnIncoming (uint16_t subStream, std::shared_ptr< Message > msg, uint32_t bytesReceived)
| Call back when a message has been reconstructed.
void | OnMessageSent (uint16_t subStream, Message *msg, uint32_t bytesSent)
bool | OnReadTimeout (uint16_t subStream) XRD_WARN_UNUSED_RESULT
| On read timeout.
std::pair< Message *, MsgHandler * > | OnReadyToWrite (uint16_t subStream)
bool | OnWriteTimeout (uint16_t subStream) XRD_WARN_UNUSED_RESULT
| On write timeout.
Status | Query (uint16_t query, AnyObject &result)
| Query the stream.
void | RegisterEventHandler (ChannelEventHandler *handler)
| Register channel event handler.
void | RemoveEventHandler (ChannelEventHandler *handler)
| Remove a channel event handler.
XRootDStatus | Send (Message *msg, MsgHandler *handler, bool stateful, time_t expires)
| Queue the message for sending.
void | SetChannelData (AnyObject *channelData)
| Set the channel data.
void | SetIncomingQueue (InQueue *incomingQueue)
| Set the incoming queue.
void | SetJobManager (JobManager *jobManager)
| Set job manager.
void | SetOnDataConnectHandler (std::shared_ptr< Job > &onConnJob)
| Set the on-connect handler for data streams.
void | SetPoller (Poller *poller)
| Set the poller.
void | SetTaskManager (TaskManager *taskManager)
| Set task manager.
void | SetTransport (TransportHandler *transport)
| Set the transport.
void | Tick (time_t now)

## Detailed Description

Stream.

Definition at line 51 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 57 of file XrdClStream.hh.

{
Disconnected = 0,
Connected = 1,
Connecting = 2,
Error = 3
};
XrdCl::Stream::Disconnected
@ Disconnected
Not connected.
Definition XrdClStream.hh:59
XrdCl::Stream::Error
@ Error
Broken.
Definition XrdClStream.hh:62
XrdCl::Stream::Connected
@ Connected
Connected.
Definition XrdClStream.hh:60
XrdCl::Stream::Connecting
@ Connecting
In the process of being connected.
Definition XrdClStream.hh:61

## Constructor & Destructor Documentation

## Stream()

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

Constructor.

Definition at line 96 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 );
}
XrdCl::DefaultEnv::GetLog
static Log * GetLog()
Get default log.
Definition XrdClDefaultEnv.cc:493
XrdCl::Utils::String2AddressType
static AddressType String2AddressType(const std::string &addressType)
Interpret a string as address type, default to IPAll.
Definition XrdClUtils.cc:123
XrdCl::Utils::GetIntParameter
static int GetIntParameter(const URL &url, const std::string &name, int defaultVal)
Get a parameter either from the environment or URL.
Definition XrdClUtils.cc:81
XrdCl::Utils::GetStringParameter
static std::string GetStringParameter(const URL &url, const std::string &name, const std::string &defaultVal)
Get a parameter either from the environment or URL.
Definition XrdClUtils.cc:104
XrdCl::Utils::IPAll
@ IPAll
Definition XrdClUtils.hh:89
XrdCl::Utils::IPv6
@ IPv6
Definition XrdClUtils.hh:90
XrdCl::Utils::IPv4
@ IPv4
Definition XrdClUtils.hh:91
XrdCl::Utils::IPAuto
@ IPAuto
Definition XrdClUtils.hh:88
XrdNetUtils::qryINIF
@ qryINIF
Only consider internet protocols via ifconfig.
Definition XrdNetUtils.hh:360
XrdNetUtils::NetProt
NetProt
Definition XrdNetUtils.hh:351
XrdNetUtils::hasIP64
@ hasIP64
Definition XrdNetUtils.hh:354
XrdNetUtils::hasIPv6
@ hasIPv6
Definition XrdNetUtils.hh:353
XrdNetUtils::hasIPv4
@ hasIPv4
Definition XrdNetUtils.hh:352
XrdNetUtils::NetConfig
static NetProt NetConfig(NetType netquery=qryINET, const char **eText=0)
Definition XrdNetUtils.cc:716
XrdCl::PostMasterMsg
const uint64_t PostMasterMsg
Definition XrdClConstants.hh:36
XrdCl::DefaultStreamErrorWindow
const int DefaultStreamErrorWindow
Definition XrdClConstants.hh:57
XrdCl::DefaultConnectionRetry
const int DefaultConnectionRetry
Definition XrdClConstants.hh:53
XrdCl::DefaultConnectionWindow
const int DefaultConnectionWindow
Definition XrdClConstants.hh:52
XrdCl::DefaultNetworkStack
const char *const DefaultNetworkStack
Definition XrdClConstants.hh:101
XrdGlobal::Log
XrdSysError Log
Definition XrdConfig.cc:113

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

Here is the call graph for this function:

## ~Stream()

XrdCl::Stream::~Stream | ( |  | ) |

Destructor.

Definition at line 153 of file XrdClStream.cc.

{
Disconnect( true );
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;
}
XrdCl::Stream::Disconnect
void Disconnect(bool force=false)
Disconnect the stream.
Definition XrdClStream.cc:363

References XrdCl::Log::Debug(), Disconnect(), 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 1197 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;
}
XrdCl::Utils::GetHostAddresses
static Status GetHostAddresses(std::vector< XrdNetAddr > &addresses, const URL &url, AddressType type)
Resolve IP addresses.
Definition XrdClUtils.cc:140

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 585 of file XrdClStream.cc.

{
XrdSysMutexHelper scopedLock( pMutex );
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.

Here is the call graph for this function:

## Disconnect()

void XrdCl::Stream::Disconnect | ( | bool | force = false | ) |

Disconnect the stream.

Definition at line 363 of file XrdClStream.cc.

{
XrdSysMutexHelper scopedLock( pMutex );
SubStreamList::iterator it;
for( it = pSubStreams.begin(); it != pSubStreams.end(); ++it )
{
(*it)->socket->Close();
(*it)->status = Socket::Disconnected;
}
}
XrdCl::Socket::Disconnected
@ Disconnected
The socket is disconnected.
Definition XrdClSocket.hh:50

References XrdCl::Socket::Disconnected.

Referenced by ~Stream().

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 187 of file XrdClStream.cc.

{
XrdSysMutexHelper 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;
}
XrdCl::Socket::Connected
@ Connected
The socket is connected.
Definition XrdClSocket.hh:51
XrdCl::Socket::Connecting
@ Connecting
The connection process is in progress.
Definition XrdClSocket.hh:52
XrdCl::Utils::LogHostAddresses
static void LogHostAddresses(Log *log, uint64_t type, const std::string &hostId, std::vector< XrdNetAddr > &addresses)
Log all the addresses on the list.
Definition XrdClUtils.cc:234
XrdCl::stFatal
const uint16_t stFatal
Fatal error, it's still an error.
Definition XrdClStatus.hh:33

References XrdCl::Socket::Connected, XrdCl::Socket::Connecting, XrdCl::Socket::Disconnected, XrdCl::PathID::down, XrdCl::Log::Error(), XrdCl::Utils::GetHostAddresses(), XrdCl::DefaultEnv::GetLog(), XrdCl::Status::IsOK(), 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:

## ForceConnect()

void XrdCl::Stream::ForceConnect | ( |  | ) |

Force connection.

Definition at line 347 of file XrdClStream.cc.

{
XrdSysMutexHelper 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 );
}
}
XrdCl::Stream::EnableLink
XRootDStatus EnableLink(PathID &path)
Definition XrdClStream.cc:187
XrdCl::Stream::OnConnectError
void OnConnectError(uint16_t subStream, XRootDStatus status)
On connect error.
Definition XrdClStream.cc:728
XrdCl::Status::IsOK
bool IsOK() const
We're fine.
Definition XrdClStatus.hh:124

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,
|  | bool | hush = false )

Force error.

Definition at line 933 of file XrdClStream.cc.

{
XrdSysMutexHelper scopedLock( pMutex );
Log *log = DefaultEnv::GetLog();
for( size_t substream = 0; substream < pSubStreams.size(); ++substream )
{
if( pSubStreams[substream]->status != Socket::Connected ) continue;
pSubStreams[substream]->socket->Close();
pSubStreams[substream]->status = Socket::Disconnected;
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
//--------------------------------------------------------------------
if( pSubStreams[substream]->outMsgHelper.msg )
{
OutQueue::MsgHelper &h = pSubStreams[substream]->outMsgHelper;
pSubStreams[substream]->outQueue->PushFront( h.msg, h.handler, h.expires,
h.stateful );
pIncomingQueue->RemoveMessageHandler(h.handler);
pSubStreams[substream]->outMsgHelper.Reset();
}
//--------------------------------------------------------------------
// Reinsert the receiving handler and reset any partially read partial
//--------------------------------------------------------------------
if( pSubStreams[substream]->inMsgHelper.handler )
{
InMessageHelper &h = pSubStreams[substream]->inMsgHelper;
pIncomingQueue->ReAddMessageHandler( h.handler, h.expires );
XRootDMsgHandler *xrdHandler = dynamic_cast<XRootDMsgHandler*>( h.handler );
if( xrdHandler ) xrdHandler->PartialReceived();
h.Reset();
}
}
pConnectionCount = 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 );
}
XrdCl::ChannelEventHandler::StreamBroken
@ StreamBroken
The stream is broken.
Definition XrdClPostMasterInterfaces.hh:218
XrdCl::MsgHandler::Broken
@ Broken
The stream is broken.
Definition XrdClPostMasterInterfaces.hh:81

References XrdCl::MsgHandler::Broken, XrdCl::Socket::Connected, XrdCl::Log::Debug(), XrdCl::Socket::Disconnected, XrdCl::InMessageHelper::expires, XrdCl::OutQueue::MsgHelper::expires, XrdCl::DefaultEnv::GetLog(), XrdCl::OutQueue::GrabItems(), XrdCl::InMessageHelper::handler, XrdCl::OutQueue::MsgHelper::handler, XrdCl::OutQueue::MsgHelper::msg, XrdCl::XRootDMsgHandler::PartialReceived(), XrdCl::PostMasterMsg, XrdCl::OutQueue::Report(), XrdCl::InMessageHelper::Reset(), XrdCl::OutQueue::MsgHelper::Reset(), XrdCl::OutQueue::MsgHelper::stateful, XrdCl::ChannelEventHandler::StreamBroken, XrdCl::Status::ToString(), and XrdSysMutexHelper::UnLock().

Here is the call graph for this function:

## GetName()

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

inline

Return stream name.

Definition at line 170 of file XrdClStream.hh.

{
return pStreamName;
}

## GetURL()

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

inline

Get the URL.

Definition at line 157 of file XrdClStream.hh.

{
return pUrl;
}

## Initialize()

XRootDStatus XrdCl::Stream::Initialize | ( |  | ) |

Initializer.

Definition at line 171 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();
}
XrdCl::errUninitialized
const uint16_t errUninitialized
Definition XrdClStatus.hh:60
XrdCl::stError
const uint16_t stError
An error occurred that could potentially be retried.
Definition XrdClStatus.hh:32

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 1166 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;
}
XrdCl::MsgHandler::Raw
@ Raw
Definition XrdClPostMasterInterfaces.hh:63
XrdCl::MsgHandler::RemoveHandler
@ RemoveHandler
Definition XrdClPostMasterInterfaces.hh:61
XrdCl::MsgHandler::None
@ None
Definition XrdClPostMasterInterfaces.hh:58
XrdCl::MsgHandler::More
@ More
there are more (non-raw) data to be read
Definition XrdClPostMasterInterfaces.hh:72
XrdCl::MsgHandler::Corrupted
@ Corrupted
Definition XrdClPostMasterInterfaces.hh:69

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

Here is the call 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 1145 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::InMessageHelper::handler, and XrdCl::MsgHandler::Raw.

## OnConnect()

void XrdCl::Stream::OnConnect | ( | uint16_t | subStream | ) |

Call back when a message has been reconstructed.

Definition at line 623 of file XrdClStream.cc.

{
XrdSysMutexHelper scopedLock( pMutex );
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 )
{
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 );
pSubStreams[i]->socket->Close();
}
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 );
}
}
XrdCl::DefaultEnv::GetMonitor
static Monitor * GetMonitor()
Get the monitor object.
Definition XrdClDefaultEnv.cc:567
XrdCl::DefaultEnv::GetPostMaster
static PostMaster * GetPostMaster()
Get default post master.
Definition XrdClDefaultEnv.cc:453
XrdCl::Monitor::EvConnect
@ EvConnect
ConnectInfo: Login into a server.
Definition XrdClMonitor.hh:220
XrdCl::PostMaster::NotifyConnectHandler
void NotifyConnectHandler(const URL &url)
Notify the global on-connect handler.
Definition XrdClPostMaster.cc:421
XrdCl::TransportQuery::Auth
static const uint16_t Auth
Transport name, returns std::string *.
Definition XrdClPostMasterInterfaces.hh:282

References XrdCl::TransportQuery::Auth, XrdCl::Monitor::ConnectInfo::auth, XrdCl::Socket::Connected, XrdCl::Socket::Connecting, XrdCl::Log::Debug(), XrdCl::Monitor::ConnectInfo::eTOD, XrdCl::Monitor::EvConnect, XrdCl::Monitor::Event(), XrdCl::AnyObject::Get(), XrdCl::DefaultEnv::GetLog(), XrdCl::DefaultEnv::GetMonitor(), XrdCl::DefaultEnv::GetPostMaster(), XrdCl::Status::IsOK(), XrdCl::PostMaster::NotifyConnectHandler(), XrdCl::PostMasterMsg, XrdCl::Monitor::ConnectInfo::server, XrdCl::Monitor::ConnectInfo::sTOD, and XrdCl::Monitor::ConnectInfo::streams.

Here is the call graph for this function:

## OnConnectError()

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

On connect error.

Definition at line 728 of file XrdClStream.cc.

{
XrdSysMutexHelper scopedLock( pMutex );
Log *log = DefaultEnv::GetLog();
pSubStreams[subStream]->socket->Close();
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[subStream]->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;
}
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 );
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) );
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 );
}
XrdCl::PostMaster::NotifyConnErrHandler
void NotifyConnErrHandler(const URL &url, const XRootDStatus &status)
Notify the global error connection handler.
Definition XrdClPostMaster.cc:434

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

Referenced by ForceConnect().

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 828 of file XrdClStream.cc.

{
XrdSysMutexHelper scopedLock( pMutex );
Log *log = DefaultEnv::GetLog();
pSubStreams[subStream]->socket->Close();
pSubStreams[subStream]->status = Socket::Disconnected;
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
//--------------------------------------------------------------------------
if( pSubStreams[subStream]->outMsgHelper.msg )
{
OutQueue::MsgHelper &h = pSubStreams[subStream]->outMsgHelper;
pSubStreams[subStream]->outQueue->PushFront( h.msg, h.handler, h.expires,
h.stateful );
pIncomingQueue->RemoveMessageHandler(h.handler);
pSubStreams[subStream]->outMsgHelper.Reset();
}
//--------------------------------------------------------------------------
// Reinsert the receiving handler and reset any partially read partial
//--------------------------------------------------------------------------
if( pSubStreams[subStream]->inMsgHelper.handler )
{
InMessageHelper &h = pSubStreams[subStream]->inMsgHelper;
pIncomingQueue->ReAddMessageHandler( h.handler, h.expires );
XRootDMsgHandler *xrdHandler = dynamic_cast<XRootDMsgHandler*>( h.handler );
if( xrdHandler ) xrdHandler->PartialReceived();
h.Reset();
}
//--------------------------------------------------------------------------
// 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 );
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(), XrdCl::InMessageHelper::expires, XrdCl::OutQueue::MsgHelper::expires, XrdCl::DefaultEnv::GetLog(), XrdCl::OutQueue::GrabStateful(), XrdCl::InMessageHelper::handler, XrdCl::OutQueue::MsgHelper::handler, XrdCl::Status::IsOK(), XrdCl::OutQueue::MsgHelper::msg, XrdCl::XRootDMsgHandler::PartialReceived(), XrdCl::PostMasterMsg, XrdCl::OutQueue::Report(), XrdCl::InMessageHelper::Reset(), XrdCl::OutQueue::MsgHelper::Reset(), XrdCl::OutQueue::MsgHelper::stateful, XrdCl::ChannelEventHandler::StreamBroken, XrdCl::Status::ToString(), and XrdSysMutexHelper::UnLock().

Referenced by 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 471 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 );
}
ServerResponseHeader::status
kXR_unt16 status
Definition XProtocol.hh:915
ServerResponseHeader::streamid
kXR_char streamid[2]
Definition XProtocol.hh:914
ServerResponse::hdr
ServerResponseHeader hdr
Definition XProtocol.hh:1288
XrdCl::MsgHandler::NoProcess
@ NoProcess
Definition XrdClPostMasterInterfaces.hh:66
XrdCl::MsgHandler::Ignore
@ Ignore
Ignore the message.
Definition XrdClPostMasterInterfaces.hh:60
XrdCl::TransportHandler::DigestMsg
@ DigestMsg
Definition XrdClPostMasterInterfaces.hh:319
XrdCl::TransportHandler::RequestClose
@ RequestClose
Send a close request.
Definition XrdClPostMasterInterfaces.hh:329

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

Here is the call graph for this function:

## OnMessageSent()

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

Definition at line 601 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::OutQueue::MsgHelper::handler, XrdCl::MsgHandler::OnStatusReady(), and XrdCl::OutQueue::MsgHelper::Reset().

Here is the call graph for this function:

## OnReadTimeout()

bool XrdCl::Stream::OnReadTimeout | ( | uint16_t | subStream | ) |

On read timeout.

Definition at line 1050 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);
XrdSysMutexHelper scopedLock( pMutex );
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() );
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.
//
// Additionally &(*pUrl) is used by ForceDisconnect to check if we are
// in a Channel that was previously collapsed in a redirect.
//----------------------------------------------------------------------
DefaultEnv::GetPostMaster()->ForceDisconnect( *pUrl );
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;
}
XrdCl::PostMaster::ForceDisconnect
Status ForceDisconnect(const URL &url)
Shut down a channel.
Definition XrdClPostMaster.cc:324
XrdCl::Stream::OnError
void OnError(uint16_t subStream, XRootDStatus status)
On error.
Definition XrdClStream.cc:828

References XrdCl::Log::Debug(), XrdCl::PostMaster::ForceDisconnect(), XrdCl::DefaultEnv::GetLog(), XrdCl::DefaultEnv::GetPostMaster(), XrdCl::Status::IsOK(), OnError(), XrdCl::PostMasterMsg, and XrdSysMutexHelper::UnLock().

Here is the call graph for this function:

## OnReadyToWrite()

std::pair< Message *, MsgHandler * > XrdCl::Stream::OnReadyToWrite | ( | uint16_t | subStream | ) |

Definition at line 545 of file XrdClStream.cc.

{
XrdSysMutexHelper scopedLock( pMutex );
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::Log::Debug(), XrdCl::Log::Dump(), XrdCl::OutQueue::MsgHelper::expires, XrdCl::DefaultEnv::GetLog(), XrdCl::Message::GetObfuscatedDescription(), XrdCl::OutQueue::MsgHelper::handler, XrdCl::OutQueue::MsgHelper::msg, XrdCl::MsgHandler::OnReadyToSend(), XrdCl::PostMasterMsg, XrdCl::OutQueue::MsgHelper::stateful, XrdSysMutexHelper::UnLock(), and XrdCl::Log::Warning().

Here is the call graph for this function:

## OnWriteTimeout()

bool XrdCl::Stream::OnWriteTimeout | ( | uint16_t | subStream | ) |

On write timeout.

Definition at line 1120 of file XrdClStream.cc.

{
return true;
}

## Query()

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

Query the stream.

Definition at line 1242 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 );
}
}
XrdCl::errQueryNotSupported
const uint16_t errQueryNotSupported
Definition XrdClStatus.hh:89
XrdCl::StreamQuery::IpAddr
static const uint16_t IpAddr
Definition XrdClPostMasterInterfaces.hh:301
XrdCl::StreamQuery::HostName
static const uint16_t HostName
Definition XrdClPostMasterInterfaces.hh:303
XrdCl::StreamQuery::IpStack
static const uint16_t IpStack
Definition XrdClPostMasterInterfaces.hh:302

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 1128 of file XrdClStream.cc.

{
pChannelEvHandlers.AddHandler( handler );
}

## RemoveEventHandler()

void XrdCl::Stream::RemoveEventHandler | ( | ChannelEventHandler * | handler | ) |

Remove a channel event handler.

Definition at line 1136 of file XrdClStream.cc.

{
pChannelEvHandlers.RemoveHandler( handler );
}

## Send()

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

Queue the message for sending.

Definition at line 297 of file XrdClStream.cc.

{
XrdSysMutexHelper 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 );
pSubStreams[path.up]->outQueue->PushBack( msg, handler,
expires, stateful );
}
else
st.status = stFatal;
return st;
}
XrdCl::errInvalidSession
const uint16_t errInvalidSession
Definition XrdClStatus.hh:79

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::PostMasterMsg, XrdCl::Status::status, XrdCl::stError, XrdCl::stFatal, XrdCl::PathID::up, and XrdCl::Log::Warning().

Here is the call graph for this function:

## SetChannelData()

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

inline

Set the channel data.

Definition at line 115 of file XrdClStream.hh.

{
pChannelData = channelData;
}

## SetIncomingQueue()

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

inline

Set the incoming queue.

Definition at line 107 of file XrdClStream.hh.

{
pIncomingQueue = incomingQueue;
}

## SetJobManager()

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

inline

Set job manager.

Definition at line 131 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 263 of file XrdClStream.hh.

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

## SetPoller()

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

inline

Set the poller.

Definition at line 99 of file XrdClStream.hh.

{
pPoller = poller;
}

## SetTaskManager()

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

inline

Set task manager.

Definition at line 123 of file XrdClStream.hh.

{
pTaskManager = taskManager;
}

## SetTransport()

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

inline

Set the transport.

Definition at line 91 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 377 of file XrdClStream.cc.

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

References XrdCl::errOperationExpired, XrdCl::OutQueue::GrabExpired(), XrdCl::OutQueue::Report(), and XrdCl::stError.

Here is the call graph for this function:

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

- XrdClStream.hh
- XrdClStream.cc