XRootD
XrdClHttpOpCopy.cc
Go to the documentation of this file.
1 /******************************************************************************/
2 /* Copyright (C) 2025, Pelican Project, Morgridge Institute for Research */
3 /* */
4 /* This file is part of the XrdClHttp client plugin for XRootD. */
5 /* */
6 /* XRootD is free software: you can redistribute it and/or modify it under */
7 /* the terms of the GNU Lesser General Public License as published by the */
8 /* Free Software Foundation, either version 3 of the License, or (at your */
9 /* option) any later version. */
10 /* */
11 /* XRootD is distributed in the hope that it will be useful, but WITHOUT */
12 /* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or */
13 /* FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public */
14 /* License for more details. */
15 /* */
16 /* The copyright holder's institutional names and contributor's names may not */
17 /* be used to endorse or promote products derived from this software without */
18 /* specific prior written permission of the institution or contributor. */
19 /******************************************************************************/
20 
21 #include "XrdClHttpOps.hh"
22 
23 #include <string>
24 
25 using namespace XrdClHttp;
26 using namespace std::string_literals;
27 
28 CurlCopyOp::CurlCopyOp(XrdCl::ResponseHandler *handler, const std::string &source_url, const Headers &source_hdrs,
29  const std::string &dest_url, const Headers &dest_hdrs, const Headers &connection_hdrs, TpcMode mode, struct timespec timeout, XrdCl::Log *logger,
30  CreateConnCalloutType callout) :
31  CurlOperation(handler, mode == TpcMode::Pull ? dest_url : source_url, timeout, logger, callout, nullptr)
32  {
33  m_minimum_rate = 1;
34 
35  // The headers of the endpoint the client contacts go on the request
36  // itself. The headers of the remote endpoint are forwarded by that
37  // endpoint, thus they need the 'TransferHeader' prefix.
38  const Headers &regular_hdrs = mode == TpcMode::Pull ? dest_hdrs : source_hdrs;
39  const Headers &transfer_hdrs = mode == TpcMode::Pull ? source_hdrs : dest_hdrs;
40 
41  if (mode == TpcMode::Pull)
42  m_headers_list.emplace_back("Source"s, source_url);
43  else
44  m_headers_list.emplace_back("Destination"s, dest_url);
45 
46  std::copy(connection_hdrs.begin(), connection_hdrs.end(), std::back_inserter(m_headers_list));
47  std::copy(regular_hdrs.begin(), regular_hdrs.end(), std::back_inserter(m_headers_list));
48 
49  for (const auto &info : transfer_hdrs) {
50  m_headers_list.emplace_back("TransferHeader"s + info.first, info.second);
51  }
52  }
53 
54  bool
56  {
57  auto rv = CurlOperation::Setup(curl, worker);
58  if (!rv) return false;
59 
60  curl_easy_setopt(m_curl.get(), CURLOPT_WRITEFUNCTION, CurlCopyOp::WriteCallback);
61  curl_easy_setopt(m_curl.get(), CURLOPT_WRITEDATA, this);
62  curl_easy_setopt(m_curl.get(), CURLOPT_CUSTOMREQUEST, "COPY");
63 
64  return true;
65  }
66 
67  void
69  {
70  SetDone(false);
71  if (m_handler == nullptr) {return;}
72  auto status = new XrdCl::XRootDStatus();
73  auto obj = new XrdCl::AnyObject();
74  auto handle = m_handler;
75  m_handler = nullptr;
76  handle->HandleResponse(status, obj);
77  }
78 
79  void
81  {
82  if (m_curl == nullptr) return;
83  curl_easy_setopt(m_curl.get(), CURLOPT_WRITEFUNCTION, nullptr);
84  curl_easy_setopt(m_curl.get(), CURLOPT_WRITEDATA, nullptr);
85  curl_easy_setopt(m_curl.get(), CURLOPT_CUSTOMREQUEST, nullptr);
86  curl_easy_setopt(m_curl.get(), CURLOPT_HTTPHEADER, nullptr);
87  curl_easy_setopt(m_curl.get(), CURLOPT_XFERINFOFUNCTION, nullptr);
89  }
90 
91  void
93  {
94  m_progress_handler = handler;
95  }
96 
97  size_t
98  CurlCopyOp::WriteCallback(char *buffer, size_t size, size_t nitems, void *this_ptr)
99  {
100  auto me = reinterpret_cast<CurlCopyOp*>(this_ptr);
101  me->UpdateBytes(size * nitems);
102  std::string_view str_data(buffer, size * nitems);
103  size_t end_line;
104  while ((end_line = std::min(str_data.size(), str_data.find('\n'))) > 0) {
105 
106  auto cur_line = str_data.substr(0, end_line);
107 
108  if (me->m_line_buffer.empty()) {
109  me->HandleLine(cur_line);
110  } else {
111  me->m_line_buffer += cur_line;
112  me->HandleLine(me->m_line_buffer);
113  me->m_line_buffer.clear();
114  }
115 
116  if (end_line == str_data.size())
117  break;
118 
119  str_data = str_data.substr(end_line + 1);
120  }
121  me->m_line_buffer = str_data;
122 
123  return size * nitems;
124  }
125 
126  void
127  CurlCopyOp::HandleLine(std::string_view line)
128  {
129  if (line == "Perf Marker") {
130  m_bytemark = -1;
131  } else if (line == "End") {
132  if (m_bytemark > -1 && m_progress_handler) {
133  m_progress_handler->HandleProgress(static_cast<std::size_t>(m_bytemark));
134  }
135  } else {
136  auto key_end_pos = line.find(':');
137  if (key_end_pos == line.npos) {
138  return; // All the other callback lines should be of key: value format
139  }
140  auto key = line.substr(0, key_end_pos);
141  auto value = ltrim_view(line.substr(key_end_pos + 1));
142  if (key == "Stripe Bytes Transferred") {
143  try {
144  m_bytemark = std::stoll(std::string(value));
145  } catch (...) {
146  // TODO: Log failure
147  }
148  } else if (key == "success") {
149  m_sent_success = true;
150  } else if (key == "failure") {
151  m_failure = value;
152  }
153  }
154  }
155 
void CURL
void Success() override
void ReleaseHandle() override
std::vector< std::pair< std::string, std::string > > Headers
void SetProgressHandler(XrdCl::ProgressHandler *handler) noexcept
bool Setup(CURL *curl, CurlWorker &) override
CurlCopyOp(XrdCl::ResponseHandler *handler, const std::string &source_url, const Headers &source_hdrs, const std::string &dest_url, const Headers &dest_hdrs, const Headers &connection_hdrs, TpcMode mode, struct timespec timeout, XrdCl::Log *logger, CreateConnCalloutType callout)
void SetDone(bool has_failed)
std::unique_ptr< CURL, void(*)(CURL *)> m_curl
virtual void ReleaseHandle()
void UpdateBytes(uint64_t bytes)
std::vector< std::pair< std::string, std::string > > m_headers_list
XrdCl::ResponseHandler * m_handler
virtual bool Setup(CURL *curl, CurlWorker &)
Handle diagnostics.
Definition: XrdClLog.hh:101
Handle the progress of an asynchronous operation.
virtual void HandleProgress(std::size_t processed, std::size_t total=0)=0
Handle an async response.
virtual void HandleResponse(XRootDStatus *status, AnyObject *response)
std::string_view ltrim_view(const std::string_view &input_view)
ConnectionCallout *(*)(const std::string &, const ResponseInfo &) CreateConnCalloutType