Loading...
Searching...
No Matches
XrdClAsyncMsgWriter.hh
Go to the documentation of this file.
//------------------------------------------------------------------------------
// Copyright (c) 2011-2012 by European Organization for Nuclear Research (CERN)
// Author: Michal Simon <michal.simon@cern.ch>
//------------------------------------------------------------------------------
// XRootD is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// XRootD is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with XRootD. If not, see <http://www.gnu.org/licenses/>.
//------------------------------------------------------------------------------
#ifndef SRC_XRDCL_XRDCLASYNCMSGWRITER_HH_
#define SRC_XRDCL_XRDCLASYNCMSGWRITER_HH_
#include "XrdCl/XrdClMessage.hh"
#include "XrdCl/XrdClPostMasterInterfaces.hh"
#include "XrdCl/XrdClXRootDResponses.hh"
#include "XrdCl/XrdClSocket.hh"
#include "XrdCl/XrdClConstants.hh"
#include "XrdCl/XrdClStream.hh"
#include "XrdSys/XrdSysE2T.hh"
#include <memory>
namespace XrdCl
{
//----------------------------------------------------------------------------
//----------------------------------------------------------------------------
class AsyncMsgWriter
{
public:
//------------------------------------------------------------------------
//------------------------------------------------------------------------
AsyncMsgWriter( TransportHandler &xrdTransport,
Socket &socket,
const std::string &strmname,
Stream &strm,
uint16_t substrmnb,
AnyObject &chdata ) : writestage( WriteStart ),
xrdTransport( xrdTransport ),
socket( socket ),
strmname( strmname ),
strm( strm ),
substrmnb( substrmnb ),
chdata( chdata ),
outmsg( nullptr ),
outmsgsize( 0 ),
outhandler( nullptr )
{
}
//------------------------------------------------------------------------
//------------------------------------------------------------------------
inline void Reset()
{
writestage = WriteStart;
outmsg = nullptr;
outmsgsize = 0;;
outhandler = nullptr;
outsign.reset();
}
//------------------------------------------------------------------------
//------------------------------------------------------------------------
XRootDStatus Write()
{
Log *log = DefaultEnv::GetLog();
while( true )
{
switch( writestage )
{
//------------------------------------------------------------------
// Pick up a message if we're not in process of writing something
//------------------------------------------------------------------
case WriteStart:
{
std::pair<Message *, MsgHandler *> toBeSent;
toBeSent = strm.OnReadyToWrite( substrmnb );
outmsg = toBeSent.first;
outhandler = toBeSent.second;
if( !outmsg ) return XRootDStatus( stOK, suAlreadyDone );
outmsg->SetCursor( 0 );
outmsgsize = outmsg->GetSize();
//----------------------------------------------------------------
// Secure the message if necessary
//----------------------------------------------------------------
Message *signature = nullptr;
XRootDStatus st = xrdTransport.GetSignature( outmsg, signature, chdata );
if( !st.IsOK() ) return st;
outsign.reset( signature );
if( outsign )
outmsgsize += outsign->GetSize();
//----------------------------------------------------------------
// The next step is to write the signature
//----------------------------------------------------------------
writestage = WriteSign;
continue;
}
//------------------------------------------------------------------
// First write the signature (if there is one)
//------------------------------------------------------------------
case WriteSign:
{
//----------------------------------------------------------------
// If there is a signature for the request send it over the socket
//----------------------------------------------------------------
if( outsign )
{
XRootDStatus st = socket.Send( *outsign, strmname );
if( !st.IsOK() || st.code == suRetry ) return st;
}
//----------------------------------------------------------------
// The next step is to write the signature
//----------------------------------------------------------------
writestage = WriteRequest;
continue;
}
//------------------------------------------------------------------
// Then write the request itself
//------------------------------------------------------------------
case WriteRequest:
{
XRootDStatus st = socket.Send( *outmsg, strmname );
if( !st.IsOK() || st.code == suRetry ) return st;
//----------------------------------------------------------------
// The next step is to write the signature
//----------------------------------------------------------------
writestage = WriteRawData;
continue;
}
//------------------------------------------------------------------
// And then write the raw data (if any)
//------------------------------------------------------------------
case WriteRawData:
{
if( outhandler->IsRaw() )
{
uint32_t wrtcnt = 0;
XRootDStatus st = outhandler->WriteMessageBody( &socket, wrtcnt );
if( !st.IsOK() || st.code == suRetry ) return st;
outmsgsize += wrtcnt;
log->Dump( AsyncSockMsg, "[%s] Wrote %d bytes of raw data of message"
"(%p) body.", strmname.c_str(), wrtcnt, (void*)outmsg );
}
//----------------------------------------------------------------
// The next step is to finalize the write operation
//----------------------------------------------------------------
writestage = WriteDone;
continue;
}
//------------------------------------------------------------------
// Finally, finalize the write operation
//------------------------------------------------------------------
case WriteDone:
{
XRootDStatus st = socket.Flash();
if( !st.IsOK() )
{
log->Error( AsyncSockMsg, "[%s] Unable to flash the socket: %s",
strmname.c_str(), XrdSysE2T( st.errNo ) );
return st;
}
log->Dump( AsyncSockMsg, "[%s] Successfully sent message: %s (%p).",
strmname.c_str(), outmsg->GetObfuscatedDescription().c_str(), (void*)outmsg );
strm.OnMessageSent( substrmnb, outmsg, outmsgsize );
return XRootDStatus();
}
}
// just in case ...
break;
}
//----------------------------------------------------------------------
// We are done
//----------------------------------------------------------------------
return XRootDStatus();
}
private:
//------------------------------------------------------------------------
//------------------------------------------------------------------------
enum Stage
{
WriteStart, //< the next step is to initialize the read
WriteSign, //< the next step is to write the signature
WriteRequest, //< the next step is to write the request
WriteRawData, //< the next step is to write the raw data
WriteDone //< the next step is to finalize the write
};
//------------------------------------------------------------------------
// Current read stage
//------------------------------------------------------------------------
Stage writestage;
//------------------------------------------------------------------------
// The context of the read operation
//------------------------------------------------------------------------
TransportHandler &xrdTransport;
Socket &socket;
const std::string &strmname;
Stream &strm;
uint16_t substrmnb;
AnyObject &chdata;
//------------------------------------------------------------------------
// The internal state of the the reader
//------------------------------------------------------------------------
Message *outmsg; //< we don't own the message
uint32_t outmsgsize;
MsgHandler *outhandler;
std::unique_ptr<Message> outsign;
};
}
#endif /* SRC_XRDCL_XRDCLASYNCMSGWRITER_HH_ */
XrdClConstants.hh
XrdClMessage.hh
XrdClPostMasterInterfaces.hh
XrdClSocket.hh
XrdClStream.hh
XrdClXRootDResponses.hh
XrdSysE2T
const char * XrdSysE2T(int errcode)
Definition XrdSysE2T.cc:104
XrdSysE2T.hh
XrdCl::AnyObject
Definition XrdClAnyObject.hh:33
XrdCl::AsyncMsgWriter::Write
XRootDStatus Write()
Write the request into the socket.
Definition XrdClAsyncMsgWriter.hh:82
XrdCl::AsyncMsgWriter::AsyncMsgWriter
AsyncMsgWriter(TransportHandler &xrdTransport, Socket &socket, const std::string &strmname, Stream &strm, uint16_t substrmnb, AnyObject &chdata)
Definition XrdClAsyncMsgWriter.hh:49
XrdCl::AsyncMsgWriter::Reset
void Reset()
Reset the state of the object (makes it ready to read out next msg).
Definition XrdClAsyncMsgWriter.hh:70
XrdCl::DefaultEnv::GetLog
static Log * GetLog()
Get default log.
Definition XrdClDefaultEnv.cc:493
XrdCl::Log
Handle diagnostics.
Definition XrdClLog.hh:101
XrdCl::Log::Error
void Error(uint64_t topic, const char *format,...)
Report an error.
Definition XrdClLog.cc:231
XrdCl::Log::Dump
void Dump(uint64_t topic, const char *format,...)
Print a dump message.
Definition XrdClLog.cc:299
XrdCl::Message
The message representation used throughout the system.
Definition XrdClMessage.hh:32
XrdCl::Socket
A network socket.
Definition XrdClSocket.hh:43
XrdCl::Stream
Stream.
Definition XrdClStream.hh:52
XrdCl::TransportHandler
Perform the handshake and the authentication for each physical stream.
Definition XrdClPostMasterInterfaces.hh:310
XrdCl::XRootDStatus
Request status.
Definition XrdClXRootDResponses.hh:219
XrdCl
Definition XrdClAction.hh:34
XrdCl::suRetry
const uint16_t suRetry
Definition XrdClStatus.hh:40
XrdCl::stOK
const uint16_t stOK
Everything went OK.
Definition XrdClStatus.hh:31
XrdCl::AsyncSockMsg
const uint64_t AsyncSockMsg
Definition XrdClConstants.hh:41
XrdCl::suAlreadyDone
const uint16_t suAlreadyDone
Definition XrdClStatus.hh:42
XrdCl::Status::code
uint16_t code
Error type, or additional hints on what to do.
Definition XrdClStatus.hh:147
XrdCl::Status::IsOK
bool IsOK() const
We're fine.
Definition XrdClStatus.hh:124
XrdCl::Status::errNo
uint32_t errNo
Errno, if any.
Definition XrdClStatus.hh:148