Merge pull request #104 from whwjiang/main

Support streaming writes against HTTP backends

Author: bbockelm

.gitignore

@@ -1 +1,2 @@
 build/
+.vscode/

src/HTTPCommands.cc

@@ -1,6 +1,6 @@
 /***************************************************************
  *
- * Copyright (C) 2024, Pelican Project, Morgridge Institute for Research
+ * Copyright (C) 2025, Pelican Project, Morgridge Institute for Research
  *
  * Licensed under the Apache License, Version 2.0 (the "License"); you
  * may not use this file except in compliance with the License.  You may
@@ -277,6 +277,15 @@ size_t HTTPRequest::ReadCallback(char *buffer, size_t size, size_t n, void *v) {
 		return CURL_READFUNC_ABORT;
 	}
 
+	if (payload->m_parent.m_log.getMsgMask() & LogMask::Dump) {
+		payload->m_parent.m_log.Log(
+			LogMask::Dump, "ReadCallback",
+			("sentSoFar=" + std::to_string(payload->sentSoFar) +
+			 " data.size=" + std::to_string(payload->data.size()) +
+			 " final=" + std::to_string(payload->final))
+				.c_str());
+	}
+
 	if (payload->sentSoFar == static_cast<off_t>(payload->data.size())) {
 		payload->sentSoFar = 0;
 		if (payload->final) {
@@ -778,19 +787,30 @@ void HTTPRequest::ProcessCurlResult(CURL *curl, CURLcode rv) {
 
 HTTPUpload::~HTTPUpload() {}
 
-bool HTTPUpload::SendRequest(const std::string &payload, off_t offset,
-							 size_t size) {
-	if (offset != 0 || size != 0) {
-		std::string range;
-		formatstr(range, "bytes=%lld-%lld", static_cast<long long int>(offset),
-				  static_cast<long long int>(offset + size - 1));
-		headers["Range"] = range.c_str();
-	}
-
+bool HTTPUpload::SendRequest(const std::string &payload) {
 	httpVerb = "PUT";
+	expectedResponseCode = 201;
 	return SendHTTPRequest(payload);
 }
 
+bool HTTPUpload::StartStreamingRequest(const std::string_view payload,
+									   off_t object_size) {
+	httpVerb = "PUT";
+	expectedResponseCode = 201;
+	headers["Content-Type"] = "binary/octet-stream";
+	return sendPreparedRequest(hostUrl, payload, object_size, false);
+}
+
+bool HTTPUpload::ContinueStreamingRequest(const std::string_view payload,
+										  off_t object_size, bool final) {
+	// Note that despite the fact that final gets passed through here,
+	// in reality the way that curl determines whether the data transfer is
+	// done is by seeing if the total amount of data sent is equal to the
+	// expected size of the entire payload, stored in m_object_size.
+	// See HTTPRequest::ReadCallback for more info
+	return sendPreparedRequest(hostUrl, payload, object_size, final);
+}
+
 void HTTPRequest::Init(XrdSysError &log) {
 	if (!m_workers_initialized) {
 		for (unsigned idx = 0; idx < CurlWorker::GetPollThreads(); idx++) {
@@ -837,3 +857,46 @@ bool HTTPHead::SendRequest() {
 }
 
 // ---------------------------------------------------------------------------
+
+int HTTPRequest::HandleHTTPError(const HTTPRequest &request, XrdSysError &log,
+								 const char *operation, const char *context) {
+	auto httpCode = request.getResponseCode();
+	if (httpCode) {
+		std::stringstream ss;
+		ss << operation << " failed: " << request.getResponseCode() << ": "
+		   << request.getResultString();
+		if (context) {
+			ss << " (context: " << context << ")";
+		}
+		log.Log(LogMask::Warning, "HTTPRequest::HandleHTTPError",
+				ss.str().c_str());
+
+		switch (httpCode) {
+		case 404:
+			return -ENOENT;
+		case 500:
+			return -EIO;
+		case 403:
+			return -EPERM;
+		case 401:
+			return -EACCES;
+		case 400:
+			return -EINVAL;
+		case 503:
+			return -EAGAIN;
+		default:
+			return -EIO;
+		}
+	} else {
+		std::stringstream ss;
+		ss << "Failed to send " << operation
+		   << " command: " << request.getErrorCode() << ": "
+		   << request.getErrorMessage();
+		if (context) {
+			ss << " (context: " << context << ")";
+		}
+		log.Log(LogMask::Warning, "HTTPRequest::HandleHTTPError",
+				ss.str().c_str());
+		return -EIO;
+	}
+}

src/HTTPCommands.hh

@@ -1,6 +1,6 @@
 /***************************************************************
  *
- * Copyright (C) 2024, Pelican Project, Morgridge Institute for Research
+ * Copyright (C) 2025, Pelican Project, Morgridge Institute for Research
  *
  * Licensed under the Apache License, Version 2.0 (the "License"); you
  * may not use this file except in compliance with the License.  You may
@@ -95,6 +95,19 @@ class HTTPRequest {
 		return m_timeout_duration;
 	}
 
+	// Handle HTTP request errors and convert them to appropriate POSIX error codes.
+	// This function can be used anywhere HTTP requests are made to provide
+	// consistent error handling.
+	//
+	// - request: The HTTPRequest object that was used for the request
+	// - log: The logger instance for error reporting
+	// - operation: A string describing the operation being performed (for logging)
+	// - context: Additional context information (for logging)
+	//
+	// Returns: A POSIX error code (-ENOENT, -EIO, -EPERM, etc.) or 0 if no error
+	static int HandleHTTPError(const HTTPRequest &request, XrdSysError &log,
+							   const char *operation, const char *context = nullptr);
+
   protected:
 	// Send the request to the HTTP server.
 	// Blocks until the request has completed.
@@ -295,8 +308,23 @@ class HTTPUpload : public HTTPRequest {
 
 	virtual ~HTTPUpload();
 
-	virtual bool SendRequest(const std::string &payload, off_t offset,
-							 size_t size);
+	virtual bool SendRequest(const std::string &payload);
+
+	// Start a streaming request.
+	//
+	// - payload: The payload contents when uploading.
+	// - object_size: Size of the entire upload payload.
+	bool StartStreamingRequest(const std::string_view payload,
+							   off_t object_size);
+
+	// Continue a streaming request.
+	//
+	// - payload: The payload contents when uploading.
+	// - object_size: Size of the entire upload payload.
+	// - final: True if this is the last or only payload for the request.  False
+	// otherwise.
+	bool ContinueStreamingRequest(const std::string_view payload,
+								  off_t object_size, bool final);
 
   protected:
 	std::string object;