Commit ba9202b4 authored by Steven Murray's avatar Steven Murray
Browse files

Moved AcsProxyZmq::createXFrame() methods to cpp

parent 2a523442
......@@ -28,59 +28,60 @@
#include "mediachanger/Constants.hpp"
#include "mediachanger/messages.hpp"
//------------------------------------------------------------------------------
// constructor
//------------------------------------------------------------------------------
cta::mediachanger::AcsProxyZmq::AcsProxyZmq(const unsigned short serverPort,
void *const zmqContext) throw():
m_serverPort(serverPort),
m_serverSocket(zmqContext, ZMQ_REQ) {
connectZmqSocketToLocalhost(m_serverSocket, serverPort);
}
namespace cta {
namespace mediachanger {
//------------------------------------------------------------------------------
// mountTapeReadOnly
//------------------------------------------------------------------------------
void cta::mediachanger::AcsProxyZmq::mountTapeReadOnly(const std::string &vid,
const cta::mediachanger::AcsLibrarySlot &librarySlot) {
std::lock_guard<std::mutex> lock(m_mutex);
namespace {
/**
* Creates a frame containing an AcsMountTapeReadOnly message.
*
* @param vid The tape to be mounted.
* @param librarySlot The slot in the library that contains the tape drive.
* @return The frame.
*/
Frame createAcsMountTapeReadOnlyFrame(const std::string &vid, const AcsLibrarySlot &librarySlot) {
try {
const Frame rqst = createAcsMountTapeReadOnlyFrame(vid, librarySlot);
sendFrame(m_serverSocket, rqst);
Frame frame;
frame.header = protoTapePreFillHeader();
frame.header.set_msgtype(MSG_TYPE_ACSMOUNTTAPEREADONLY);
frame.header.set_bodysignature("PIPO");
AcsMountTapeReadOnly body;
body.set_vid(vid);
body.set_acs(librarySlot.getAcs());
body.set_lsm(librarySlot.getLsm());
body.set_panel(librarySlot.getPanel());
body.set_drive(librarySlot.getDrive());
frame.serializeProtocolBufferIntoBody(body);
return frame;
ReturnValue reply;
recvTapeReplyOrEx(m_serverSocket, reply);
if(0 != reply.value()) {
// Should never get here
cta::exception::Exception ex;
ex.getMessage() << "Received an unexpected return value"
": expected=0 actual=" << reply.value();
throw ex;
}
} catch(cta::exception::Exception &ne) {
cta::exception::Exception ex;
ex.getMessage() <<
"Failed to request CASTOR ACS daemon to mount tape for read only access: "
<< librarySlot.str() << ": " << ne.getMessage().str();
ex.getMessage() << "Failed to create AcsMountTapeReadOnly frame: " <<
ne.getMessage().str();
throw ex;
}
}
//------------------------------------------------------------------------------
// createAcsMountTapeReadnOnlyFrame
//------------------------------------------------------------------------------
cta::mediachanger::Frame cta::mediachanger::AcsProxyZmq::
createAcsMountTapeReadOnlyFrame(const std::string &vid,
const cta::mediachanger::AcsLibrarySlot &librarySlot) {
/**
* Creates a frame containing a AcsMountTapeReadWrite message.
*
* @param vid The tape to be mounted.
* @param librarySlot The slot in the library that contains the tape drive.
* @return The frame.
*/
Frame createAcsMountTapeReadWriteFrame(const std::string &vid, const AcsLibrarySlot &librarySlot) {
try {
Frame frame;
frame.header = protoTapePreFillHeader();
frame.header.set_msgtype(MSG_TYPE_ACSMOUNTTAPEREADONLY);
frame.header.set_msgtype(MSG_TYPE_ACSMOUNTTAPEREADWRITE);
frame.header.set_bodysignature("PIPO");
AcsMountTapeReadOnly body;
AcsMountTapeReadWrite body;
body.set_vid(vid);
body.set_acs(librarySlot.getAcs());
body.set_lsm(librarySlot.getLsm());
......@@ -92,55 +93,61 @@ cta::mediachanger::Frame cta::mediachanger::AcsProxyZmq::
} catch(cta::exception::Exception &ne) {
cta::exception::Exception ex;
ex.getMessage() << "Failed to create AcsMountTapeReadOnly frame: " <<
ex.getMessage() << "Failed to create AcsMountTapeReadWrite frame: " <<
ne.getMessage().str();
throw ex;
}
}
//------------------------------------------------------------------------------
// mountTapeReadWrite
//------------------------------------------------------------------------------
void cta::mediachanger::AcsProxyZmq::mountTapeReadWrite(const std::string &vid,
const cta::mediachanger::AcsLibrarySlot &librarySlot) {
std::lock_guard<std::mutex> lock(m_mutex);
/**
* Creates a frame containing an AcsDismountTape message.
*
* @param vid The tape to be dismounted.
* @param librarySlot The slot in the library that contains the tape drive.
* @return The frame.
*/
Frame createAcsDismountTapeFrame(const std::string &vid, const AcsLibrarySlot &librarySlot) {
try {
const Frame rqst = createAcsMountTapeReadWriteFrame(vid, librarySlot);
sendFrame(m_serverSocket, rqst);
Frame frame;
frame.header = protoTapePreFillHeader();
frame.header.set_msgtype(MSG_TYPE_ACSDISMOUNTTAPE);
frame.header.set_bodysignature("PIPO");
AcsDismountTape body;
body.set_vid(vid);
body.set_acs(librarySlot.getAcs());
body.set_lsm(librarySlot.getLsm());
body.set_panel(librarySlot.getPanel());
body.set_drive(librarySlot.getDrive());
frame.serializeProtocolBufferIntoBody(body);
return frame;
ReturnValue reply;
recvTapeReplyOrEx(m_serverSocket, reply);
if(0 != reply.value()) {
// Should never get here
cta::exception::Exception ex;
ex.getMessage() << "Received an unexpected return value"
": expected=0 actual=" << reply.value();
throw ex;
}
} catch(cta::exception::Exception &ne) {
cta::exception::Exception ex;
ex.getMessage() <<
"Failed to request CASTOR ACS daemon to mount tape for read/write "
"access: " << librarySlot.str() << ": " << ne.getMessage().str();
ex.getMessage() << "Failed to create AcsDismountTape frame: " <<
ne.getMessage().str();
throw ex;
}
}
//------------------------------------------------------------------------------
// createAcsMountTapeReadWriteFrame
//------------------------------------------------------------------------------
cta::mediachanger::Frame cta::mediachanger::AcsProxyZmq::
createAcsMountTapeReadWriteFrame(const std::string &vid,
const cta::mediachanger::AcsLibrarySlot &librarySlot) {
/**
* Creates a frame containing an AcsForceDismountTape message.
*
* @param vid The tape to be dismounted.
* @param librarySlot The slot in the library that contains the tape drive.
* @return The frame.
*/
Frame createAcsForceDismountTapeFrame(const std::string &vid, const AcsLibrarySlot &librarySlot) {
try {
Frame frame;
frame.header = protoTapePreFillHeader();
frame.header.set_msgtype(MSG_TYPE_ACSMOUNTTAPEREADWRITE);
frame.header.set_msgtype(MSG_TYPE_ACSFORCEDISMOUNTTAPE);
frame.header.set_bodysignature("PIPO");
AcsMountTapeReadWrite body;
AcsForceDismountTape body;
body.set_vid(vid);
body.set_acs(librarySlot.getAcs());
body.set_lsm(librarySlot.getLsm());
......@@ -152,21 +159,33 @@ cta::mediachanger::Frame cta::mediachanger::AcsProxyZmq::
} catch(cta::exception::Exception &ne) {
cta::exception::Exception ex;
ex.getMessage() << "Failed to create AcsMountTapeReadWrite frame: " <<
ex.getMessage() << "Failed to create AcsForceDismountTape frame: " <<
ne.getMessage().str();
throw ex;
}
}
} // anonyous namespace
//------------------------------------------------------------------------------
// dismountTape
// constructor
//------------------------------------------------------------------------------
void cta::mediachanger::AcsProxyZmq::dismountTape(const std::string &vid,
const cta::mediachanger::AcsLibrarySlot &librarySlot) {
AcsProxyZmq::AcsProxyZmq(const unsigned short serverPort,
void *const zmqContext) throw():
m_serverPort(serverPort),
m_serverSocket(zmqContext, ZMQ_REQ) {
connectZmqSocketToLocalhost(m_serverSocket, serverPort);
}
//------------------------------------------------------------------------------
// mountTapeReadOnly
//------------------------------------------------------------------------------
void AcsProxyZmq::mountTapeReadOnly(const std::string &vid,
const AcsLibrarySlot &librarySlot) {
std::lock_guard<std::mutex> lock(m_mutex);
try {
const Frame rqst = createAcsDismountTapeFrame(vid, librarySlot);
const Frame rqst = createAcsMountTapeReadOnlyFrame(vid, librarySlot);
sendFrame(m_serverSocket, rqst);
ReturnValue reply;
......@@ -181,52 +200,50 @@ void cta::mediachanger::AcsProxyZmq::dismountTape(const std::string &vid,
} catch(cta::exception::Exception &ne) {
cta::exception::Exception ex;
ex.getMessage() <<
"Failed to request CASTOR ACS daemon to dismount tape: " <<
librarySlot.str() << ": " << ne.getMessage().str();
"Failed to request CASTOR ACS daemon to mount tape for read only access: "
<< librarySlot.str() << ": " << ne.getMessage().str();
throw ex;
}
}
//------------------------------------------------------------------------------
// createAcsDismountTapeFrame
// mountTapeReadWrite
//------------------------------------------------------------------------------
cta::mediachanger::Frame cta::mediachanger::AcsProxyZmq::
createAcsDismountTapeFrame(const std::string &vid,
const cta::mediachanger::AcsLibrarySlot &librarySlot) {
void AcsProxyZmq::mountTapeReadWrite(const std::string &vid,
const AcsLibrarySlot &librarySlot) {
std::lock_guard<std::mutex> lock(m_mutex);
try {
Frame frame;
frame.header = protoTapePreFillHeader();
frame.header.set_msgtype(MSG_TYPE_ACSDISMOUNTTAPE);
frame.header.set_bodysignature("PIPO");
AcsDismountTape body;
body.set_vid(vid);
body.set_acs(librarySlot.getAcs());
body.set_lsm(librarySlot.getLsm());
body.set_panel(librarySlot.getPanel());
body.set_drive(librarySlot.getDrive());
frame.serializeProtocolBufferIntoBody(body);
return frame;
const Frame rqst = createAcsMountTapeReadWriteFrame(vid, librarySlot);
sendFrame(m_serverSocket, rqst);
ReturnValue reply;
recvTapeReplyOrEx(m_serverSocket, reply);
if(0 != reply.value()) {
// Should never get here
cta::exception::Exception ex;
ex.getMessage() << "Received an unexpected return value"
": expected=0 actual=" << reply.value();
throw ex;
}
} catch(cta::exception::Exception &ne) {
cta::exception::Exception ex;
ex.getMessage() << "Failed to create AcsDismountTape frame: " <<
ne.getMessage().str();
ex.getMessage() <<
"Failed to request CASTOR ACS daemon to mount tape for read/write "
"access: " << librarySlot.str() << ": " << ne.getMessage().str();
throw ex;
}
}
//------------------------------------------------------------------------------
// forceDismountTape
// dismountTape
//------------------------------------------------------------------------------
void cta::mediachanger::AcsProxyZmq::forceDismountTape(const std::string &vid,
const cta::mediachanger::AcsLibrarySlot &librarySlot) {
void AcsProxyZmq::dismountTape(const std::string &vid,
const AcsLibrarySlot &librarySlot) {
std::lock_guard<std::mutex> lock(m_mutex);
try {
const Frame rqst = createAcsForceDismountTapeFrame(vid, librarySlot);
const Frame rqst = createAcsDismountTapeFrame(vid, librarySlot);
sendFrame(m_serverSocket, rqst);
ReturnValue reply;
......@@ -241,39 +258,40 @@ void cta::mediachanger::AcsProxyZmq::forceDismountTape(const std::string &vid,
} catch(cta::exception::Exception &ne) {
cta::exception::Exception ex;
ex.getMessage() <<
"Failed to request CASTOR ACS daemon to force dismount tape: " <<
"Failed to request CASTOR ACS daemon to dismount tape: " <<
librarySlot.str() << ": " << ne.getMessage().str();
throw ex;
}
}
//------------------------------------------------------------------------------
// createAcsForceDismountTapeFrame
// forceDismountTape
//------------------------------------------------------------------------------
cta::mediachanger::Frame cta::mediachanger::AcsProxyZmq::
createAcsForceDismountTapeFrame(const std::string &vid,
const cta::mediachanger::AcsLibrarySlot &librarySlot) {
try {
Frame frame;
void AcsProxyZmq::forceDismountTape(const std::string &vid,
const AcsLibrarySlot &librarySlot) {
std::lock_guard<std::mutex> lock(m_mutex);
frame.header = protoTapePreFillHeader();
frame.header.set_msgtype(MSG_TYPE_ACSFORCEDISMOUNTTAPE);
frame.header.set_bodysignature("PIPO");
AcsForceDismountTape body;
body.set_vid(vid);
body.set_acs(librarySlot.getAcs());
body.set_lsm(librarySlot.getLsm());
body.set_panel(librarySlot.getPanel());
body.set_drive(librarySlot.getDrive());
frame.serializeProtocolBufferIntoBody(body);
return frame;
try {
const Frame rqst = createAcsForceDismountTapeFrame(vid, librarySlot);
sendFrame(m_serverSocket, rqst);
ReturnValue reply;
recvTapeReplyOrEx(m_serverSocket, reply);
if(0 != reply.value()) {
// Should never get here
cta::exception::Exception ex;
ex.getMessage() << "Received an unexpected return value"
": expected=0 actual=" << reply.value();
throw ex;
}
} catch(cta::exception::Exception &ne) {
cta::exception::Exception ex;
ex.getMessage() << "Failed to create AcsForceDismountTape frame: " <<
ne.getMessage().str();
ex.getMessage() <<
"Failed to request CASTOR ACS daemon to force dismount tape: " <<
librarySlot.str() << ": " << ne.getMessage().str();
throw ex;
}
}
} // namespace mediachanger
} // namespace cta
......@@ -22,7 +22,6 @@
#pragma once
#include "mediachanger/AcsProxy.hpp"
#include "mediachanger/Frame.hpp"
#include "mediachanger/ZmqSocketMT.hpp"
#include <mutex>
......@@ -105,46 +104,6 @@ private:
*/
ZmqSocketMT m_serverSocket;
/**
* Creates a frame containing a AcsMountTapeForRecall message.
*
* @param vid The tape to be mounted.
* @param librarySlot The slot in the library that contains the tape drive.
* @return The frame.
*/
Frame createAcsMountTapeReadOnlyFrame(const std::string &vid,
const cta::mediachanger::AcsLibrarySlot &librarySlot);
/**
* Creates a frame containing a AcsMountTapeForMigration message.
*
* @param vid The tape to be mounted.
* @param librarySlot The slot in the library that contains the tape drive.
* @return The frame.
*/
Frame createAcsMountTapeReadWriteFrame(const std::string &vid,
const cta::mediachanger::AcsLibrarySlot &librarySlot);
/**
* Creates a frame containing a AcsDismountTape message.
*
* @param vid The tape to be dismounted.
* @param librarySlot The slot in the library that contains the tape drive.
* @return The frame.
*/
Frame createAcsDismountTapeFrame(const std::string &vid,
const cta::mediachanger::AcsLibrarySlot &librarySlot);
/**
* Creates a frame containing a AcsDismountTape message.
*
* @param vid The tape to be dismounted.
* @param librarySlot The slot in the library that contains the tape drive.
* @return The frame.
*/
Frame createAcsForceDismountTapeFrame(const std::string &vid,
const cta::mediachanger::AcsLibrarySlot &librarySlot);
}; // class AcsProxyZmq
} // namespace mediachanger
......
......@@ -31,6 +31,7 @@
#include "mediachanger/SmartZmqContext.hpp"
#include <exception>
#include <google/protobuf/stubs/common.h>
#include <iostream>
#include <zmq.h>
......
......@@ -31,6 +31,7 @@
#include "common/utils/utils.hpp"
#include <exception>
#include <google/protobuf/stubs/common.h>
#include <iostream>
#include <zmq.h>
......
......@@ -130,7 +130,6 @@ add_library(ctatapeserverdaemonunittests SHARED
)
target_link_libraries(ctatapeserverdaemonunittests
ctamediachangerutils
ctamessagesutils
ctatapeserverdaemonutils)
#ctaschedulerutils
......
......@@ -16,26 +16,27 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "DriveHandler.hpp"
#include "DriveHandlerProxy.hpp"
#include "catalogue/CatalogueFactory.hpp"
#include "common/log/LogContext.hpp"
#include "common/exception/Errnum.hpp"
#include "tapeserver/daemon/WatchdogMessage.pb.h"
#include "common/processCap/ProcessCap.hpp"
#include "tapeserver/castor/messages/SmartZmqContext.hpp"
#include "tapeserver/castor/messages/AcsProxyZmq.hpp"
#include "tapeserver/castor/acs/Constants.hpp"
#include "tapeserver/castor/mediachanger/MmcProxyLog.hpp"
#include "tapeserver/castor/tape/tapeserver/daemon/CleanerSession.hpp"
#include "tapeserver/castor/tape/tapeserver/daemon/DataTransferSession.hpp"
#include "tapeserver/castor/legacymsg/RmcProxyTcpIp.hpp"
#include "DriveHandler.hpp"
#include "DriveHandlerProxy.hpp"
#include "mediachanger/AcsProxyZmq.hpp"
#include "mediachanger/MmcProxyLog.hpp"
#include "mediachanger/RmcProxyTcpIp.hpp"
#include "objectstore/Backend.hpp"
#include "objectstore/BackendFactory.hpp"
#include "objectstore/BackendVFS.hpp"
#include "objectstore/BackendPopulator.hpp"
#include "scheduler/OStoreDB/OStoreDBWithAgent.hpp"
#include "rdbms/Login.hpp"
#include "catalogue/CatalogueFactory.hpp"
#include "scheduler/OStoreDB/OStoreDBWithAgent.hpp"
#include "tapeserver/castor/messages/SmartZmqContext.hpp"
#include "tapeserver/castor/acs/Constants.hpp"
#include "tapeserver/castor/tape/tapeserver/daemon/CleanerSession.hpp"
#include "tapeserver/castor/tape/tapeserver/daemon/DataTransferSession.hpp"
#include "tapeserver/daemon/WatchdogMessage.pb.h"
#include <unistd.h>
#include <signal.h>
#include <sys/wait.h>
......@@ -669,23 +670,23 @@ int DriveHandler::runChild() {
castor::messages::SmartZmqContext zmqContext(
castor::messages::SmartZmqContext::instantiateZmqContext(sizeOfIOThreadPoolForZMQ,
m_processManager.logContext().logger()));
castor::messages::AcsProxyZmq acs(castor::acs::ACS_PORT, zmqContext.get());
mediachanger::AcsProxyZmq acs(castor::acs::ACS_PORT, zmqContext.get());
castor::mediachanger::MmcProxyLog mmc(m_processManager.logContext().logger());
cta::mediachanger::MmcProxyLog mmc(m_processManager.logContext().logger());
// The network timeout of rmc communications should be several minutes due
// to the time it takes to mount and unmount tapes
const int rmcNetTimeout = 600; // Timeout in seconds
castor::legacymsg::RmcProxyTcpIp rmc(RMC_PORT, rmcNetTimeout,
RMC_MAXRQSTATTEMPTS);
cta::mediachanger::RmcProxyTcpIp rmc(cta::mediachanger::RMC_PORT, rmcNetTimeout,
cta::mediachanger::RMC_MAXRQSTATTEMPTS);
castor::mediachanger::MediaChangerFacade mediaChangerFacade(acs, mmc, rmc);
cta::mediachanger::MediaChangerFacade mediaChangerFacade(acs, mmc, rmc);
castor::tape::System::realWrapper sWrapper;
castor::tape::tapeserver::daemon::DriveConfig driveConfig;
driveConfig.m_unitName = m_configLine.unitName;
driveConfig.m_librarySlot = castor::mediachanger::LibrarySlotParser::parse(m_configLine.librarySlot);
driveConfig.m_librarySlot = cta::mediachanger::LibrarySlotParser::parse(m_configLine.librarySlot);
driveConfig.m_devFilename = m_configLine.devFilename;
driveConfig.m_logicalLibrary = m_configLine.logicalLibrary;
......@@ -710,23 +711,23 @@ int DriveHandler::runChild() {
castor::messages::SmartZmqContext zmqContext(
castor::messages::SmartZmqContext::instantiateZmqContext(sizeOfIOThreadPoolForZMQ,
m_processManager.logContext().logger()));
castor::messages::AcsProxyZmq acs(castor::acs::ACS_PORT, zmqContext.get());
mediachanger::AcsProxyZmq acs(castor::acs::ACS_PORT, zmqContext.get());
castor::mediachanger::MmcProxyLog mmc(m_processManager.logContext().logger());
cta::mediachanger::MmcProxyLog mmc(m_processManager.logContext().logger());
// The network timeout of rmc communications should be several minutes due
// to the time it takes to mount and unmount tapes
const int rmcNetTimeout = 600; // Timeout in seconds
castor::legacymsg::RmcProxyTcpIp rmc(RMC_PORT, rmcNetTimeout,
RMC_MAXRQSTATTEMPTS);
cta::mediachanger::RmcProxyTcpIp rmc(cta::mediachanger::RMC_PORT, rmcNetTimeout,
cta::mediachanger::RMC_MAXRQSTATTEMPTS);
castor::mediachanger::MediaChangerFacade mediaChangerFacade(acs, mmc, rmc);
cta::mediachanger::MediaChangerFacade mediaChangerFacade(acs, mmc, rmc);
castor::tape::System::realWrapper sWrapper;
castor::tape::tapeserver::daemon::DriveConfig driveConfig;
driveConfig.m_unitName = m_configLine.unitName;
driveConfig.m_librarySlot = castor::mediachanger::LibrarySlotParser::parse(m_configLine.librarySlot);
driveConfig.m_librarySlot = cta::mediachanger::LibrarySlotParser::parse(m_configLine.librarySlot);
driveConfig.m_devFilename = m_configLine.devFilename;
driveConfig.m_logicalLibrary = m_configLine.logicalLibrary;
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment