text:
// Copyright Robert Eisele 2017
#include <iostream>
#include <fstream>
#include <Eigen/Dense>
#include "readCSV.h"

Eigen::MatrixXd readCSV(std::string file, int rows, int cols) {

  std::ifstream in(file);

  std::string line;

  int row = 0;
  int col = 0;

  Eigen::MatrixXd res = Eigen::MatrixXd(rows, cols);

  if (in.is_open()) {

    while (std::getline(in, line)) {

      char *ptr = (char *) line.c_str();
      int len = line.length();

      col = 0;

      char *start = ptr;
      for (int i = 0; i < len; i++) {

        if (ptr[i] == ',') {
          res(row, col++) = atof(start);
          start = ptr + i + 1;
        }
      }
      res(row, col) = atof(start);

      row++;
    }

    in.close();
  }
  return res;
}

max_stars_repo_name: Jonghyun-Yun/LSA

max_issues_repo_name: Jonghyun-Yun/LSA

max_forks_repo_name: Jonghyun-Yun/LSA


text:
/*
==============================================================================
KratosStructuralApplication
A library based on:
Kratos
A General Purpose Software for Multi-Physics Finite Element Analysis
Version 1.0 (Released on march 05, 2007).

Copyright 2007
Pooyan Dadvand, Riccardo Rossi, Janosch Stascheit, Felix Nagel
pooyan@cimne.upc.edu
rrossi@cimne.upc.edu
janosch.stascheit@rub.de
nagel@sd.rub.de
- CIMNE (International Center for Numerical Methods in Engineering),
Gran Capita' s/n, 08034 Barcelona, Spain
- Ruhr-University Bochum, Institute for Structural Mechanics, Germany

Permission is hereby granted, free  of charge, to any person obtaining
a  copy  of this  software  and  associated  documentation files  (the
"Software"), to  deal in  the Software without  restriction, including
without limitation  the rights to  use, copy, modify,  merge, publish,
distribute,  sublicense and/or  sell copies  of the  Software,  and to
permit persons to whom the Software  is furnished to do so, subject to
the following condition:

Distribution of this code for  any  commercial purpose  is permissible
ONLY BY DIRECT ARRANGEMENT WITH THE COPYRIGHT OWNERS.

The  above  copyright  notice  and  this permission  notice  shall  be
included in all copies or substantial portions of the Software.

THE  SOFTWARE IS  PROVIDED  "AS  IS", WITHOUT  WARRANTY  OF ANY  KIND,
EXPRESS OR  IMPLIED, INCLUDING  BUT NOT LIMITED  TO THE  WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
IN NO EVENT  SHALL THE AUTHORS OR COPYRIGHT HOLDERS  BE LIABLE FOR ANY
CLAIM, DAMAGES OR  OTHER LIABILITY, WHETHER IN AN  ACTION OF CONTRACT,
TORT  OR OTHERWISE, ARISING  FROM, OUT  OF OR  IN CONNECTION  WITH THE
SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

==============================================================================
*/

//
//   Project Name:        Kratos
//   Last Modified by:    $Author: anonymous $
//   Date:                $Date: 2008-10-23 14:27:01 $
//   Revision:            $Revision: 1.20 $
//
//

#if !defined(KRATOS_ADD_CUSTOM_UTILITIES_TO_PYTHON_H_INCLUDED )
#define  KRATOS_ADD_CUSTOM_UTILITIES_TO_PYTHON_H_INCLUDED

// System includes
#include <boost/python.hpp>
#include <boost/python/suite/indexing/vector_indexing_suite.hpp>

// External includes
#include "boost/smart_ptr.hpp"

// Project includes
#include "custom_python/add_custom_utilities_to_python.h"
#include "includes/define.h"
#include "custom_utilities/deactivation_utility.h"
#include "custom_utilities/variable_transfer_utility.h"

#ifdef _OPENMP
#include "custom_utilities/parallel_variable_transfer_utility.h"
#endif

#include "spaces/ublas_space.h"
#include "linear_solvers/linear_solver.h"
#include "custom_utilities/contact_utility.h"
#include "custom_utilities/volume_utility.h"
#include "custom_utilities/restart_utility.h"
#include "custom_utilities/node_snapping_utility.h"
#include "custom_elements/rigid_body_3D.h"
#include "custom_utilities/output_utility.h"
#include "custom_utilities/dof_utility.h"
#include "custom_utilities/smoothing_utility.h"

//#include "custom_utilities/detect_elements_utility.h"
#include "custom_utilities/intra_fracture_triangle_utility.h"
#include "custom_utilities/inter_fracture_triangle_utility.h"
#include "custom_utilities/inter_fracture_tetrahedra_utility.h"
//#include "custom_utilities/mark_element_for_refinement.h"
#include "custom_utilities/disconnect_utility.h"

namespace Kratos
{

namespace Python
{

using namespace boost::python;

void AddNewRigidBody3D( ModelPart& structural_model_part,
                        ModelPart& skin_model_part,
                        Variable<double>& rSelectionVariable,
                        double selection_value,
                        Node<3>::Pointer CenterNode,
                        Element::PropertiesType::Pointer pProperties,
                        double nodal_mass,
                        Matrix& Inertia
                      )
{
    Geometry<Node<3> >::Pointer skin_nodes_geometry( new Geometry<Node<3> > ); ;

    //selecting the nodes in the model part having rSelectionVariable==selection_value

    for ( ModelPart::NodesContainerType::iterator it = skin_model_part.NodesBegin(); it != skin_model_part.NodesEnd(); it++ )
    {
        if ( it->FastGetSolutionStepValue( rSelectionVariable ) == selection_value )
            skin_nodes_geometry->push_back( *( it.base() ) );
    }

    //creating a geometry containing the center node
    Geometry<Node<3> >::Pointer center_node_geometry( new Geometry<Node<3> > ) ;

    center_node_geometry->push_back( Node<3>::Pointer( CenterNode ) );

    unsigned int last_id = 1;

    if ( structural_model_part.Elements().size() != 0 )
        last_id = ( structural_model_part.ElementsEnd() - 1 )->Id() + 1;

    array_1d<double, 3> zero = ZeroVector( 3 );

    Element::Pointer new_el = RigidBody3D::Pointer( new  RigidBody3D( last_id,
                              center_node_geometry,
                              pProperties,
                              skin_nodes_geometry,
                              nodal_mass,
                              Inertia, zero, zero ) );

    structural_model_part.Elements().push_back(
        new_el
    );
}

void AddNewRigidBodyAndSpring3D( ModelPart& structural_model_part,
                                 ModelPart& skin_model_part,
                                 Variable<double>& rSelectionVariable,
                                 double selection_value,
                                 Node<3>::Pointer CenterNode,
                                 Element::PropertiesType::Pointer pProperties,
                                 double nodal_mass,
                                 Matrix& Inertia,
                                 array_1d<double, 3>& translational_stiffness,
                                 array_1d<double, 3>& rotational_stiffness
                               )
{
    Geometry<Node<3> >::Pointer skin_nodes_geometry( new Geometry<Node<3> > ); ;

    //selecting the nodes in the model part having rSelectionVariable==selection_value

    for ( ModelPart::NodesContainerType::iterator it = skin_model_part.NodesBegin(); it != skin_model_part.NodesEnd(); it++ )
    {
        if ( it->FastGetSolutionStepValue( rSelectionVariable ) == selection_value )
            skin_nodes_geometry->push_back( *( it.base() ) );
    }

    //creating a geometry containing the center node
    Geometry<Node<3> >::Pointer center_node_geometry( new Geometry<Node<3> > ) ;

    center_node_geometry->push_back( Node<3>::Pointer( CenterNode ) );

    unsigned int last_id = 1;

    if ( structural_model_part.Elements().size() != 0 )
        last_id = ( structural_model_part.ElementsEnd() - 1 )->Id() + 1;

    array_1d<double, 3> zero = ZeroVector( 3 );

    Element::Pointer new_el = RigidBody3D::Pointer( new  RigidBody3D( last_id,
                              center_node_geometry,
                              pProperties,
                              skin_nodes_geometry,
                              nodal_mass,
                              Inertia,
                              translational_stiffness,
                              rotational_stiffness ) );

    structural_model_part.Elements().push_back(
        new_el
    );
}

void DoubleTransferVariablesToNodes(VariableTransferUtility& dummy,
        ModelPart& model_part, Variable<double>& rThisVariable)
{
    dummy.TransferVariablesToNodes(model_part, rThisVariable);
}

void VectorTransferVariablesToNodes(VariableTransferUtility& dummy,
        ModelPart& model_part, Variable<Vector>& rThisVariable)
{
    dummy.TransferVariablesToNodes(model_part, rThisVariable);
}

void DoubleTransferVariablesToGaussPoints(VariableTransferUtility& dummy,
        ModelPart& source_model_part, ModelPart& target_model_part, Variable<double>& rThisVariable)
{
    dummy.TransferVariablesToGaussPoints(source_model_part, target_model_part, rThisVariable);
}

void VectorTransferVariablesToGaussPoints(VariableTransferUtility& dummy,
        ModelPart& source_model_part, ModelPart& target_model_part, Variable<Vector>& rThisVariable)
{
    dummy.TransferVariablesToGaussPoints(source_model_part, target_model_part, rThisVariable);
}

void  AddCustomUtilitiesToPython()
{
    class_<DeactivationUtility, boost::noncopyable >
    ( "DeactivationUtility", init<>() )
    .def( init<int>() )
    .def( "Deactivate", &DeactivationUtility::Deactivate )
    .def( "Reactivate", &DeactivationUtility::Reactivate )
    .def( "ReactivateStressFree", &DeactivationUtility::ReactivateStressFree )
    .def( "ReactivateAll", &DeactivationUtility::ReactivateAll )
    .def( "Initialize", &DeactivationUtility::Initialize )
    .def( "GetName", &DeactivationUtility::GetName<Element> )
    .def( "GetName", &DeactivationUtility::GetName<Condition> )
    ;

    class_<VariableTransferUtility, boost::noncopyable >
    ( "VariableTransferUtility", init<>() )
    .def(init<VariableTransferUtility::LinearSolverType::Pointer>())
    .def( "TransferNodalVariables", &VariableTransferUtility::TransferNodalVariables )
    .def( "TransferConstitutiveLawVariables", &VariableTransferUtility::TransferConstitutiveLawVariables )
    .def( "TransferInSituStress", &VariableTransferUtility::TransferInSituStress )
    .def( "TransferPrestress", &VariableTransferUtility::TransferPrestress )
    .def( "TransferPrestressIdentically", &VariableTransferUtility::TransferPrestressIdentically )
    .def( "TransferSpecificVariable", &VariableTransferUtility::TransferSpecificVariable )
    .def( "InitializeModelPart", &VariableTransferUtility::InitializeModelPart )
    .def("TransferVariablesToNodes", &DoubleTransferVariablesToNodes)
    .def("TransferVariablesToNodes", &VectorTransferVariablesToNodes)
    .def("TransferVariablesToGaussPoints", &DoubleTransferVariablesToGaussPoints)
    .def("TransferVariablesToGaussPoints", &VectorTransferVariablesToGaussPoints)
    ;

#ifdef _OPENMP
    class_<ParallelVariableTransferUtility, boost::noncopyable >
    ( "ParallelVariableTransferUtility",
      init<>() )
    .def( "TransferNodalVariables", &ParallelVariableTransferUtility::TransferNodalVariables )
    .def( "TransferConstitutiveLawVariables", &ParallelVariableTransferUtility::TransferConstitutiveLawVariables )
    .def( "TransferInSituStress", &ParallelVariableTransferUtility::TransferInSituStress )
    .def( "InitializeModelPart", &ParallelVariableTransferUtility::InitializeModelPart )
    ;
#endif

    class_<ContactUtility, boost::noncopyable >
    ( "ContactUtility",
      init<int>() )
    .def( "SetUpContactConditions", &ContactUtility::SetUpContactConditions )
    .def( "SetUpContactConditionsLagrangeTying", &ContactUtility::SetUpContactConditionsLagrangeTying )
    .def( "Update", &ContactUtility::Update )
    .def( "IsConverged", &ContactUtility::IsConverged )
    .def( "Clean", &ContactUtility::Clean )
    .def( "CleanLagrangeTying", &ContactUtility::CleanLagrangeTying )
    ;
// VM
    class_<VolumeUtility, boost::noncopyable >
    ( "VolumeUtility",
      init<int>() )
    .def( "Calculate_this_Volume", &VolumeUtility::CalculateVolume ) // VM
    ;
//VM

    class_<RestartUtility, boost::noncopyable >
    ( "RestartUtility",
      init< std::string const& >() )
    .def( "ChangeFileName", &RestartUtility::ChangeFileName )
    .def( "StoreNodalVariables", &RestartUtility::StoreNodalVariables )
    .def( "WriteNodalVariables", &RestartUtility::WriteNodalVariables )
    .def( "StoreInSituStress", &RestartUtility::StoreInSituStress )
    .def( "WriteConstitutiveLawVariables", &RestartUtility::WriteConstitutiveLawVariables )
    .def( "StoreConstitutiveLawVariables", &RestartUtility::StoreConstitutiveLawVariables )
    .def( "WriteInSituStress", &RestartUtility::WriteInSituStress )
    ;

    class_<NodeSnappingUtility, boost::noncopyable >
    ( "NodeSnappingUtility",
      init<>() )
    .def( "MoveNode", &NodeSnappingUtility::MoveNode )
    .def( "AdjustNodes", &NodeSnappingUtility::AdjustNodes )
    .def( "AdjustToCircle", &NodeSnappingUtility::AdjustToCircle )
    .def( "AdjustToCylinder", &NodeSnappingUtility::AdjustToCylinder )
    .def( "AdjustToClosedCylinder", &NodeSnappingUtility::AdjustToClosedCylinder )
    .def( "IdentifyInsideElements", &NodeSnappingUtility::IdentifyInsideElements )
    .def( "SetInsituStress", &NodeSnappingUtility::SetInsituStress )
    .def( "ExtractCapNodes", &NodeSnappingUtility::ExtractCapNodes )
    .def( "TestElements", &NodeSnappingUtility::TestElements )
    ;

    class_<OutputUtility, boost::noncopyable >
    ( "OutputUtility",
      init<>() )
    .def( "GetStrain", &OutputUtility::GetStrain )
    .def( "GetStress", &OutputUtility::GetStress )
    .def( "GetInternalVariables", &OutputUtility::GetInternalVariables )
    ;

    def( "AddNewRigidBody3D", AddNewRigidBody3D );
    def( "AddNewRigidBodyAndSpring3D", AddNewRigidBodyAndSpring3D );
    ;

    /*
                class_<Detect_Elements_And_Nodes, boost::noncopyable >
                        ("DetectElementsAndNodes", init<ModelPart&, int >() )
          .def("DetectNode",              &Detect_Elements_And_Nodes::Detect_Node_To_Be_Splitted)
                        .def("DetectElements",          &Detect_Elements_And_Nodes::Detect_Elements_To_Be_Splitted)
                        .def("CalculateMapFailure",     &Detect_Elements_And_Nodes::Calculate_Map_Failure)
                        .def("Finalize",                &Detect_Elements_And_Nodes::Finalize)
                        ;
    */
    class_<Smoothing_Utility, boost::noncopyable >
    ( "SmoothingUtility", init<ModelPart&, int >() )
    .def( "WeightedRecoveryGradients", &Smoothing_Utility::WeightedRecoveryGradients<double> )
    .def( "WeightedRecoveryGradients", &Smoothing_Utility::WeightedRecoveryGradients<Matrix> ) // for matrices
    .def( "InterpolatedRecoveryGradients", &Smoothing_Utility::InterpolatedRecoveryGradients<Matrix> )
    .def( "SettingNodalValues", &Smoothing_Utility::SettingNodalValues )
    .def( "RecomputeValuesForNewMesh", &Smoothing_Utility::Recompute_Values_For_New_Mesh )
    .def( "Finalize", &Smoothing_Utility::Finalize )
    .def( "SettingNodalValues", &Smoothing_Utility::SettingNodalValues )
    ;

    class_<Disconnect_Triangle_Utilities, boost::noncopyable >
    ( "DisconnectTriangle", init<ModelPart&>() )
    .def( "DisconnectElements", &Disconnect_Triangle_Utilities::Disconnect_Elements )
    ;

    class_<Intra_Fracture_Triangle, boost::noncopyable >
    ( "IntraFractureTriangle", init<ModelPart&, int >() )
    .def( "DetectAndSplitElements",              &Intra_Fracture_Triangle::Detect_And_Split_Elements )
    ;

    class_<Inter_Fracture_Triangle, boost::noncopyable >
    ( "InterFractureTriangle", init<ModelPart&, int >() )
    .def( "DetectAndSplitElementsHeuristicFormula", &Inter_Fracture_Triangle::Detect_And_Split_Elements_Heuristic_Formula )
    .def( "DetectAndSplitElements",              &Inter_Fracture_Triangle::Detect_And_Split_Elements )
    .def( "Finalize",                            &Inter_Fracture_Triangle::Finalize )
    ;

    class_<Inter_Fracture_Tetrahedra, boost::noncopyable >
    ( "InterFractureTetrahedra", init<ModelPart&, int >() )
    .def( "DetectAndSplitElements",              &Inter_Fracture_Tetrahedra::Detect_And_Split_Elements )
    ;

    class_<DofUtility, boost::noncopyable >
    ( "DofUtility", init<>() )
    .def( "ListDofs", &DofUtility::ListDofs )
    ;
}
}  // namespace Python.
}  // namespace Kratos.

#endif // KRATOS_ADD_CUSTOM_UTILITIES_TO_PYTHON_H_INCLUDED  defined

ext: cpp


text:
/*    Copyright 2009 10gen Inc.
 *
 *    This program is free software: you can redistribute it and/or  modify
 *    it under the terms of the GNU Affero General Public License, version 3,
 *    as published by the Free Software Foundation.
 *
 *    This program 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 Affero General Public License for more details.
 *
 *    You should have received a copy of the GNU Affero General Public License
 *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 *    As a special exception, the copyright holders give permission to link the
 *    code of portions of this program with the OpenSSL library under certain
 *    conditions as described in each individual source file and distribute
 *    linked combinations including the program with the OpenSSL library. You
 *    must comply with the GNU Affero General Public License in all respects
 *    for all of the code used other than as permitted herein. If you modify
 *    file(s) with this exception, you may extend this exception to your
 *    version of the file(s), but you are not obligated to do so. If you do not
 *    wish to do so, delete this exception statement from your version. If you
 *    delete this exception statement from all source files in the program,
 *    then also delete it in the license file.
 */

#include "mongo/platform/basic.h"

#include "mongo/util/net/ssl_manager.h"

#include <boost/algorithm/string.hpp>
#include <string>
#include <vector>

#include "mongo/bson/bsonobjbuilder.h"
#include "mongo/config.h"
#include "mongo/db/server_parameters.h"
#include "mongo/transport/session.h"
#include "mongo/util/mongoutils/str.h"
#include "mongo/util/net/ssl_options.h"
#include "mongo/util/net/ssl_types.h"
#include "mongo/util/text.h"

namespace mongo {

namespace {

const transport::Session::Decoration<SSLPeerInfo> peerInfoForSession =
    transport::Session::declareDecoration<SSLPeerInfo>();

/**
 * Configurable via --setParameter disableNonSSLConnectionLogging=true. If false (default)
 * if the sslMode is set to preferSSL, we will log connections that are not using SSL.
 * If true, such log messages will be suppressed.
 */
ExportedServerParameter<bool, ServerParameterType::kStartupOnly>
    disableNonSSLConnectionLoggingParameter(ServerParameterSet::getGlobal(),
                                            "disableNonSSLConnectionLogging",
                                            &sslGlobalParams.disableNonSSLConnectionLogging);

ExportedServerParameter<std::string, ServerParameterType::kStartupOnly>
    setDiffieHellmanParameterPEMFile(ServerParameterSet::getGlobal(),
                                     "opensslDiffieHellmanParameters",
                                     &sslGlobalParams.sslPEMTempDHParam);
}  // namespace

SSLPeerInfo& SSLPeerInfo::forSession(const transport::SessionHandle& session) {
    return peerInfoForSession(session.get());
}

SSLParams sslGlobalParams;

const SSLParams& getSSLGlobalParams() {
    return sslGlobalParams;
}

class OpenSSLCipherConfigParameter
    : public ExportedServerParameter<std::string, ServerParameterType::kStartupOnly> {
public:
    OpenSSLCipherConfigParameter()
        : ExportedServerParameter<std::string, ServerParameterType::kStartupOnly>(
              ServerParameterSet::getGlobal(),
              "opensslCipherConfig",
              &sslGlobalParams.sslCipherConfig) {}
    Status validate(const std::string& potentialNewValue) final {
        if (!sslGlobalParams.sslCipherConfig.empty()) {
            return Status(
                ErrorCodes::BadValue,
                "opensslCipherConfig setParameter is incompatible with net.ssl.sslCipherConfig");
        }
        // Note that there is very little validation that we can do here.
        // OpenSSL exposes no API to validate a cipher config string. The only way to figure out
        // what a string maps to is to make an SSL_CTX object, set the string on it, then parse the
        // resulting STACK_OF object. If provided an invalid entry in the string, it will silently
        // ignore it. Because an entry in the string may map to multiple ciphers, or remove ciphers
        // from the final set produced by the full string, we can't tell if any entry failed
        // to parse.
        return Status::OK();
    }
} openSSLCipherConfig;

#ifdef MONGO_CONFIG_SSL

namespace {
void canonicalizeClusterDN(std::vector<std::string>* dn) {
    // remove all RDNs we don't care about
    for (size_t i = 0; i < dn->size(); i++) {
        std::string& comp = dn->at(i);
        boost::algorithm::trim(comp);
        if (!mongoutils::str::startsWith(comp.c_str(), "DC=") &&
            !mongoutils::str::startsWith(comp.c_str(), "O=") &&
            !mongoutils::str::startsWith(comp.c_str(), "OU=")) {
            dn->erase(dn->begin() + i);
            i--;
        }
    }
    std::stable_sort(dn->begin(), dn->end());
}
}  // namespace

bool SSLConfiguration::isClusterMember(StringData subjectName) const {
    std::vector<std::string> clientRDN = StringSplitter::split(subjectName.toString(), ",");
    std::vector<std::string> serverRDN = StringSplitter::split(serverSubjectName, ",");

    canonicalizeClusterDN(&clientRDN);
    canonicalizeClusterDN(&serverRDN);

    if (clientRDN.size() == 0 || clientRDN.size() != serverRDN.size()) {
        return false;
    }

    for (size_t i = 0; i < serverRDN.size(); i++) {
        if (clientRDN[i] != serverRDN[i]) {
            return false;
        }
    }
    return true;
}

BSONObj SSLConfiguration::getServerStatusBSON() const {
    BSONObjBuilder security;
    security.append("SSLServerSubjectName", serverSubjectName);
    security.appendBool("SSLServerHasCertificateAuthority", hasCA);
    security.appendDate("SSLServerCertificateExpirationDate", serverCertificateExpirationDate);
    return security.obj();
}

SSLManagerInterface::~SSLManagerInterface() {}

SSLConnectionInterface::~SSLConnectionInterface() {}

#endif

}  // namespace mongo

max_issues_repo_name: visemet/mongo

max_forks_repo_name: visemet/mongo


text:
#include <boost/any.hpp>
#include <iostream>

int main()
{
  boost::any a = 1;
  std::cout << boost::any_cast<int>(a) << std::endl;
  a = 3.14;
  std::cout << boost::any_cast<double>(a) << std::endl;
  a = true;
  std::cout << boost::any_cast<bool>(a) << std::endl;
}

max_stars_repo_path: Boost/The Boost C++ Libraries/src/14.3.3/main.cpp


text:
/* Copyright(C)
* For free
* All right reserved
*
*/
/**
* @file RCFPublisherImpl.hpp
* @brief 发布者实现类
* @author highway-9, 787280310@qq.com
* @version 1.1.0
* @date 2016-01-12
*/

#ifndef _RCFPUBLISHERIMPL_H
#define _RCFPUBLISHERIMPL_H

#include <RCF/RCF.hpp>
#include <assert.h>
#include <iostream>
#include <map>
#include <boost/smart_ptr.hpp>
#include <boost/thread.hpp>

/**
* @brief 发布者参数
*/
class PublisherParam
{
public:
    PublisherParam()
        : m_onSubscriberConnect(NULL),
        m_onSubscriberDisconnect(NULL)
    {
        // Do nothing
    }

    std::string m_topicName;                                 ///< 发布的主题
    RCF::OnSubscriberConnect m_onSubscriberConnect;          ///< 响应订阅者连接的函数
    RCF::OnSubscriberDisconnect m_onSubscriberDisconnect;    ///< 响应订阅者断开连接的函数
};

/**
* @brief 发布者实现类
*
* @tparam I_RCFMessageHandler 类类型
*/
template<typename I_RCFMessageHandler>
class RCFPublisherImpl
{
public:
    typedef typename boost::shared_ptr<RCF::Publisher<I_RCFMessageHandler> > RcfPublisherPtr;
    typedef typename std::map<std::string, RcfPublisherPtr > RcfPublisherMap;

    /**
    * @brief RCFPublisherImpl 构造函数
    *
    * @param port 发布的端口号
    */
    RCFPublisherImpl(unsigned int port)
        : m_port(port)
    {
        m_rcfInit.reset();
        m_rcfServer.reset();
    }

    ~RCFPublisherImpl()
    {
        bool ok = stop();
        assert(ok);
    }

    /**
    * @brief start 开启服务器
    *
    * @note 开启服务器之后，才能调用createPublisher函数
    *
    * @return 成功返回true，否则返回false
    */
    bool start()
    {
        try
        {
            if (m_rcfInit == NULL)
            {
                m_rcfInit = boost::make_shared<RCF::RcfInitDeinit>();
            }

            if (m_rcfServer == NULL)
            {
                m_rcfServer = boost::make_shared<RCF::RcfServer>(RCF::TcpEndpoint(m_port));
                m_rcfServer->start();
            }
        }
        catch (const RCF::Exception& e)
        {
            std::cout << "Error: " << e.getErrorString() << std::endl;
            return false;
        }

        return true;
    }

    /**
    * @brief createPublisher 通过主题来创建发布者
    *
    * @param param 发布者参数
    *
    * @note 调用该函数之前，请先调用start函数开启服务器
    *
    * @return 成功返回true，否则返回false
    */
    bool createPublisher(const PublisherParam& param)
    {
        boost::lock_guard<boost::mutex> locker(m_mutex);

        if (isPublisherExists(param.m_topicName))
        {
            return false;
        }

        try
        {
            assert(m_rcfServer != NULL);
            assert(param.m_onSubscriberConnect != NULL);
            assert(param.m_onSubscriberDisconnect != NULL);

            RCF::PublisherParms pubParms;
            pubParms.setTopicName(param.m_topicName);
            pubParms.setOnSubscriberConnect(param.m_onSubscriberConnect);
            pubParms.setOnSubscriberDisconnect(param.m_onSubscriberDisconnect);
            RcfPublisherPtr rcfPublisher = m_rcfServer->createPublisher<I_RCFMessageHandler>(pubParms);
            m_rcfPublisherMap.insert(std::make_pair(param.m_topicName, rcfPublisher));
        }
        catch (const RCF::Exception& e)
        {
            std::cout << "Error: " << e.getErrorString() << std::endl;
            return false;
        }

        return true;
    }

    /**
    * @brief stop 停止发布者服务器
    *
    * @return 成功返回true，否则返回false
    */
    bool stop()
    {
        bool ok = closeAllPublisher();
        if (!ok)
        {
            return false;
        }

        try
        {
            assert(m_rcfServer != NULL);
            m_rcfServer->stop();
        }
        catch (const RCF::Exception& e)
        {
            std::cout << "Error: " << e.getErrorString() << std::endl;
            return false;
        }

        return true;
    }

    /**
    * @brief publishObject 通过主题名称得到发布者对象
    *
    * @param topicName 主题名称
    *
    * @return 成功返回发布和对象，失败返回NULL
    */
    RcfPublisherPtr rcfPublishObject(const std::string& topicName)
    {
        typename RcfPublisherMap::iterator iter = m_rcfPublisherMap.find(topicName);
        if (iter != m_rcfPublisherMap.end())
        {
            return iter->second;
        }

        return RcfPublisherPtr();
    }

    /**
    * @brief closePublisher 通过主题来停止发布者
    *
    * @param topicName 主题名称
    *
    * @return 成功返回true，否则返回false
    */
    bool closePublisher(const std::string& topicName)
    {
        boost::lock_guard<boost::mutex> locker(m_mutex);

        typename RcfPublisherMap::iterator iter = m_rcfPublisherMap.find(topicName);
        if (iter != m_rcfPublisherMap.end())
        {
            try
            {
                iter->second->close();
            }
            catch (const RCF::Exception& e)
            {
                std::cout << "Error: " << e.getErrorString() << std::endl;
                return false;
            }

            m_rcfPublisherMap.erase(iter);
            return true;
        }

        return false;
    }

    /**
    * @brief closeAllPublisher 停止所有的发布者
    *
    * @return 成功返回true，否则返回false
    */
    bool closeAllPublisher()
    {
        boost::lock_guard<boost::mutex> locker(m_mutex);

        typename RcfPublisherMap::iterator begin = m_rcfPublisherMap.begin();
        typename RcfPublisherMap::iterator end = m_rcfPublisherMap.end();
        while (begin != end)
        {
            try
            {
                begin->second->close();
            }
            catch (const RCF::Exception& e)
            {
                std::cout << "Error: " << e.getErrorString() << std::endl;
                return false;
            }

            ++begin;
        }

        m_rcfPublisherMap.clear();
        return true;
    }

private:
    /**
    * @brief isPublisherExists 判断发布者是否存在
    *
    * @param topicName 主题名称
    *
    * @return 存在返回true，否则返回false
    */
    bool isPublisherExists(const std::string& topicName)
    {
        typename RcfPublisherMap::const_iterator iter = m_rcfPublisherMap.find(topicName);
        if (iter != m_rcfPublisherMap.end())
        {
            return true;
        }

        return false;
    }

private:
    typedef boost::shared_ptr<RCF::RcfInitDeinit> RcfInitDeinitPtr;
    RcfInitDeinitPtr m_rcfInit;              ///< RCF服务器初始化对象

    typedef boost::shared_ptr<RCF::RcfServer> RcfServerPtr;
    RcfServerPtr m_rcfServer;                ///< RCF服务器对象

    unsigned int m_port;                     ///< 发布的端口号
    RcfPublisherMap m_rcfPublisherMap;       ///< 发布者map，key：主题名，value：发布者
    boost::mutex m_mutex;                    ///< 发布者map互斥锁
};

#endif

ext: hpp


text:
#include <stream.hxx>
#include <backup.hxx>
#include <backupprocesses.hxx>
#include <xlogdefs.hxx>
#include <memorybuffer.hxx>
#include <boost/log/trivial.hpp>

/* Required for select() */
extern "C" {
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>
}

using namespace pgbckctl;

/******************************************************************************
 * Implementation of WALStreamerProcess
 ******************************************************************************/

WALStreamerProcess::WALStreamerProcess(PGconn *prepared_connection,
                                       StreamIdentification streamident) {

  this->current_state = ARCHIVER_STARTUP;
  this->pgconn        = prepared_connection;
  this->streamident   = streamident;
  this->stopHandler   = nullptr;

  /*
   * prepared_connection needs to be a connected
   * libpq connection, otherwise we have to error out.
   */
  if (!((this->pgconn != NULL)
        && (PQstatus(this->pgconn) == CONNECTION_OK)))
    throw StreamingFailure("WAL stream not connected");
}

WALStreamerProcess::~WALStreamerProcess() {

  /*
   * Don't close the inherited PostgreSQL connection here.
   *
   * We are expecting to operate as a sub
   * on a calling PGStream handle, which
   * does all the legwork for us.
   */
}

ArchiverState WALStreamerProcess::reason() {
  return this->current_state;
}

ArchiverState WALStreamerProcess::receivePoll() {

  /*
   * Polling code like src/bin/pg_basebackup/receivelog.c
   * is implemented. See also
   *
   * https://www.postgresql.org/docs/10/static/libpq-example.html
   *
   * for an example.
   */

  int result;
  fd_set input_mask;
  int maxfd;
  timeval select_timeout;
  timeval *select_timeoutptr = NULL;

  /*
   * We need the PG socket to operate on.
   */
  int serversocket = PQsocket(this->pgconn);

  if (serversocket < 0) {
    std::ostringstream oss;
    oss << "error polling on server connection: "
        << PQerrorMessage(this->pgconn);
    throw StreamingFailure(oss.str());
  }

  FD_ZERO(&input_mask);
  FD_SET(serversocket, &input_mask);
  maxfd = serversocket;
  this->timeoutSelectValue(&select_timeout);

  /*
   * Calculate timeout value.
   */
  if (this->timeout > 0) {
    select_timeoutptr = &select_timeout;
  }

  result = ::select(maxfd + 1, &input_mask, NULL, NULL, select_timeoutptr);

  /*
   * Checkout what happened to select() after returning.
   */
  if (result < 0) {
    /*
     * A negative return code here can also be originated
     * from a EINTR, check. Otherwise we bail out hard.
     */
    if (errno == EINTR) {
      /* Archiver stream is interrupted by signal */
      return ARCHIVER_STREAMING_INTR;
    }

    /* ... else this is really an error. */
    return ARCHIVER_STREAMING_ERROR;
  }

  if ( (result > 0)
       && (FD_ISSET(serversocket, &input_mask)) ) {
    /* data is available */
    return ARCHIVER_STREAMING;
  }

  /* timeout on waiting for data */
  return ARCHIVER_STREAMING_TIMEOUT;
}

void WALStreamerProcess::timeoutSelectValue(timeval *timeoutptr) {

  if (timeoutptr == NULL) {
    return;
  } else {

    timeoutptr->tv_sec = this->timeout / 1000L;
    timeoutptr->tv_usec = (this->timeout % 1000L) / 1000L;

  }

}

ArchiverState WALStreamerProcess::sendStatusUpdate() {

  ReceiverStatusUpdateMessage rsum(this->pgconn);

#ifdef __DEBUG_XLOG__
  BOOST_LOG_TRIVIAL(debug) << " ... sending status update to primary ";
  BOOST_LOG_TRIVIAL(debug) << "     -> write position: "
                           << PGStream::encodeXLOGPos(this->streamident.write_position);
  BOOST_LOG_TRIVIAL(debug) << "     -> flush position: "
                           << PGStream::encodeXLOGPos(this->streamident.flush_position);
  BOOST_LOG_TRIVIAL(debug) << "     -> last reported flush position: "
                           << PGStream::encodeXLOGPos(this->streamident.last_reported_flush_position);
  BOOST_LOG_TRIVIAL(debug) << "     -> apply position: "
                           << PGStream::encodeXLOGPos(this->streamident.apply_position);
#endif

  if (this->streamident.flush_position == InvalidXLogRecPtr) {

    rsum.setStatus(this->streamident.write_position,
                   this->streamident.last_reported_flush_position,
                   this->streamident.apply_position);

  } else {

    rsum.setStatus(this->streamident.write_position,
                   this->streamident.flush_position,
                   this->streamident.apply_position);

  }

  /* Always report flush position back to upstream */
  rsum.reportFlushPosition();

  rsum.send();
  return this->current_state;

}

ArchiverState WALStreamerProcess::handleReceive(char **buffer, int *bufferlen) {

  /* Holds length of incoming buffer data */
  ArchiverState status = ARCHIVER_STREAMING_ERROR;

  /* bufferlen should be not null! */
  if (bufferlen == NULL) {
    throw StreamingFailure("buffer length attribute cannot be NULL");
  }

  if (*buffer != NULL) {
    PQfreemem(*buffer);
  }

  /* make sure buffer is NULL */
  *buffer = NULL;

  /* Get input data, if any */
  *bufferlen = PQgetCopyData(this->pgconn, buffer, 1);

  if (*bufferlen == 0) {

    /* no data yet, poll on input from WAL sender */
    if (buffer != NULL) {
      PQfreemem(*buffer);
      *buffer = NULL;
    }

    status = this->current_state = this->receivePoll();

    /* check for errors, timeout et al. */
    if ( (status != ARCHIVER_STREAMING)
         && (status != ARCHIVER_STREAMING_INTR) ) {
      return status;
    }

    /* Consume input */
    if (PQconsumeInput(this->pgconn) == 0) {

      /*
       * Okay, there's some trouble here, get the error
       * and bail out hard.
       */
      status = this->current_state = ARCHIVER_STREAMING_ERROR;

    }

    /* Read data into buffer, again */
    *bufferlen = PQgetCopyData(this->pgconn, buffer, 1);

    /*
     * If there's still zero bytes just return ARCHIVER_STREAMING_NO_DATA.
     */
    if (*bufferlen == 0) {
      status = this->current_state = ARCHIVER_STREAMING_NO_DATA;
    }

    /*
     * In case the connection was terminated, check
     * for a buffer length set to < -1.
     */
    if (*bufferlen <= -1) {
      status = this->current_state = ARCHIVER_END_POSITION;
    }

  } else if (*bufferlen == -1) {

    /*
     * Indicate end-of-stream, either shutdown
     * or end of current timeline
     */
    status = this->current_state = ARCHIVER_END_POSITION;

  } else if (*bufferlen == -2) {

    /* Oops, something went wrong here */
    status = this->current_state = ARCHIVER_STREAMING_ERROR;

  } else if (*bufferlen > 0) {

    status = this->current_state = ARCHIVER_STREAMING;

  }

  return status;
}

std::string WALStreamerProcess::xlogpos() {

  if (this->current_state != ARCHIVER_STREAMING_ERROR)
    return this->streamident.xlogpos;
  else
    throw StreamingFailure("attempt to query xlog position from wal streamer not in streaming state");

}

void WALStreamerProcess::setBackupHandler(std::shared_ptr<TransactionLogBackup> backupHandler) {

  this->backupHandler = backupHandler;

}

void WALStreamerProcess::handleMessage(XLOGStreamMessage *message) {

  char msgType;

  /*
   * message shouldn't be a nullptr...
   */
  if (message == nullptr)
    return;

  msgType = message->what();

  /*
   * WAL data (XLOGDataStreamMessage object) message should update xlogpos
   */
  switch(msgType) {
  case 'w':
    {
      XLOGDataStreamMessage *datamsg = dynamic_cast<XLOGDataStreamMessage *>(message);
      XLogRecPtr serverpos;

      this->streamident.xlogpos
        = PGStream::encodeXLOGPos(datamsg->getXLOGStartPos());

#ifdef __DEBUG_XLOG__
      BOOST_LOG_TRIVIAL(debug) << "NEW XLOG POS " << this->streamident.xlogpos;
#endif

      /* get server position, reported by the connected WAL sender */
      serverpos = datamsg->getXLOGServerPos();

#ifdef __DEBUG_XLOG__
      BOOST_LOG_TRIVIAL(debug) << "SERVER XLOG POS "
                               << PGStream::encodeXLOGPos(serverpos);
#endif
      this->streamident.server_position = serverpos;

      /*
       * Update XLOG write position and, if flushed, also
       * the flush position.
       *
       * NOTE: If no backup handler was configured,
       *       Flush is currently the same as the
       *       current write position.
       *
       * Our backup handler does all the necessary work
       * to spool the transaction logs into the archive.
       */
      this->streamident.write_position = datamsg->getXLOGServerPos();
      this->streamident.updateStartSegmentWriteOffset();

      if (this->backupHandler != nullptr) {

        /*
         * Everything should be in shape so far. A current
         * WAL segment file should be allocated and we can go
         * forward and spool the XLOG message data block into that
         * current WAL file. Though, we must take care for
         * a WAL segment boundary, in that case we have to switch to
         * new WAL segment file, again.
         *
         * We must remember the current flush position in
         * last_reported_flush_position, since we want to continously
         * report them back to upstream. In case no XLOG switch happened,
         * the backuphandler write() method will set an InvalidXLogRecPtr!
         */

        this->streamident.write_position
          = this->backupHandler->write(datamsg,
                                       this->streamident.flush_position,
                                       this->streamident.timeline);

        if (this->streamident.flush_position != InvalidXLogRecPtr) {

          this->streamident.last_reported_flush_position
            = this->streamident.flush_position;

        }

#ifdef __DEBUG_XLOG__
        BOOST_LOG_TRIVIAL(debug) << "DEBUG: flush position after write(): "
                                 << PGStream::encodeXLOGPos(this->streamident.flush_position);
#endif

      } else { /* if ... backupHandler != nullptr */

        this->streamident.flush_position = datamsg->getXLOGServerPos();
        this->streamident.last_reported_flush_position
          = this->streamident.flush_position;

      }

      break;
    }

  case 'k':
    {
      PrimaryFeedbackMessage *pm = dynamic_cast<PrimaryFeedbackMessage *>(message);

#ifdef __DEBUG_XLOG__
      BOOST_LOG_TRIVIAL(debug) << "primary feedback message";
#endif

      /*
       * Upstream wants an immediate status update?
       */
      if (pm->responseRequested()) {

#ifdef __DEBUG_XLOG__
        BOOST_LOG_TRIVIAL(debug) << "receiver status update forced";
#endif

        /* ...send it */
        this->sendStatusUpdate();

        /*
         * Update timeout interval.
         */
        this->last_status_update = CPGBackupCtlBase::current_hires_time_point();

      }

      break;

    }
  default:
    {
      /* unhandled message type */
#ifdef __DEBUG_XLOG__
      BOOST_LOG_TRIVIAL(debug) << "unhandled message type " << msgType;
#endif
      break;
    }

  } /* switch...msgType */

}

bool WALStreamerProcess::stopHandlerWantsExit() {

  if (this->stopHandler != nullptr) {
    return this->stopHandler->check();
  } else
    return false;

}

void WALStreamerProcess::assignStopHandler(JobSignalHandler *handler) {

  /* if handler is NULL, this is a no-op */
  if (handler == nullptr) {
    return;
  }

  this->stopHandler = handler;
}

void WALStreamerProcess::setReceiverStatusTimeout(long value) {

  /*
   * We don't allow values lower than ower
   * fixed internal select() timeout.
   */
  if (value < this->timeout) {
    std::ostringstream oss;

    oss << "receiver status timeout cannot be lower than " << this->timeout;
    throw StreamingFailure(oss.str());
  }

  this->receiver_status_timeout = value;

}

bool WALStreamerProcess::receive() {

  bool can_continue = false;
  char *buffer = NULL; /* temporary recv buffer, handled by libpq */
  int bufferlen = 0;
  ArchiverState reason;

  /*
   * If we're not in streaming state, handle the state
   * accordingly.
   */
  if (this->current_state == ARCHIVER_STREAMING_ERROR) {
    /* error condition on streaming connection, abort */
    throw StreamingFailure("error on streaming connection");
  }

  if (this->current_state == ARCHIVER_STARTUP) {
    throw StreamingFailure("attempt to stream from uninitialized streaming handle, call start() before");
  }

  if (this->current_state == ARCHIVER_END_POSITION) {
    throw StreamingFailure("new log segment requested, use finalize() instead");
  }

  BOOST_LOG_TRIVIAL(info) << "entering WAL streaming receive() ";

  /*
   * Initialize status update start interval.
   */
  this->last_status_update = CPGBackupCtlBase::current_hires_time_point();

  this->current_state = reason = this->handleReceive(&buffer, &bufferlen);
  while (reason != ARCHIVER_END_POSITION
         && reason != ARCHIVER_SHUTDOWN
         && reason != ARCHIVER_STREAMING_ERROR
         && reason != ARCHIVER_TIMELINE_SWITCH
         && reason != ARCHIVER_STARTUP
         && reason != ARCHIVER_START_POSITION) {

    XLOGStreamMessage *message = nullptr;

    /**
     * Before doing anything, check whether our stop
     * handler wants to exit.
     */
    if (this->stopHandlerWantsExit()) {
      reason = this->current_state = ARCHIVER_SHUTDOWN;
      this->streamident.status = StreamIdentification::STREAM_PROGRESS_SHUTDOWN;
      break;
    }

    /*
     * Next we check whether we should send a status update message
     * to upstream. This is always the case if our internal
     * timeout value forces us to do.
     *
     * Since we also poll on the socket with a specific timeout, we
     * also recognize this timeout value here.
     */
    if (CPGBackupCtlBase::calculate_duration_ms(this->last_status_update
                                                - std::chrono::milliseconds(this->timeout),
                                                CPGBackupCtlBase::current_hires_time_point())
        >= std::chrono::milliseconds(this->receiver_status_timeout)) {

#ifdef __DEBUG_XLOG__
      BOOST_LOG_TRIVIAL(debug) << "standby status update overdue";
#endif

      this->sendStatusUpdate();
      this->last_status_update = CPGBackupCtlBase::current_hires_time_point();

    }

    /*
     * If not streaming, don't try to handle XLOG messages.
     */
    if (reason != ARCHIVER_STREAMING) {

      /*
       * Before trying next, copy local buffer ...
       */
      if (buffer != NULL) {
        PQfreemem(buffer);
        buffer = NULL;
      }

      /* ..next try */
      reason = this->handleReceive(&buffer, &bufferlen);
      continue;
    }

    /*
     * Interpret buffer, create a corresponding WAL message object.
     */
    this->receiveBuffer.allocate(bufferlen);
    this->receiveBuffer.write(buffer, bufferlen, 0);

    /*
     * Free copy buffers, they are copied over
     * into our local receive buffer, so we don't
     * need them anymore.
     */
    if (buffer != NULL) {
      PQfreemem(buffer);
      buffer = NULL;
    }

    message = XLOGStreamMessage::message(this->pgconn,
                                         this->receiveBuffer,
                                         this->streamident.wal_segment_size);

#ifdef __DEBUG_XLOG__
    BOOST_LOG_TRIVIAL(debug) << "write XLOG message ";
#endif

    try {
      if (message != nullptr) {

        this->handleMessage(message);

      }
    } catch(CPGBackupCtlFailure &e) {
      if (message != nullptr)
        delete message;

      throw e;
    }

    delete message;

    /* ..next try */
    this->current_state = reason = this->handleReceive(&buffer, &bufferlen);

  }

  /*
   * Internal handleReceive() loop exited, check
   * further actions depending on current state.
   */
  if (this->current_state == ARCHIVER_STREAMING_NO_DATA) {

    BOOST_LOG_TRIVIAL(warning) << "no data, continue ... ";
    can_continue = true;

  }

  if (this->current_state == ARCHIVER_STREAMING_ERROR) {

    BOOST_LOG_TRIVIAL(error) << "failure on PQgetCopyData(): "
                             << PQerrorMessage(this->pgconn);
    can_continue = false;

  }

  /*
   * Handle end-of-copy conditions.
   *
   * This could either a controlled shutdown
   * of the server or we reached the end of the current
   * timeline.
   */
  if (this->current_state == ARCHIVER_END_POSITION) {

    PGresult *pgres = this->handleEndOfStream();

    if (this->current_state == ARCHIVER_TIMELINE_SWITCH) {

      BOOST_LOG_TRIVIAL(info) << "end of archive stream because of timeline switch";

      /*
       * This is a timeline switch. Handle it accordingly.
       */
      this->endOfStreamTimelineSwitch(pgres,
                                      this->streamident.timeline,
                                      this->streamident.xlogpos);

      BOOST_LOG_TRIVIAL(debug) << "timeline switch to TLI="
                               << this->streamident.timeline
                               << ", XLOGPOS="
                               << this->streamident.xlogpos;

      PQclear(pgres);
      can_continue = true;

    } else if (this->current_state == ARCHIVER_SHUTDOWN) {

      BOOST_LOG_TRIVIAL(info) << "end of archive stream, shutting down";
      /*
       * Server side has terminated, schedule own shutdown accordingly.
       */
      can_continue = false;
    }

  } else if (this->current_state == ARCHIVER_SHUTDOWN) {

    PGresult *res = NULL;

    BOOST_LOG_TRIVIAL(info) << "forced archive stream shutdown";

    /*
     * Inner loop recognized a shutdown request and set the
     * current state to ARCHIVER_SHUTDOWN.
     */
    if ((res != NULL) && (PQresultStatus(res) == PGRES_COPY_IN))
        this->end();

    can_continue = false;

  }

  return can_continue;
}

XLogRecPtr WALStreamerProcess::getCurrentXLOGPos() {
  return this->streamident.write_position;
}

unsigned int WALStreamerProcess::getCurrentTimeline() {
  return this->streamident.timeline;
}

void WALStreamerProcess::endOfStreamTimelineSwitch(PGresult *result,
                                                   unsigned int& timeline,
                                                   std::string& xlogpos) {

  if (this->current_state != ARCHIVER_TIMELINE_SWITCH) {
    throw StreamingFailure("timeline switch requested on unsupported state in streaming connection");
  }

  if (result == NULL) {
    throw StreamingFailure("could not get timeline switch data from streaming connection: result is undefined");
  }

  /*
   * Get timeline and xlogpos from the next timeline.
   */
  if (PQnfields(result) < 2
      || PQntuples(result) != 1) {
    std::ostringstream oss;
    oss << "unexpected result after timeline switch: got "
        << "columns " << PQnfields(result) << " expected " << 2 << ", "
        << "tuples " << PQntuples(result) << "expected " << 1;
    throw StreamingFailure(oss.str());
  }

  /*
   * Read timeline
   */
  timeline = CPGBackupCtlBase::strToInt(PQgetvalue(result, 0, 0));

  /*
   * The XLOG position returned by the result set is encoded. We just
   * return the encoded position, the caller is responsible to encode it
   * into a proper representation.
   */
  xlogpos = PQgetvalue(result, 0, 1);

  /* and we're done */

}

PGresult *WALStreamerProcess::handleEndOfStream() {

  PGresult *result = NULL;

  if (this->current_state != ARCHIVER_END_POSITION) {
    throw CPGBackupCtlFailure("could not call handleEndOfStream() on unterminated WAL stream");
  }

  result = PQgetResult(this->pgconn);

  /*
   * Check whether we are still in COPY IN mode.
   */
  if (result != NULL && (PQresultStatus(result) == PGRES_COPY_IN)) {

    PQclear(result);

    /*
     * Finalize the COPY IN mode on our side. Iff the server is
     * still alive this will succeed.
     *
     * We need to ackknowledge this case in order to get the
     * timeline switch, if any in progress.
     *
     * end() will set the internal Archiver state to ARCHIVER_SHUTDOWN
     * in case of success, but we need to check if we really want
     * to have a shutdown or to continue. This is indicated by
     * a PGRES_TUPLES_OK or PGRES_COMMAND_OK return flag
     * by the returned PGresult object. Check that and set the
     * flag accordingly.
     *
     * PGGRES_TUPLES_OK means that the server had acknowledged our
     * end of stream message, but just want to switch its timeline
     *
     * PGRES_COMMAND_OK is set in case the server is commanded to
     * shutdown.
     *
     * Set the archiver state according to the examined results above.
     *
     * NOTE: end() may throw an exception in case the COPY stream
     *       couldn't be terminated properly.
     */
    try {

      BOOST_LOG_TRIVIAL(debug) << "end of COPY mode, telling upstream to quit";

      result = this->end();

    } catch(StreamingFailure &e) {

      /*
       * WALStreamer::end() might throw, make sure we don't leak
       * PQresult references
       */
      if (result != NULL)
        PQclear(result);

      result = NULL;

      this->current_state = ARCHIVER_STREAMING_ERROR;
      throw e;

    }

  } else {

    /*
     * Reaching this point usually means that the
     * server has disconnected, most likely without having established
     * a working streaming connection before.
     *
     * We now force a WAL Streamer shutdown without sending the
     * end-of-stream message.
     */
    this->current_state = ARCHIVER_SHUTDOWN;

  }

  return result;
}

StreamIdentification WALStreamerProcess::identification() {
  return this->streamident;
}

PGresult *WALStreamerProcess::end() {

  PGresult *result = NULL;

  if (PQputCopyEnd(this->pgconn, NULL) <= 0 || PQflush(this->pgconn)) {

    std::ostringstream oss;

    oss << "could not terminate COPY in progress: "
        << PQerrorMessage(this->pgconn);
    this->current_state = ARCHIVER_STREAMING_ERROR;
    throw StreamingFailure(oss.str());

  }

  /*
   * Make sure we indicate success
   *
   * We must tell the caller whether this was a graceful shutdown
   * or a timeline switch.
   */

  result = PQgetResult(this->pgconn);

  if (result != NULL && ((PQresultStatus(result) == PGRES_TUPLES_OK))) {

    this->current_state = ARCHIVER_TIMELINE_SWITCH;
    this->streamident.status = StreamIdentification::STREAM_PROGRESS_TIMELINE_SWITCH;

  } else if (result != NULL && ((PQresultStatus(result) == PGRES_COMMAND_OK))) {

    this->current_state = ARCHIVER_SHUTDOWN;
    this->streamident.status = StreamIdentification::STREAM_PROGRESS_SHUTDOWN;

  } else {

    std::ostringstream oss;

    oss << "could not terminate COPY in progress: "
        << PQerrorMessage(this->pgconn);
    this->current_state = ARCHIVER_STREAMING_ERROR;
    throw StreamingFailure(oss.str());

  }

  return result;

}

void WALStreamerProcess::finalizeSegment() {

  if (this->current_state != ARCHIVER_END_POSITION) {
    throw StreamingFailure("cannot finalize current transaction log segment");
  }

}

void WALStreamerProcess::start() {

  /*
   * NOTE: we expect a valid initialized Stream Identification handle,
   *       holding the starting XLOG position and a valid replication
   *       slot identifier.
   */
  std::ostringstream query;

  /*
   * buffer for escaped identifiers.
   */
  char escapedLabel[MAXPGPATH];
  char escapedXlogPos[MAXPGPATH];
  int escape_error;

  /* Result from START_REPLICATION command */
  PGresult *res;

  /*
   * If the ArchiveState is already set to ARCHIVER_STREAMING, calling
   * start() is supposed to be an error (we indeed already seem to
   * have issued a start() command, so there's no reason to stack them here.
   * Indicate the state of the WALStreamer to be not correct in this case
   * and throw an error.
   *
   * Also, if the current_state is set to ARCHIVE_END_POSITION we need to
   * handle a graceful streaming connection shutdown, so stop() is
   * required before calling start(). Other states are just handled by
   * receive().
   */
  if (this->current_state == ARCHIVER_STREAMING
      || this->current_state == ARCHIVER_END_POSITION) {
    throw StreamingFailure("invalid call to start() in archiver: already started");
  }

  /*
   * XXX: slot_name and xlogpos are incorporated as
   *      strings here, so we might have to deal with
   *      injection attempts here.
   *
   *      Normally they're comming from a PGStream
   *      object instance, but be paranoid here.
   */
  PQescapeStringConn(this->pgconn,
                     escapedLabel,
                     this->streamident.slot_name.c_str(),
                     this->streamident.slot_name.length(),
                     &escape_error);

  PQescapeStringConn(this->pgconn,
                     escapedXlogPos,
                     this->streamident.xlogpos.c_str(),
                     this->streamident.xlogpos.length(),
                     &escape_error);

  query << "START_REPLICATION SLOT "
        << escapedLabel
        << " PHYSICAL "
        << escapedXlogPos
        << " TIMELINE "
        << this->streamident.timeline
        << ";";

#ifdef __DEBUG__
  BOOST_LOG_TRIVIAL(debug) << "WALStreamer start() query: " << query.str();
#endif

  /*
   * Fire the query ...
   */
  res = PQexec(this->pgconn, query.str().c_str());

  if (PQresultStatus(res) != PGRES_COPY_BOTH) {

    std::ostringstream oss;
    oss << "START_REPLICATION command failed: "
        << PQerrorMessage(this->pgconn);

    /* Don't leak result set */
    PQclear(res);
    throw StreamingFailure(oss.str());

  }

  PQclear(res);

  /*
   * If everything went smoothly, change internal state
   * to ARCHIVER_STREAMING, which enables the caller
   * to use receive()...
   */
  this->current_state = ARCHIVER_STREAMING;
  this->streamident.status = StreamIdentification::STREAM_PROGRESS_STREAMING;

}

/******************************************************************************
 * Implementation of BaseBackupProcess
 ******************************************************************************/

BaseBackupProcess::BaseBackupProcess(PGconn *prepared_connection,
                                     std::shared_ptr<BackupProfileDescr> profile,
                                     std::string systemid,
                                     unsigned long long wal_segment_size) {

  this->current_state = BASEBACKUP_INIT;

  this->pgconn = prepared_connection;
  this->profile = profile;
  this->systemid = systemid;
  this->wal_segment_size = wal_segment_size;

  /*
   * prepared_connection needs to be a connected
   * libpq connection, otherwise we have to error out.
   */
  if (!((this->pgconn != NULL)
        && (PQstatus(this->pgconn) == CONNECTION_OK)))
    throw StreamingFailure("basebackup stream not connected");

}

BaseBackupProcess::~BaseBackupProcess() {

  /*
   * Don't close the associated PostgreSQL
   * connection handle here!
   *
   * We are expected to operate as a sub
   * on a calling PGStream handle, which does
   * all the legwork for us.
   */

}

void BaseBackupProcess::assignStopHandler(JobSignalHandler *handler) {

  /* if handler is NULL, this is a no-op */
  if (handler == nullptr) {
    return;
  }

  this->stopHandler = handler;
}

bool BaseBackupProcess::stopHandlerWantsExit() {

  if (this->stopHandler != nullptr) {
    return this->stopHandler->check();
  } else
    return false;

}

void BaseBackupProcess::start() {

  PGresult *result;
  ExecStatusType es;
  std::ostringstream query;
  char escapedlabel[MAXPGPATH];
  int  escape_error;

  /*
   * Sanity check: profile is not a nullptr.
   */
  query << "BASE_BACKUP";

  if (this->profile == nullptr)
    throw StreamingFailure("backup profile not initialized for streaming basebackup");

  /*
   * Special LABEL requested?
   */
  if (this->profile->label != "") {
    PQescapeStringConn(this->pgconn,
                       escapedlabel,
                       this->profile->label.c_str(),
                       this->profile->label.length(),
                       &escape_error);
    query << " LABEL '" << escapedlabel << "'";
  }

  /*
   * We always request PROGRESS
   */
  query << " PROGRESS ";

  /*
   * immediate CHECKPOINT requested?
   */
  if (this->profile->fast_checkpoint)
    query << " FAST ";

  /*
   * WAL stream request for base backup?
   */
  if (this->profile->include_wal)
    query << " WAL ";

  /*
   * We usually wait for WAL segments to be archived.
   */
  if (!this->profile->wait_for_wal)
    query << " NOWAIT ";

  /*
   * MAX_RATE limits the used bandwidth of the stream.
   * Check if this is requested...
   */
  if (this->profile->max_rate > 0)
    query << " MAX_RATE " << this->profile->max_rate;

  /*
   * Starting with PostgreSQL 11, we have the NOVERFIY_CHECKSUMS
   * available.
   */
  if ( (this->profile->noverify_checksums)
       && (PQserverVersion(this->pgconn) >= 110000) )
    query << " NOVERIFY_CHECKSUMS ";

  /*
   * PostgreSQL 13 introduces backup manifests, check
   * if the profile requested that feature and make sure
   * we have a server version >= 13.
   */
  if ( (this->profile->manifest)
       && (PQserverVersion(this->pgconn) >= 130000) ) {

    query << " MANIFEST 'yes' ";
    query << " MANIFEST_CHECKSUMS "
          << "'" << this->profile->manifest_checksums << "'";

  }

  /*
   * We always request the tablespace map from the stream.
   */
  query << " TABLESPACE_MAP;";

  /*
   * Fire the query...
   */
  if (PQsendQuery(this->pgconn, query.str().c_str()) == 0) {
    std::ostringstream oss;
    oss << "BASE_BACKUP command failed: " << PQerrorMessage(this->pgconn);
    throw StreamingFailure(oss.str());
  }

  result = PQgetResult(this->pgconn);

  if ((es = PQresultStatus(result)) != PGRES_TUPLES_OK) {
    std::string sqlstate(PQresultErrorField(result, PG_DIAG_SQLSTATE));
    std::ostringstream oss;

    oss << "basebackup streaming failed: " << PQresultErrorMessage(result);
    PQclear(result);
    throw StreamingExecutionFailure(oss.str(), es, sqlstate);
  }

  if (PQntuples(result) != 1) {
    std::ostringstream oss;
    oss << "unexpected result for BASE_BACKUP command, expected(rows/fields) 1/2"
        << " but got " << PQntuples(result) << "/" << PQnfields(result);
    PQclear(result);
    throw StreamingFailure(oss.str());
  }

  this->current_state = BASEBACKUP_START_POSITION;

  /*
   * First result set is the starting position of
   * the basebackup stream, with two columns:
   * 1 - XLogRecPtr
   * 2 - TimelineID
   *
   * This is also the time we record the start time and
   * other properties into our internal basebackup descriptor.
   */
  this->baseBackupDescr = std::make_shared<BaseBackupDescr>();
  this->baseBackupDescr->xlogpos = PQgetvalue(result, 0, 0);

  /* Save system identifier and WAL segment size to descriptor */

  this->baseBackupDescr->systemid         = this->systemid;
  this->baseBackupDescr->wal_segment_size = this->wal_segment_size;

  /*
   * Also the backup profile used for this basebackup.
   *
   * NOTE: This will be safed into the catalog later, which will
   *       protect the used backup profile from being deleted,
   *       as long as this basebackup exists.
   */
  this->baseBackupDescr->used_profile = this->profile->profile_id;

  /*
   * We always expect the timeline from the server here. Older
   * PostgreSQL instances than 9.3 don't send the timeline via
   * the BASE_BACKUP result set!
   */
  this->baseBackupDescr->timeline
    = CPGBackupCtlBase::strToInt(std::string(PQgetvalue(result, 0, 1)));

  if (this->profile->label != "")
    this->baseBackupDescr->label = this->profile->label;

  /*
   * And the start timestamp...
   */
  this->baseBackupDescr->started = CPGBackupCtlBase::current_timestamp();

  /*
   * Mark BaseBackupProcess as successfully started
   */
  this->current_state = BASEBACKUP_STARTED;
}

std::shared_ptr<BaseBackupDescr> BaseBackupProcess::getBaseBackupDescr() {

  /*
   * NOTE. We don't bother which state the process handle has,
   * just return the descriptor.
   */
  return this->baseBackupDescr;
}

std::string BaseBackupProcess::getSystemIdentifier() {

  if (this->baseBackupDescr != nullptr)
    return this->baseBackupDescr->systemid;

  return "";

}

void BaseBackupProcess::readTablespaceInfo() {

  PGresult *res;
  int i;
  ExecStatusType es;

  /*
   * Stack tablespace meta info only if
   * successfully started.
   */
  if (this->current_state != BASEBACKUP_STARTED)
    throw StreamingFailure("reading tablespace meta information requires a started base backup process");

  this->current_state = BASEBACKUP_TABLESPACE_META;

  /*
   * Read tablespace meta info from result set.
   */
  res = PQgetResult(this->pgconn);

  if ((es = PQresultStatus(res)) != PGRES_TUPLES_OK) {
    std::string sqlstate(PQresultErrorField(res, PG_DIAG_SQLSTATE));
    std::ostringstream oss;

    oss << "could not read tablespace meta info: "
        << PQresultErrorMessage(res);
    PQclear(res);
    throw StreamingExecutionFailure(oss.str(), es, sqlstate);
  }

  if (PQntuples(res) < 1) {
    PQclear(res);
    throw StreamingFailure("unexpected number of tuples for tablespace meta info");
  }

  if (PQnfields(res) != 3) {
    std::ostringstream oss;
    oss << "unexpected number of fields for tablespace meta info, expected 3, got "
        << PQnfields(res);
    PQclear(res);
    throw StreamingFailure(oss.str());
  }

  for (i = 0; i < PQntuples(res); i++) {

    std::shared_ptr<BackupTablespaceDescr> descr = std::make_shared<BackupTablespaceDescr>();

    /*
     * Tablespace OID
     */
    descr->spcoid = CPGBackupCtlBase::strToUInt(std::string(PQgetvalue(res, i, 0)));

    /*
     * Tablespace location (path, can be NULL in case of base directory)
     */
    if (!PQgetisnull(res, i, 1))
      descr->spclocation = std::string(PQgetvalue(res, i, 1));
    else
      descr->spclocation = "";

    /*
     * Tablespace size.
     *
     * Since we rely on PROGRESS in the BASE_BACKUP command, this is
     * assumed to be never NULL!
     */
    descr->spcsize = CPGBackupCtlBase::strToInt(std::string(PQgetvalue(res, i, 2)));

    /*
     * Push the new tablespace descriptor at the end of our internal
     * tablespace queue. Please note that the order of the tablespaces
     * retrieved via BASE_BACKUP does matter, so we use a FIFO concept here.
     */
    this->tablespaces.push(descr);
  }

  /*
   * Free result set
   */
  PQclear(res);

  /*
   * okay, looks like all meta information is there...
   */
  this->current_state = BASEBACKUP_TABLESPACE_READY;
}

void BaseBackupProcess::receiveManifest(std::shared_ptr<StreamBaseBackup> backupHandle) {

  std::shared_ptr<BackupFile> manifest_file = nullptr;
  bool interrupted = false;
  char *copybuf = NULL;
  PGresult *res;
  int rc;

  /* This is a no-op in case PostgreSQL version is lower than 13.0 */
  if (PQserverVersion(this->pgconn) < 130000) {
    BOOST_LOG_TRIVIAL(warning) << "backup manifest requested, but upstream server does not have support for it, ignoring";
    return;
  }

  if (!backupHandle->isInitialized())
    throw StreamingFailure("could not receive backup manifest");

  if (this->current_state != BASEBACKUP_EOB)
    throw StreamingFailure("unexpected state while retrieving backup manifest");

  /* Should have a valid result set with copy data */
  res = PQgetResult(pgconn);

  if (PQresultStatus(res) != PGRES_COPY_OUT) {
    std::ostringstream oss;

    oss << "could not get COPY data for backup manifest: ";
    oss << PQerrorMessage(pgconn);
    PQclear(res);
    throw StreamingFailure(oss.str());

  }

  /*
   * Make the manifest file handle.
   *
   * The contents of a manifest are plain, so we have to
   * make sure we get the right compression type and reset
   * it to the former one to not confuse our callers.
   */
  BackupProfileCompressType former_compression = backupHandle->getCompression();
  backupHandle->setCompression(BACKUP_COMPRESS_TYPE_NONE);
  manifest_file = backupHandle->stackFile("backup_manifest");
  backupHandle->setCompression(former_compression);

  while(true) {

    if (copybuf != NULL) {
      PQfreemem(copybuf);
      copybuf = NULL;
    }

    rc = PQgetCopyData(pgconn, &copybuf, 0);

    if (rc == -1) {
      /* end of copy data */
      break;
    } else if (rc == -2) {
      std::ostringstream oss;

      oss << "could not read manifest COPY data: ";
      oss << PQerrorMessage(pgconn);
      throw StreamingFailure(oss.str());

    }

    manifest_file->write(copybuf, rc);

    /*
     * Check if we are requested to stop.
     *
     * If true, we remember that we were interrupted.
     */
    if (this->stopHandlerWantsExit()) {
      interrupted = true;
      break;
    }

  }

  /*
   * Receiving the manifest is the last action during base backups,
   * mark the internal start to finish backup.
   */
  if (!interrupted)
    this->current_state = BASEBACKUP_EOB;
  else
    this->current_state = BASEBACKUP_MANIFEST_INTERRUPTED;

}

bool BaseBackupProcess::stepTablespace(std::shared_ptr<StreamBaseBackup> backupHandle,
                                       std::shared_ptr<BackupTablespaceDescr> &current_tablespace) {

  std::shared_ptr<BackupTablespaceDescr> descr = nullptr;

  if (this->tablespaces.size() < 1) {
    this->current_state = BASEBACKUP_EOB;
    current_tablespace = nullptr;
    return false;
  }

  /*
   * backupHandle must be initialized.
   */
  if (!backupHandle->isInitialized())
    throw StreamingFailure("cannot write into uninitialized streaming backup handle");

  if (this->current_state != BASEBACKUP_TABLESPACE_READY)
    throw StreamingFailure("unexpected state in base backup stream, could not start tablespace backup");

  /*
   * Check, if we are starting to step into a fresh
   * tablespace backup. This is indicated by an internal
   * state of -1 stored in the stepInfo struct property.
   *
   * In case of -1, we need to initialize the PGresult
   * handle of the current stepping.
   */
  if (this->stepInfo.current_step == -1) {

    this->stepInfo.current_step = 0;

  }
  else {

    this->stepInfo.current_step++;

  }

  /*
   * Dequeue the current tablespace from the
   * internal queue, if any.
   */
  this->stepInfo.descr = this->tablespaces.front();
  this->tablespaces.pop();
  this->stepInfo.handle = PQgetResult(this->pgconn);

  if (PQresultStatus(this->stepInfo.handle) != PGRES_COPY_OUT) {

    std::ostringstream oss;
    oss << "could not get COPY data stream: " << PQerrorMessage(this->pgconn);
    throw StreamingFailure(oss.str());
  }

  current_tablespace = descr = this->stepInfo.descr;

  /*
   * Make the file access handle.
   */
  this->stepInfo.file = backupHandle->stackFile(CPGBackupCtlBase::intToStr(descr->spcoid)
                                                + ".tar");

  if (descr->spclocation == "")
    this->current_state = BASEBACKUP_STEP_TABLESPACE_BASE;
  else
    this->current_state = BASEBACKUP_STEP_TABLESPACE;

  return true;
}

void BaseBackupProcess::end() {

  PGresult *res;
  ExecStatusType es;

  /*
   * We must have reached internal state of BASEBACKUP_EOB, otherwise
   * we cannot receive the WAL end position from the stream.
   */
  if (this->current_state != BASEBACKUP_EOB) {
    throw StreamingFailure("cannot finalize basebackup stream, not in end position");
  }

  /*
   * Get stop location.
   */
  res = PQgetResult(this->pgconn);

  if ((es = PQresultStatus(res)) != PGRES_TUPLES_OK) {
    std::string sqlstate = PQresultErrorField(res, PG_DIAG_SQLSTATE);
    std::ostringstream oss;

    oss << "could not read WAL end position from stream: "
        << PQresultErrorMessage(res);
    PQclear(res);
    throw StreamingExecutionFailure(oss.str(), es, sqlstate);
  }

  if (PQntuples(res) != 1) {
    throw StreamingFailure("unexpected number of tuples for stream end position");
  }

  if (PQnfields(res) != 2) {
    std::ostringstream oss;
    oss << "unexpected number of fields in WAL end position result set (got "
        << PQnfields(res) << " expected 2)";
    throw StreamingFailure(oss.str());
  }

  /*
   * Store the XLOG end position in the BaseBackupDescr.
   */
  this->baseBackupDescr->xlogposend = PQgetvalue(res, 0, 0);
  PQclear(res);

  /*
   * Update internal state that we have reached XLOG end position.
   */
  this->current_state = BASEBACKUP_END_POSITION;
}

void BaseBackupProcess::backupTablespace(std::shared_ptr<BackupTablespaceDescr> descr) {

  char *copybuf = NULL;
  int rc;
  bool interrupted = false;

  while(true) {

    if (copybuf != NULL) {
      PQfreemem(copybuf);
      copybuf = NULL;
    }

    rc = PQgetCopyData(this->pgconn, &copybuf, 0);

    if (rc == -1) {
      /*
       * End of COPY data stream.
       *
       * PostgreSQL sends a tar format stream as defined in
       * POSIX 1003.1-2008 standard, but omits the last two trailing
       * 512-bytes zero chunks.
       */
      char zerochunk[1024];
      memset(zerochunk, 0, sizeof(zerochunk));

      this->stepInfo.file->write(zerochunk, sizeof(zerochunk));
      PQfreemem(copybuf);
      break;
    }

    /*
     * ... else write the chunk to the file.
     *
     * NOTE: We don't need to free the copy buffer here,
     *       since we go through the next loop where PQfreemem()
     *       will do this when necessary.
     */
    this->stepInfo.file->write(copybuf, rc);

    /*
     * Check if we are requested to stop.
     *
     * If true, we remember that we were interrupted.
     */
    if (this->stopHandlerWantsExit()) {
      interrupted = true;
      break;
    }
  }

  /*
   * Mark this tablespace as ready, but only in case we werent'
   * interrupted.
   */
  if (!interrupted)
    this->current_state = BASEBACKUP_TABLESPACE_READY;
  else
    this->current_state = BASEBACKUP_STEP_TABLESPACE_INTERRUPTED;

}

BaseBackupState BaseBackupProcess::getState() {
  return this->current_state;
}

ext: cxx


text:
//
// Copyright yutopp 2015 - .
//
// Distributed under the Boost Software License, Version 1.0.
// (See accompanying file LICENSE_1_0.txt or copy at
// http://www.boost.org/LICENSE_1_0.txt)
//
#pragma once

#include <sstream>

#include <sys/types.h>
#include <sys/mount.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>
#include <errno.h>

#include <boost/filesystem.hpp>

#include "linux_user.hpp"

namespace awaho
{
    namespace fs = boost::filesystem;

    void mount(
        fs::path const& host_mount_point,
        fs::path const& guest_mount_point,
        char const* const filesystemtype,
        unsigned long const mountflags
        )
    {
        // TODO: add timeout
        std::cout << "Mounting: " << host_mount_point << " to " << guest_mount_point << std::endl;

        fs::create_directories( guest_mount_point );    // throw exception if failed

        //
        if ( ::mount(
                 host_mount_point.c_str(),
                 guest_mount_point.c_str(),
                 filesystemtype,
                 mountflags,
                 nullptr    // there is no data
                 ) == -1 ) {
            std::stringstream ss;
            ss << "Failed to mount: " << guest_mount_point
               << " errno=" << errno << " : " << std::strerror( errno );
            throw std::runtime_error( ss.str() );
        }
    }

    void mount_directory(
        fs::path const& host_mount_point,
        fs::path const& guest_mount_point,
        char const* const filesystemtype,
        unsigned long const mountflags
        )
    {
        if ( !host_mount_point.is_absolute() ) {
            std::stringstream ss;
            ss << "Failed to mount: "
               << host_mount_point << " must be ablosute path";
            throw std::runtime_error( ss.str() );
        }

        if ( !fs::is_directory( host_mount_point ) ) {
            std::stringstream ss;
            ss << "Failed to mount: "
               << host_mount_point << " is not directory";
            throw std::runtime_error( ss.str() );
        }

        //
        if ( fs::exists( guest_mount_point ) ) {
            std::stringstream ss;
            ss << "Failed to mount: GUEST "
               << guest_mount_point << " is already exists";
            throw std::runtime_error( ss.str() );
        }

        mount( host_mount_point, guest_mount_point, filesystemtype, mountflags );
    }

    // throw exception if failed
    void mount_directory_ro(
        fs::path const& host_mount_point,
        fs::path const& guest_mount_point
        )
    {
        mount_directory(
            host_mount_point,
            guest_mount_point,
            nullptr,    // MS_BIND ignores this option
            MS_BIND | MS_RDONLY | MS_NOSUID | MS_NODEV
            );
    }

    void mount_procfs( fs::path const& guest_mount_point )
    {
        mount(
            "proc",
            guest_mount_point,
            "proc",
            MS_RDONLY | MS_NOSUID | MS_NOEXEC | MS_NODEV
            );
    }

    void mount_tmpfs( fs::path const& guest_mount_point )
    {
        mount(
            "",
            guest_mount_point,
            "tmpfs",
            MS_NOEXEC | MS_NODEV
            );
    }

    bool umount_directory(
        fs::path const& guest_mount_point
        ) noexcept
    {
        std::cout << "Un mounting: " << guest_mount_point << std::endl;

        if ( ::umount2(
                 guest_mount_point.c_str(),
                 MNT_DETACH | UMOUNT_NOFOLLOW
                 ) != 0 ) {
            std::cerr << "Failed to umount: " << guest_mount_point << " errno=" << errno << " : " << std::strerror( errno ) << std::endl;
            return false;
        }

        return true;
    }

    // return number of files include in THIS directory(not recursive).
    // throws: exception
    auto number_of_files(
        fs::path const& dir
        )
        -> std::size_t
    {
        return std::distance(
            fs::directory_iterator( dir ),
            fs::directory_iterator()
            );
    }

    //
    bool remove_directory_if_empty(
        fs::path const& guest_mount_point
        ) noexcept
    {
        std::cout << "Removing: " << guest_mount_point << std::endl;
        if ( !fs::exists( guest_mount_point ) ) {
            return false;
        }

        try {
            auto const count = number_of_files( guest_mount_point );
            if ( count != 0 ) {
                std::cerr << "Failed to remove: " << guest_mount_point << " There are some files (num = " << count << ") " << std::endl;

                return false;
            }

            fs::remove_all( guest_mount_point );

            return true;

        } catch(...) {
            // TODO: error handling
            return false;
        }
    }

    bool cleanup_directory(
        fs::path const& guest_mount_point
        ) noexcept
    {
        auto const f = umount_directory( guest_mount_point );
        auto const s = remove_directory_if_empty( guest_mount_point );

        return f && s;
    }

    void change_file_owner(
        fs::path const& guest_path,
        linux::user const& user
        )
    {
        std::cout << "Chown: " << guest_path << std::endl;

        if ( ::chown( guest_path.c_str(), user.user_id(), user.group_id() ) != 0 ) {
            std::stringstream ss;
            ss << "Failed to chown."
               << " errno=" << errno << " : " << std::strerror( errno );
            throw std::runtime_error( ss.str() );
        }
    }

    void change_directory_owner_rec(
        fs::path const& guest_dir,
        linux::user const& user
        )
    {
        if ( !fs::is_directory( guest_dir ) || fs::is_symlink( guest_dir ) ) {
            return change_file_owner( guest_dir, user );
        }

        auto const begin = fs::directory_iterator( guest_dir );
        auto const end = fs::directory_iterator();

        change_file_owner( guest_dir, user );

        for( auto&& it : boost::make_iterator_range( begin, end ) ) {
            auto const& p = it.path();

            if ( fs::is_directory( p ) && !fs::is_symlink( p ) ) {
                change_directory_owner_rec( p, user );
            } else {
                change_file_owner( p, user );
            }
        }
    }

    bool remove_node(
        fs::path const& guest_node_path
        ) noexcept
    {
        try {
            std::cout << "Removing node: " << guest_node_path << std::endl;

            fs::remove( guest_node_path );

        } catch( fs::filesystem_error const& e ) {
            std::cerr << "Failed to remove_node: "
                      << e.what() << std::endl;
            return false;

        } catch(...) {
            std::cerr << "Failed to remove_node: "
                      << "unexpected" << std::endl;
            return false;
        }

        return true;
    }

    bool remove_file(
        fs::path const& guest_file_path
        ) noexcept
    {
        try {
            std::cout << "Removing file: " << guest_file_path << std::endl;

            fs::remove( guest_file_path );

        } catch( fs::filesystem_error const& e ) {
            std::cerr << "Failed to remove_file: " << guest_file_path
                      << " / reason: " << e.what() << std::endl;
            return false;

        } catch(...) {
            std::cerr << "Failed to remove_file: " << guest_file_path
                      << " / unexpected" << std::endl;
            return false;
        }

        return true;
    }

    void make_node(
        fs::path const& guest_node_path,
        dev_t const& dev,
        mode_t const& perm
        )
    {
        std::cout << "Creating node: " << guest_node_path << std::endl;

        if ( fs::exists( guest_node_path ) ) {
            if ( !remove_node( guest_node_path ) ) {
                std::stringstream ss;
                ss << "Failed to make_node: "
                   << " " << guest_node_path << " is already exists and couldn't remove.";
                throw std::runtime_error( ss.str() );
            }
        }

        if ( ::mknod( guest_node_path.c_str(), S_IFCHR, dev ) != 0 ) {
            std::stringstream ss;
            ss << "Failed to mknod: " << guest_node_path
               << " errno=" << errno << " : " << std::strerror( errno );
            throw std::runtime_error( ss.str() );
        }

        if ( ::chmod( guest_node_path.c_str(), perm ) != 0 ) {
            std::stringstream ss;
            ss << "Failed to chmod: " << guest_node_path
               << " errno=" << errno << " : " << std::strerror( errno );
            throw std::runtime_error( ss.str() );
        }
    }

    bool remove_symlink(
        fs::path const& dest
        ) noexcept
    {
        if ( ::unlink( dest.c_str() ) != 0 ) {
            std::cerr << "Failed to unlink: " << dest
                      << " errno=" << errno << " : " << std::strerror( errno ) << std::endl;
            return false;
        }

        return true;
    }

    void make_symlink(
        fs::path const& src,
        fs::path const& dest
        )
    {
        if ( fs::exists( dest ) ) {
            if ( !remove_symlink( dest ) ) {
                std::stringstream ss;
                ss << "Failed to make_symlink: "
                   << " " << dest << " is already exists and couldn't remove.";
                throw std::runtime_error( ss.str() );
            }
        }

        if ( ::symlink( src.c_str(), dest.c_str() ) != 0 ) {
            std::stringstream ss;
            ss << "Failed to symlink: " << dest << " -> " << src
               << " errno=" << errno << " : " << std::strerror( errno );
            throw std::runtime_error( ss.str() );
        }
    }

    void copy_user_files(
        fs::path const& host_copy_point,
        fs::path const& guest_copy_point
        )
    {
        // TODO: add timeout
        std::cout << "Cp(userfile): " << host_copy_point << " to " << guest_copy_point
                  << " / is dir: " << fs::is_directory( host_copy_point )
                  << " / is sym: " << fs::is_symlink( host_copy_point ) << std::endl;

        //
        if ( fs::exists( guest_copy_point ) ) {
            std::stringstream ss;
            ss << "Failed to copy: GUEST "
               << guest_copy_point << " is already exists";
            throw std::runtime_error( ss.str() );
        }

        bool const do_rec =
            fs::is_directory( host_copy_point ) &&
            !fs::is_symlink( host_copy_point );

        if ( !fs::exists( guest_copy_point.parent_path() ) ) {
            fs::create_directory( guest_copy_point.parent_path() );
        }

        fs::copy( host_copy_point, guest_copy_point );

        if ( do_rec ) {
            // copy recuesive
            auto const begin = fs::directory_iterator( host_copy_point );
            auto const end = fs::directory_iterator();

            for( auto&& it : boost::make_iterator_range( begin, end ) ) {
                auto const& p = it.path();
                auto const& filename = p.filename();

                copy_user_files( host_copy_point / filename, guest_copy_point / filename );
            }
        }
    }

    // ...?
    bool remove_user_files(
        fs::path const& guest_copy_point
        )
    try {
        std::cout << "Rm(userfile): " << guest_copy_point << std::endl;

        //
        if ( fs::is_symlink( guest_copy_point ) ||
             fs::is_regular_file( guest_copy_point )
            ) {
            fs::remove( guest_copy_point );

            return true;

        } else if ( fs::is_directory( guest_copy_point ) ) {
            // remove recuesive
            auto const begin = fs::directory_iterator( guest_copy_point );
            auto const end = fs::directory_iterator();

            bool succeeded = true;
            for( auto&& it : boost::make_iterator_range( begin, end ) ) {
                auto const& p = it.path();

                succeeded &= remove_user_files( p );
            }

            succeeded &= remove_directory_if_empty( guest_copy_point );

            return succeeded;

        } else {
            return false;
        }

    } catch( fs::filesystem_error const& e ) {
        std::cerr << "Failed to remove_node: "
                  << e.what() << std::endl;
        return false;

    } catch(...) {
        std::cerr << "Failed to remove_node: "
                  << "unexpected" << std::endl;
        return false;
    }

} // namespace awaho

max_stars_repo_name: yutopp/awaho

max_issues_repo_name: yutopp/awaho

max_forks_repo_name: yutopp/awaho


text:
#pragma once

#include <atomic>
#include <iostream>
#include <cassert>
#include <cstring>
#include <deque>
#include <map>
#include <memory>
#include <mutex>
#include <optional>
#include <sstream>
#include <thread>
#include <unordered_map>
#include <boost/lexical_cast.hpp>
#include <sys/epoll.h>
#include <mysql.h>

#include <li/callable_traits/callable_traits.hh>
#include <li/metamap/metamap.hh>
#include <li/sql/sql_common.hh>
#include <li/sql/symbols.hh>
#include <li/sql/mysql_async_wrapper.hh>
#include <li/sql/mysql_statement.hh>
#include <li/sql/type_hashmap.hh>

namespace li {

int max_mysql_connections_per_thread = 200;

struct mysql_tag {};

struct mysql_database;

struct mysql_connection_data {

  ~mysql_connection_data() {
    mysql_close(connection);
  }

  MYSQL* connection;
  std::unordered_map<std::string, std::shared_ptr<mysql_statement_data>> statements;
  type_hashmap<std::shared_ptr<mysql_statement_data>> statements_hashmap;
};

thread_local std::deque<std::shared_ptr<mysql_connection_data>> mysql_connection_pool;
thread_local std::deque<std::shared_ptr<mysql_connection_data>> mysql_connection_async_pool;
thread_local int total_number_of_mysql_connections = 0;

template <typename B> // must be mysql_functions_blocking or mysql_functions_non_blocking
struct mysql_result {

  B& mysql_wrapper_;
  MYSQL* con_;
  std::shared_ptr<int> connection_status_;
  MYSQL_RES* result_ = nullptr;

  template <typename T>
  T read() {
    result_ = mysql_use_result(con_);
    MYSQL_ROW row = mysql_wrapper_.mysql_fetch_row(connection_status_, result_);
    return boost::lexical_cast<T>(row[0]);
    mysql_wrapper_.mysql_free_result(connection_status_, result_);
  }

  void wait() {}

};

template <typename B> // must be mysql_functions_blocking or mysql_functions_non_blocking
struct mysql_connection {

  typedef mysql_tag db_tag;

  inline mysql_connection(B mysql_wrapper, std::shared_ptr<li::mysql_connection_data> data)
      : mysql_wrapper_(mysql_wrapper), data_(data), stm_cache_(data->statements),
        con_(data->connection){

    connection_status_ = std::shared_ptr<int>(new int(0), [=](int* p) mutable {

      if (*p or (mysql_connection_pool.size() + mysql_connection_async_pool.size())  > max_mysql_connections_per_thread)
      {
      //  if constexpr (!B::is_blocking)
      //    mysql_wrapper.yield_.unsubscribe(mysql_get_socket(data->connection));
       total_number_of_mysql_connections--;
        //std::cerr << "Discarding broken mysql connection." << std::endl;
        return;
      }

      if constexpr (B::is_blocking)
        mysql_connection_pool.push_back(data);
      else
        mysql_connection_async_pool.push_back(data);

    });
  }

  long long int last_insert_rowid() { return mysql_insert_id(con_); }

  mysql_result<B> operator()(const std::string& rq) {
    mysql_wrapper_.mysql_real_query(connection_status_, con_, rq.c_str(), rq.size());
    return mysql_result<B>{mysql_wrapper_, con_, connection_status_};
  }

  template <typename F, typename... K>
  mysql_statement<B> cached_statement(F f, K... keys) {
    if (data_->statements_hashmap(f).get() == nullptr)
    {
      mysql_statement<B> res = prepare(f());
      data_->statements_hashmap(f, keys...) = res.data_.shared_from_this();
      return res;
    }
    else
      return mysql_statement<B>{mysql_wrapper_,
                                *data_->statements_hashmap(f, keys...),
                                 connection_status_};
  }

  mysql_statement<B> prepare(const std::string& rq) {
    auto it = stm_cache_.find(rq);
    if (it != stm_cache_.end())
    {
      //mysql_wrapper_.mysql_stmt_free_result(it->second->stmt_);
      //mysql_wrapper_.mysql_stmt_reset(it->second->stmt_);
      return mysql_statement<B>{mysql_wrapper_, *it->second, connection_status_};
    }
    //std::cout << "prepare " << rq << std::endl;
    MYSQL_STMT* stmt = mysql_stmt_init(con_);
    if (!stmt)
    {
      *connection_status_ = true;
      throw std::runtime_error(std::string("mysql_stmt_init error: ") + mysql_error(con_));
    }
    if (mysql_wrapper_.mysql_stmt_prepare(connection_status_, stmt, rq.data(), rq.size()))
    {
      *connection_status_ = true;
      throw std::runtime_error(std::string("mysql_stmt_prepare error: ") + mysql_error(con_));
    }

    auto pair = stm_cache_.emplace(rq, std::make_shared<mysql_statement_data>(stmt));
    return mysql_statement<B>{mysql_wrapper_, *pair.first->second, connection_status_};
  }

  template <typename T>
  inline std::string type_to_string(const T&, std::enable_if_t<std::is_integral<T>::value>* = 0) {
    return "INT";
  }
  template <typename T>
  inline std::string type_to_string(const T&,
                                    std::enable_if_t<std::is_floating_point<T>::value>* = 0) {
    return "DOUBLE";
  }
  inline std::string type_to_string(const std::string&) { return "MEDIUMTEXT"; }
  inline std::string type_to_string(const sql_blob&) { return "BLOB"; }
  template <unsigned S> inline std::string type_to_string(const sql_varchar<S>) {
    std::ostringstream ss;
    ss << "VARCHAR(" << S << ')';
    return ss.str();
  }

  B mysql_wrapper_;
  std::shared_ptr<mysql_connection_data> data_;
  std::unordered_map<std::string, std::shared_ptr<mysql_statement_data>>& stm_cache_;
  MYSQL* con_;
  std::shared_ptr<int> connection_status_;
};

struct mysql_database : std::enable_shared_from_this<mysql_database> {

  template <typename... O> inline mysql_database(O... opts) {

    auto options = mmm(opts...);
    static_assert(has_key(options, s::host), "open_mysql_connection requires the s::host argument");
    static_assert(has_key(options, s::database),
                  "open_mysql_connection requires the s::databaser argument");
    static_assert(has_key(options, s::user), "open_mysql_connection requires the s::user argument");
    static_assert(has_key(options, s::password),
                  "open_mysql_connection requires the s::password argument");

    host_ = options.host;
    database_ = options.database;
    user_ = options.user;
    passwd_ = options.password;
    port_ = get_or(options, s::port, 0);

    character_set_ = get_or(options, s::charset, "utf8");

    if (mysql_library_init(0, NULL, NULL))
      throw std::runtime_error("Could not initialize MySQL library.");
    if (!mysql_thread_safe())
      throw std::runtime_error("Mysql is not compiled as thread safe.");
  }

  ~mysql_database() { mysql_library_end(); }

  template <typename Y>
  inline mysql_connection<mysql_functions_non_blocking<Y>> connect(Y& fiber) {

    //std::cout << "nconnection " << total_number_of_mysql_connections << std::endl;
    int ntry = 0;
    std::shared_ptr<mysql_connection_data> data = nullptr;
    while (!data)
    {
      // if (ntry > 20)
      //   throw std::runtime_error("Cannot connect to the database");
      ntry++;

      if (!mysql_connection_async_pool.empty()) {
        data = mysql_connection_async_pool.back();
        mysql_connection_async_pool.pop_back();
        fiber.epoll_add(mysql_get_socket(data->connection), EPOLLIN | EPOLLOUT | EPOLLRDHUP | EPOLLET);
      }
      else
      {
        // std::cout << total_number_of_mysql_connections << " connections. "<< std::endl;
        if (total_number_of_mysql_connections > max_mysql_connections_per_thread)
        {
          //std::cout << "Waiting for a free mysql connection..." << std::endl;
          fiber.yield();
          continue;
        }
        total_number_of_mysql_connections++;
        //std::cout << "NEW MYSQL CONNECTION "  << std::endl;
        MYSQL* mysql;
        int mysql_fd = -1;
        int status;
        MYSQL* connection;
        //while (mysql_fd == -1)
        {
          mysql = new MYSQL;
          mysql_init(mysql);
          mysql_options(mysql, MYSQL_OPT_NONBLOCK, 0);
          connection = nullptr;
          status = mysql_real_connect_start(&connection, mysql, host_.c_str(), user_.c_str(), passwd_.c_str(),
                                            database_.c_str(), port_, NULL, 0);

          //std::cout << "after: " << mysql_get_socket(mysql) << " " << status == MYSQL_ << std::endl;
          mysql_fd = mysql_get_socket(mysql);
          if (mysql_fd == -1)
          {
             //std::cout << "Invalid mysql connection bad mysql_get_socket " << status << " " << mysql << std::endl;
            mysql_close(mysql);
            total_number_of_mysql_connections--;
            //usleep(1e6);
            fiber.yield();
            continue;
          }
        }
        if (status)
          fiber.epoll_add(mysql_fd, EPOLLIN | EPOLLOUT | EPOLLRDHUP | EPOLLET);
        while (status) try {
          fiber.yield();
          status = mysql_real_connect_cont(&connection, mysql, status);
        }  catch (typename Y::exception_type& e) {
          // Yield thrown a exception (probably because a closed connection).
          // std::cerr << "Warning: yield threw an exception while connecting to mysql: "
          //  << total_number_of_mysql_connections << std::endl;
          total_number_of_mysql_connections--;
          mysql_close(mysql);
          throw std::move(e);
        }
        if (!connection)
        {
          //std::cout << "Error in mysql_real_connect_cont" << std::endl;
          total_number_of_mysql_connections--;
          fiber.yield();
          continue;
        }
          //throw std::runtime_error("Cannot connect to the database");
        mysql_set_character_set(mysql, character_set_.c_str());
        data = std::shared_ptr<mysql_connection_data>(new mysql_connection_data{mysql});
      }
    }
    assert(data);
    return std::move(mysql_connection(mysql_functions_non_blocking<Y>{fiber}, data));
  }

  inline mysql_connection<mysql_functions_blocking> connect() {
    std::shared_ptr<mysql_connection_data> data = nullptr;
    if (!mysql_connection_pool.empty()) {
      data = mysql_connection_pool.back();
      mysql_connection_pool.pop_back();
    }

    if (!data) {
      MYSQL* con_ = mysql_init(nullptr);
      con_ = mysql_real_connect(con_, host_.c_str(), user_.c_str(), passwd_.c_str(),
                                database_.c_str(), port_, NULL, 0);
      if (!con_)
        throw std::runtime_error("Cannot connect to the database");

      //total_number_of_mysql_connections++;
      mysql_set_character_set(con_, character_set_.c_str());
      data = std::shared_ptr<mysql_connection_data>(new mysql_connection_data{con_});
    }

    assert(data);
    return std::move(mysql_connection(mysql_functions_blocking{}, data));
  }

  std::mutex mutex_;
  std::string host_, user_, passwd_, database_;
  unsigned int port_;
  std::deque<MYSQL*> _;
  std::string character_set_;
};

} // namespace li

ext: hh

max_stars_repo_name: c-cesar/lithium

max_issues_repo_name: c-cesar/lithium

max_forks_repo_name: c-cesar/lithium