Hawkin
/
arangodb
mirror of https://gitee.com/bigwinds/arangodb
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Multihreaded import

This commit is contained in:
Simon Grätzer 2017-05-24 18:37:45 +02:00
parent b2cd86ad8b
commit 2f2d07ab9a
22 changed files with 866 additions and 460 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
arangod/Agency/AgencyComm.cpp
View File

@ -1555,10 +1555,11 @@ AgencyCommResult AgencyComm::send(
<< connection->getEndpoint()->specification() << "', url '" << url
<< "': " << body;
arangodb::httpclient::SimpleHttpClient client(connection, timeout, false);
client.setJwt(ClusterComm::instance()->jwt());
client.keepConnectionOnDestruction(true);
arangodb::httpclient::SimpleHttpClientParams params(timeout, false);
params.setJwt(ClusterComm::instance()->jwt());
params.keepConnectionOnDestruction(true);
arangodb::httpclient::SimpleHttpClient client(connection, params);
// set up headers
std::unordered_map<std::string, std::string> headers;

1
arangod/Cluster/ClusterComm.cpp
View File

@ -33,7 +33,6 @@
#include "Scheduler/JobGuard.h"
#include "Scheduler/SchedulerFeature.h"
#include "SimpleHttpClient/ConnectionManager.h"
#include "SimpleHttpClient/SimpleHttpClient.h"
#include "SimpleHttpClient/SimpleHttpCommunicatorResult.h"
#include "Transaction/Methods.h"
#include "VocBase/ticks.h"

40
arangod/Replication/Syncer.cpp
View File

@ -95,26 +95,24 @@ Syncer::Syncer(TRI_vocbase_t* vocbase,
(uint32_t)_configuration._sslProtocol);
if (_connection != nullptr) {
_client = new SimpleHttpClient(_connection,
_configuration._requestTimeout, false);
SimpleHttpClientParams params(_configuration._requestTimeout, false);
params.setMaxRetries(2);
params.setRetryWaitTime(2 * 1000 * 1000);
params.setRetryMessage(std::string("retrying failed HTTP request for endpoint '") +
_configuration._endpoint +
std::string("' for replication applier in database '" +
_vocbase->name() + "'"));
std::string username = _configuration._username;
std::string password = _configuration._password;
if (!username.empty()) {
_client->setUserNamePassword("/", username, password);
params.setUserNamePassword("/", username, password);
} else {
_client->setJwt(_configuration._jwt);
params.setJwt(_configuration._jwt);
}
_client->setLocationRewriter(this, &rewriteLocation);
_client->_maxRetries = 2;
_client->_retryWaitTime = 2 * 1000 * 1000;
_client->_retryMessage =
std::string("retrying failed HTTP request for endpoint '") +
_configuration._endpoint +
std::string("' for replication applier in database '" +
_vocbase->name() + "'");
params.setLocationRewriter(this, &rewriteLocation);
_client = new SimpleHttpClient(_connection, params);
}
}
}
@ -634,19 +632,19 @@ int Syncer::getMasterState(std::string& errorMsg) {
BaseUrl + "/logger-state?serverId=" + _localServerIdString;
// store old settings
uint64_t maxRetries = _client->_maxRetries;
uint64_t retryWaitTime = _client->_retryWaitTime;
size_t maxRetries = _client->params().getMaxRetries();
uint64_t retryWaitTime = _client->params().getRetryWaitTime();
// apply settings that prevent endless waiting here
_client->_maxRetries = 1;
_client->_retryWaitTime = 500 * 1000;
_client->params().setMaxRetries(1);
_client->params().setRetryWaitTime(500 * 1000);
std::unique_ptr<SimpleHttpResult> response(
_client->retryRequest(rest::RequestType::GET, url, nullptr, 0));
// restore old settings
_client->_maxRetries = static_cast<size_t>(maxRetries);
_client->_retryWaitTime = retryWaitTime;
_client->params().setMaxRetries(maxRetries);
_client->params().setRetryWaitTime(retryWaitTime);
if (response == nullptr || !response->isComplete()) {
errorMsg = "could not connect to master at " + _masterInfo._endpoint +

1
arangod/RocksDBEngine/RocksDBReplicationTailing.cpp
View File

@ -380,6 +380,7 @@ class WALParser : public rocksdb::WriteBatch::Handler {
_builder.add("tid", VPackValue(std::to_string(_currentTrxId)));
_builder.close();
}
_lastLogType = RocksDBLogType::Invalid;
_seenBeginTransaction = false;
_singleOp = false;
_startOfBatch = true;

6
arangosh/Benchmark/BenchmarkThread.h
View File

@ -90,10 +90,10 @@ class BenchmarkThread : public arangodb::Thread {
FATAL_ERROR_EXIT();
}
_httpClient->setLocationRewriter(this, &rewriteLocation);
_httpClient->params().setLocationRewriter(this, &rewriteLocation);
_httpClient->setUserNamePassword("/", _username, _password);
_httpClient->setKeepAlive(_keepAlive);
_httpClient->params().setUserNamePassword("/", _username, _password);
_httpClient->params().setKeepAlive(_keepAlive);
// test the connection
httpclient::SimpleHttpResult* result =

2
arangosh/CMakeLists.txt
View File

@ -162,6 +162,7 @@ add_executable(${BIN_ARANGOIMP}
${PROJECT_SOURCE_DIR}/lib/Basics/WorkMonitorDummy.cpp
Import/ImportFeature.cpp
Import/ImportHelper.cpp
Import/SenderThread.cpp
Import/arangoimp.cpp
Shell/ClientFeature.cpp
Shell/ConsoleFeature.cpp
@ -254,6 +255,7 @@ add_executable(${BIN_ARANGOSH}
${ProductVersionFiles_arangosh}
${PROJECT_SOURCE_DIR}/lib/Basics/WorkMonitorDummy.cpp
Import/ImportHelper.cpp
Import/SenderThread.cpp
Shell/ClientFeature.cpp
Shell/ConsoleFeature.cpp
Shell/ShellFeature.cpp

4
arangosh/Dump/DumpFeature.cpp
View File

@ -1024,9 +1024,9 @@ void DumpFeature::start() {
std::string dbName = client->databaseName();
_httpClient->setLocationRewriter(static_cast<void*>(client),
_httpClient->params().setLocationRewriter(static_cast<void*>(client),
&rewriteLocation);
_httpClient->setUserNamePassword("/", client->username(), client->password());
_httpClient->params().setUserNamePassword("/", client->username(), client->password());
std::string const versionString = _httpClient->getServerVersion();

4
arangosh/Export/ExportFeature.cpp
View File

@ -228,8 +228,8 @@ void ExportFeature::start() {
FATAL_ERROR_EXIT();
}
httpClient->setLocationRewriter(static_cast<void*>(client), &rewriteLocation);
httpClient->setUserNamePassword("/", client->username(), client->password());
httpClient->params().setLocationRewriter(static_cast<void*>(client), &rewriteLocation);
httpClient->params().setUserNamePassword("/", client->username(), client->password());
// must stay here in order to establish the connection
httpClient->getServerVersion();

197
arangosh/Import/ImportFeature.cpp
View File

@ -32,8 +32,8 @@
#include "SimpleHttpClient/GeneralClientConnection.h"
#include "SimpleHttpClient/SimpleHttpClient.h"
#include <regex>
#include <iostream>
#include <regex>
using namespace arangodb;
using namespace arangodb::basics;
@ -47,6 +47,7 @@ ImportFeature::ImportFeature(application_features::ApplicationServer* server,
_useBackslash(false),
_convert(true),
_chunkSize(1024 * 1024 * 16),
_threadCount(2),
_collectionName(""),
_fromCollectionPrefix(""),
_toCollectionPrefix(""),
@ -81,29 +82,41 @@ void ImportFeature::collectOptions(
"size for individual data batches (in bytes)",
new UInt64Parameter(&_chunkSize));
options->addOption(
"--threads",
"Number of parallel import threads. Most useful for the rocksdb engine",
new UInt32Parameter(&_threadCount));
options->addOption("--collection", "collection name",
new StringParameter(&_collectionName));
options->addOption("--from-collection-prefix", "_from collection name prefix (will be prepended to all values in '_from')",
options->addOption("--from-collection-prefix",
"_from collection name prefix (will be prepended to all "
"values in '_from')",
new StringParameter(&_fromCollectionPrefix));
options->addOption("--to-collection-prefix", "_to collection name prefix (will be prepended to all values in '_to')",
new StringParameter(&_toCollectionPrefix));
options->addOption(
"--to-collection-prefix",
"_to collection name prefix (will be prepended to all values in '_to')",
new StringParameter(&_toCollectionPrefix));
options->addOption("--create-collection",
"create collection if it does not yet exist",
new BooleanParameter(&_createCollection));
options->addOption("--skip-lines",
"number of lines to skip for formats (csv and tsv only)",
new UInt64Parameter(&_rowsToSkip));
options->addOption("--convert",
"convert the strings 'null', 'false', 'true' and strings containing numbers into non-string types (csv and tsv only)",
"convert the strings 'null', 'false', 'true' and strings "
"containing numbers into non-string types (csv and tsv "
"only)",
new BooleanParameter(&_convert));
options->addOption("--translate",
"translate an attribute name (use as --translate \"from=to\", for csv and tsv only)",
"translate an attribute name (use as --translate "
"\"from=to\", for csv and tsv only)",
new VectorParameter<StringParameter>(&_translations));
std::unordered_set<std::string> types = {"document", "edge"};
@ -116,7 +129,8 @@ void ImportFeature::collectOptions(
new DiscreteValuesParameter<StringParameter>(&_createCollectionType,
types));
std::unordered_set<std::string> imports = {"csv", "tsv", "json", "jsonl", "auto"};
std::unordered_set<std::string> imports = {"csv", "tsv", "json", "jsonl",
"auto"};
options->addOption(
"--type", "type of import file",
@ -142,12 +156,12 @@ void ImportFeature::collectOptions(
std::vector<std::string> actionsVector(actions.begin(), actions.end());
std::string actionsJoined = StringUtils::join(actionsVector, ", ");
options->addOption(
"--on-duplicate",
"action to perform when a unique key constraint "
"violation occurs. Possible values: " +
actionsJoined,
new DiscreteValuesParameter<StringParameter>(&_onDuplicateAction, actions));
options->addOption("--on-duplicate",
"action to perform when a unique key constraint "
"violation occurs. Possible values: " +
actionsJoined,
new DiscreteValuesParameter<StringParameter>(
&_onDuplicateAction, actions));
}
void ImportFeature::validateOptions(
@ -162,84 +176,110 @@ void ImportFeature::validateOptions(
_filename = positionals[0];
}
} else if (1 < n) {
LOG_TOPIC(FATAL, arangodb::Logger::FIXME) << "expecting at most one filename, got " +
StringUtils::join(positionals, ", ");
LOG_TOPIC(FATAL, arangodb::Logger::FIXME)
<< "expecting at most one filename, got " +
StringUtils::join(positionals, ", ");
FATAL_ERROR_EXIT();
}
static unsigned const MaxBatchSize = 768 * 1024 * 1024;
if (_chunkSize > MaxBatchSize) {
// it's not sensible to raise the batch size beyond this value
// because the server has a built-in limit for the batch size too
// it's not sensible to raise the batch size beyond this value
// because the server has a built-in limit for the batch size too
// and will reject bigger HTTP request bodies
LOG_TOPIC(WARN, arangodb::Logger::FIXME) << "capping --batch-size value to " << MaxBatchSize;
LOG_TOPIC(WARN, arangodb::Logger::FIXME) << "capping --batch-size value to "
<< MaxBatchSize;
_chunkSize = MaxBatchSize;
}
if (_threadCount <= 1) {
// it's not sensible to use just one thread
LOG_TOPIC(WARN, arangodb::Logger::FIXME) << "capping --threads value to "
<< 1;
_threadCount = 1;
}
if (_threadCount > TRI_numberProcessors()) {
// it's not sensible to use just one thread
LOG_TOPIC(WARN, arangodb::Logger::FIXME) << "capping --threads value to "
<< TRI_numberProcessors();
_threadCount = (uint32_t)TRI_numberProcessors();
}
for (auto const& it : _translations) {
auto parts = StringUtils::split(it, "=");
if (parts.size() != 2) {
LOG_TOPIC(FATAL, arangodb::Logger::FIXME) << "invalid translation '" << it << "'";
LOG_TOPIC(FATAL, arangodb::Logger::FIXME) << "invalid translation '" << it
<< "'";
FATAL_ERROR_EXIT();
}
}
StringUtils::trimInPlace(parts[0]);
StringUtils::trimInPlace(parts[1]);
if (parts[0].empty() || parts[1].empty()) {
LOG_TOPIC(FATAL, arangodb::Logger::FIXME) << "invalid translation '" << it << "'";
LOG_TOPIC(FATAL, arangodb::Logger::FIXME) << "invalid translation '" << it
<< "'";
FATAL_ERROR_EXIT();
}
}
}
void ImportFeature::start() {
ClientFeature* client = application_features::ApplicationServer::getFeature<ClientFeature>("Client");
ClientFeature* client =
application_features::ApplicationServer::getFeature<ClientFeature>(
"Client");
int ret = EXIT_SUCCESS;
*_result = ret;
if (_typeImport == "auto") {
std::regex re = std::regex(".*?\\.([a-zA-Z]+)", std::regex::ECMAScript);
std::smatch match;
if (!std::regex_match(_filename, match, re)) {
LOG_TOPIC(FATAL, arangodb::Logger::FIXME)
<< "Cannot auto-detect file type from filename '" << _filename << "'";
FATAL_ERROR_EXIT();
}
std::string extension = match[1].str();
if (extension == "json" || extension == "jsonl" || extension == "csv" ||
extension == "tsv") {
_typeImport = extension;
} else {
LOG_TOPIC(FATAL, arangodb::Logger::FIXME)
<< "Unsupported file extension '" << extension << "'";
FATAL_ERROR_EXIT();
}
}
std::unique_ptr<SimpleHttpClient> httpClient;
try {
httpClient = client->createHttpClient();
} catch (...) {
LOG_TOPIC(FATAL, arangodb::Logger::FIXME) << "cannot create server connection, giving up!";
LOG_TOPIC(FATAL, arangodb::Logger::FIXME)
<< "cannot create server connection, giving up!";
FATAL_ERROR_EXIT();
}
httpClient->setLocationRewriter(static_cast<void*>(client), &rewriteLocation);
httpClient->setUserNamePassword("/", client->username(), client->password());
httpClient->params().setLocationRewriter(static_cast<void*>(client),
&rewriteLocation);
httpClient->params().setUserNamePassword("/", client->username(),
client->password());
// must stay here in order to establish the connection
httpClient->getServerVersion();
if (!httpClient->isConnected()) {
LOG_TOPIC(ERR, arangodb::Logger::FIXME) << "Could not connect to endpoint '" << client->endpoint()
<< "', database: '" << client->databaseName() << "', username: '"
<< client->username() << "'";
LOG_TOPIC(FATAL, arangodb::Logger::FIXME) << httpClient->getErrorMessage() << "'";
LOG_TOPIC(ERR, arangodb::Logger::FIXME)
<< "Could not connect to endpoint '" << client->endpoint()
<< "', database: '" << client->databaseName() << "', username: '"
<< client->username() << "'";
LOG_TOPIC(FATAL, arangodb::Logger::FIXME) << httpClient->getErrorMessage()
<< "'";
FATAL_ERROR_EXIT();
}
if (_typeImport == "auto") {
std::regex re = std::regex(".*?\\.([a-zA-Z]+)", std::regex::ECMAScript);
std::smatch match;
if (!std::regex_match(_filename, match, re)) {
LOG_TOPIC(FATAL, arangodb::Logger::FIXME) << "Cannot auto-detect file type from filename '" << _filename << "'";
FATAL_ERROR_EXIT();
}
std::string extension = match[1].str();
if (extension == "json" || extension == "jsonl" || extension == "csv" || extension == "tsv") {
_typeImport = extension;
} else {
LOG_TOPIC(FATAL, arangodb::Logger::FIXME) << "Unsupported file extension '" << extension << "'";
FATAL_ERROR_EXIT();
}
}
// successfully connected
std::cout << "Connected to ArangoDB '"
<< httpClient->getEndpointSpecification() << "', version "
@ -248,10 +288,12 @@ void ImportFeature::start() {
<< "'" << std::endl;
std::cout << "----------------------------------------" << std::endl;
std::cout << "database: " << client->databaseName() << std::endl;
std::cout << "database: " << client->databaseName()
<< std::endl;
std::cout << "collection: " << _collectionName << std::endl;
if (!_fromCollectionPrefix.empty()) {
std::cout << "from collection prefix: " << _fromCollectionPrefix << std::endl;
std::cout << "from collection prefix: " << _fromCollectionPrefix
<< std::endl;
}
if (!_toCollectionPrefix.empty()) {
std::cout << "to collection prefix: " << _toCollectionPrefix << std::endl;
@ -263,16 +305,21 @@ void ImportFeature::start() {
if (_typeImport == "csv") {
std::cout << "quote: " << _quote << std::endl;
}
}
if (_typeImport == "csv" || _typeImport == "tsv") {
std::cout << "separator: " << _separator << std::endl;
}
std::cout << "connect timeout: " << client->connectionTimeout() << std::endl;
std::cout << "request timeout: " << client->requestTimeout() << std::endl;
std::cout << "connect timeout: " << client->connectionTimeout()
<< std::endl;
std::cout << "request timeout: " << client->requestTimeout()
<< std::endl;
std::cout << "----------------------------------------" << std::endl;
httpClient->disconnect(); // we do not reuse this anymore
arangodb::import::ImportHelper ih(httpClient.get(), _chunkSize);
SimpleHttpClientParams params = httpClient->params();
arangodb::import::ImportHelper ih(client, client->endpoint(), params,
_chunkSize, _threadCount);
// create colletion
if (_createCollection) {
@ -287,11 +334,11 @@ void ImportFeature::start() {
ih.setRowsToSkip(static_cast<size_t>(_rowsToSkip));
ih.setOverwrite(_overwrite);
ih.useBackslash(_useBackslash);
std::unordered_map<std::string, std::string> translations;
std::unordered_map<std::string, std::string> translations;
for (auto const& it : _translations) {
auto parts = StringUtils::split(it, "=");
TRI_ASSERT(parts.size() == 2); // already validated before
TRI_ASSERT(parts.size() == 2); // already validated before
StringUtils::trimInPlace(parts[0]);
StringUtils::trimInPlace(parts[1]);
@ -304,7 +351,8 @@ void ImportFeature::start() {
if (_quote.length() <= 1) {
ih.setQuote(_quote);
} else {
LOG_TOPIC(FATAL, arangodb::Logger::FIXME) << "Wrong length of quote character.";
LOG_TOPIC(FATAL, arangodb::Logger::FIXME)
<< "Wrong length of quote character.";
FATAL_ERROR_EXIT();
}
@ -316,10 +364,12 @@ void ImportFeature::start() {
}
// separator
if (_separator.length() == 1 || _separator == "\\r" || _separator == "\\n" || _separator == "\\t") {
if (_separator.length() == 1 || _separator == "\\r" || _separator == "\\n" ||
_separator == "\\t") {
ih.setSeparator(_separator);
} else {
LOG_TOPIC(FATAL, arangodb::Logger::FIXME) << "_separator must be exactly one character.";
LOG_TOPIC(FATAL, arangodb::Logger::FIXME)
<< "_separator must be exactly one character.";
FATAL_ERROR_EXIT();
}
@ -337,12 +387,15 @@ void ImportFeature::start() {
if (_filename != "-" && !FileUtils::isRegularFile(_filename)) {
if (!FileUtils::exists(_filename)) {
LOG_TOPIC(FATAL, arangodb::Logger::FIXME) << "Cannot open file '" << _filename << "'. File not found.";
LOG_TOPIC(FATAL, arangodb::Logger::FIXME)
<< "Cannot open file '" << _filename << "'. File not found.";
} else if (FileUtils::isDirectory(_filename)) {
LOG_TOPIC(FATAL, arangodb::Logger::FIXME) << "Specified file '" << _filename
<< "' is a directory. Please use a regular file.";
LOG_TOPIC(FATAL, arangodb::Logger::FIXME)
<< "Specified file '" << _filename
<< "' is a directory. Please use a regular file.";
} else {
LOG_TOPIC(FATAL, arangodb::Logger::FIXME) << "Cannot open '" << _filename << "'. Invalid file type.";
LOG_TOPIC(FATAL, arangodb::Logger::FIXME) << "Cannot open '" << _filename
<< "'. Invalid file type.";
}
FATAL_ERROR_EXIT();
@ -389,7 +442,8 @@ void ImportFeature::start() {
}
else {
LOG_TOPIC(FATAL, arangodb::Logger::FIXME) << "Wrong type '" << _typeImport << "'.";
LOG_TOPIC(FATAL, arangodb::Logger::FIXME) << "Wrong type '" << _typeImport
<< "'.";
FATAL_ERROR_EXIT();
}
@ -407,12 +461,15 @@ void ImportFeature::start() {
}
} else {
LOG_TOPIC(ERR, arangodb::Logger::FIXME) << "error message: " << ih.getErrorMessage();
LOG_TOPIC(ERR, arangodb::Logger::FIXME) << "error message: "
<< ih.getErrorMessage();
}
} catch (std::exception const& ex) {
LOG_TOPIC(ERR, arangodb::Logger::FIXME) << "Caught exception " << ex.what() << " during import";
LOG_TOPIC(ERR, arangodb::Logger::FIXME) << "Caught exception " << ex.what()
<< " during import";
} catch (...) {
LOG_TOPIC(ERR, arangodb::Logger::FIXME) << "Got an unknown exception during import";
LOG_TOPIC(ERR, arangodb::Logger::FIXME)
<< "Got an unknown exception during import";
}
*_result = ret;

8
arangosh/Import/ImportFeature.h
View File

@ -34,10 +34,9 @@ class SimpleHttpResult;
}
class ImportFeature final : public application_features::ApplicationFeature,
public ArangoClientHelper {
public ArangoClientHelper {
public:
ImportFeature(application_features::ApplicationServer* server,
int* result);
ImportFeature(application_features::ApplicationServer* server, int* result);
public:
void collectOptions(std::shared_ptr<options::ProgramOptions>) override;
@ -50,6 +49,7 @@ class ImportFeature final : public application_features::ApplicationFeature,
bool _useBackslash;
bool _convert;
uint64_t _chunkSize;
uint32_t _threadCount;
std::string _collectionName;
std::string _fromCollectionPrefix;
std::string _toCollectionPrefix;
@ -63,7 +63,7 @@ class ImportFeature final : public application_features::ApplicationFeature,
bool _progress;
std::string _onDuplicateAction;
uint64_t _rowsToSkip;
int* _result;
};
}

232
arangosh/Import/ImportHelper.cpp
View File

@ -23,15 +23,17 @@
////////////////////////////////////////////////////////////////////////////////
#include "ImportHelper.h"
#include "Basics/MutexLocker.h"
#include "Basics/OpenFilesTracker.h"
#include "Basics/StringUtils.h"
#include "Basics/files.h"
#include "Logger/Logger.h"
#include "Basics/tri-strings.h"
#include "Basics/VelocyPackHelper.h"
#include "Basics/files.h"
#include "Basics/tri-strings.h"
#include "Import/SenderThread.h"
#include "Logger/Logger.h"
#include "Rest/GeneralResponse.h"
#include "Rest/HttpRequest.h"
#include "Shell/ClientFeature.h"
#include "SimpleHttpClient/SimpleHttpClient.h"
#include "SimpleHttpClient/SimpleHttpResult.h"
@ -136,9 +138,11 @@ double const ImportHelper::ProgressStep = 3.0;
/// constructor and destructor
////////////////////////////////////////////////////////////////////////////////
ImportHelper::ImportHelper(httpclient::SimpleHttpClient* client,
uint64_t maxUploadSize)
: _client(client),
ImportHelper::ImportHelper(ClientFeature const* client,
std::string const& endpoint,
httpclient::SimpleHttpClientParams const& params,
uint64_t maxUploadSize, uint32_t threadCount)
: _httpClient(client->createHttpClient(endpoint, params)),
_maxUploadSize(maxUploadSize),
_separator(","),
_quote("\""),
@ -149,11 +153,6 @@ ImportHelper::ImportHelper(httpclient::SimpleHttpClient* client,
_overwrite(false),
_progress(false),
_firstChunk(true),
_numberLines(0),
_numberCreated(0),
_numberErrors(0),
_numberUpdated(0),
_numberIgnored(0),
_rowsRead(0),
_rowOffset(0),
_rowsToSkip(0),
@ -161,11 +160,20 @@ ImportHelper::ImportHelper(httpclient::SimpleHttpClient* client,
_onDuplicateAction("error"),
_collectionName(),
_lineBuffer(TRI_UNKNOWN_MEM_ZONE),
_outputBuffer(TRI_UNKNOWN_MEM_ZONE) {
_hasError = false;
_outputBuffer(TRI_UNKNOWN_MEM_ZONE),
_hasError(false) {
for (uint32_t i = 0; i < threadCount; i++) {
auto http = client->createHttpClient(endpoint, params);
_senderThreads.emplace_back(new SenderThread(std::move(http), &_stats));
_senderThreads.back()->start();
}
}
ImportHelper::~ImportHelper() {}
ImportHelper::~ImportHelper() {
for (auto const& t : _senderThreads) {
t->beginShutdown();
}
}
////////////////////////////////////////////////////////////////////////////////
/// @brief imports a delimited file
@ -276,6 +284,7 @@ bool ImportHelper::importDelimited(std::string const& collectionName,
TRI_TRACKED_CLOSE_FILE(fd);
}
waitForSenders();
_outputBuffer.clear();
return !_hasError;
}
@ -400,11 +409,12 @@ bool ImportHelper::importJson(std::string const& collectionName,
TRI_TRACKED_CLOSE_FILE(fd);
}
waitForSenders();
MUTEX_LOCKER(guard, _stats._mutex);
// this is an approximation only. _numberLines is more meaningful for CSV
// imports
_numberLines =
_numberErrors + _numberCreated + _numberIgnored + _numberUpdated;
_numberLines = _stats._numberErrors + _stats._numberCreated +
_stats._numberIgnored + _stats._numberUpdated;
_outputBuffer.clear();
return !_hasError;
}
@ -426,15 +436,17 @@ void ImportHelper::reportProgress(int64_t totalLength, int64_t totalRead,
static int64_t nextProcessed = 10 * 1000 * 1000;
if (totalRead >= nextProcessed) {
LOG_TOPIC(INFO, arangodb::Logger::FIXME) << "processed " << totalRead << " bytes of input file";
LOG_TOPIC(INFO, arangodb::Logger::FIXME) << "processed " << totalRead
<< " bytes of input file";
nextProcessed += 10 * 1000 * 1000;
}
} else {
double pct = 100.0 * ((double)totalRead / (double)totalLength);
if (pct >= nextProgress && totalLength >= 1024) {
LOG_TOPIC(INFO, arangodb::Logger::FIXME) << "processed " << totalRead << " bytes (" << (int)nextProgress
<< "%) of input file";
LOG_TOPIC(INFO, arangodb::Logger::FIXME)
<< "processed " << totalRead << " bytes (" << (int)nextProgress
<< "%) of input file";
nextProgress = (double)((int)(pct + ProgressStep));
}
}
@ -459,7 +471,8 @@ void ImportHelper::ProcessCsvBegin(TRI_csv_parser_t* parser, size_t row) {
void ImportHelper::beginLine(size_t row) {
if (_lineBuffer.length() > 0) {
// error
++_numberErrors;
MUTEX_LOCKER(guard, _stats._mutex);
++_stats._numberErrors;
_lineBuffer.clear();
}
@ -483,7 +496,7 @@ void ImportHelper::ProcessCsvAdd(TRI_csv_parser_t* parser, char const* field,
if (importHelper->getRowsRead() < importHelper->getRowsToSkip()) {
return;
}
importHelper->addField(field, fieldLength, row, column, escaped);
}
@ -502,11 +515,13 @@ void ImportHelper::addField(char const* field, size_t fieldLength, size_t row,
}
}
if (_keyColumn == -1 && row == _rowsToSkip && fieldLength == 4 && memcmp(field, "_key", 4) == 0) {
if (_keyColumn == -1 && row == _rowsToSkip && fieldLength == 4 &&
memcmp(field, "_key", 4) == 0) {
_keyColumn = column;
}
if (row == _rowsToSkip || escaped || _keyColumn == static_cast<decltype(_keyColumn)>(column)) {
if (row == _rowsToSkip || escaped ||
_keyColumn == static_cast<decltype(_keyColumn)>(column)) {
// head line or escaped value
_lineBuffer.appendJsonEncoded(field, fieldLength);
return;
@ -536,9 +551,10 @@ void ImportHelper::addField(char const* field, size_t fieldLength, size_t row,
if (fieldLength > 8) {
// long integer numbers might be problematic. check if we get out of
// range
(void) std::stoll(std::string(
field,
fieldLength)); // this will fail if the number cannot be converted
(void)std::stoll(std::string(field,
fieldLength)); // this will fail if the
// number cannot be
// converted
}
int64_t num = StringUtils::int64(field, fieldLength);
@ -595,12 +611,12 @@ void ImportHelper::ProcessCsvEnd(TRI_csv_parser_t* parser, char const* field,
size_t fieldLength, size_t row, size_t column,
bool escaped) {
auto importHelper = static_cast<ImportHelper*>(parser->_dataAdd);
if (importHelper->getRowsRead() < importHelper->getRowsToSkip()) {
importHelper->incRowsRead();
return;
}
importHelper->addLastField(field, fieldLength, row, column, escaped);
importHelper->incRowsRead();
}
@ -622,7 +638,8 @@ void ImportHelper::addLastField(char const* field, size_t fieldLength,
_firstLine = _lineBuffer.c_str();
} else if (row > _rowsToSkip && _firstLine.empty()) {
// error
++_numberErrors;
MUTEX_LOCKER(guard, _stats._mutex);
++_stats._numberErrors;
_lineBuffer.reset();
return;
}
@ -633,7 +650,8 @@ void ImportHelper::addLastField(char const* field, size_t fieldLength,
_outputBuffer.appendText(_lineBuffer);
_lineBuffer.reset();
} else {
++_numberErrors;
MUTEX_LOCKER(guard, _stats._mutex);
++_stats._numberErrors;
}
if (_outputBuffer.length() > _maxUploadSize) {
@ -646,12 +664,12 @@ void ImportHelper::addLastField(char const* field, size_t fieldLength,
/// @brief check if we must create the target collection, and create it if
/// required
////////////////////////////////////////////////////////////////////////////////
bool ImportHelper::checkCreateCollection() {
if (!_firstChunk || !_createCollection) {
if (!_createCollection) {
return true;
}
std::string const url("/_api/collection");
VPackBuilder builder;
@ -661,35 +679,66 @@ bool ImportHelper::checkCreateCollection() {
builder.close();
std::string data = builder.slice().toJson();
std::unordered_map<std::string, std::string> headerFields;
std::unique_ptr<SimpleHttpResult> result(_client->request(
rest::RequestType::POST, url, data.c_str(),
data.size(), headerFields));
std::unique_ptr<SimpleHttpResult> result(_httpClient->request(
rest::RequestType::POST, url, data.c_str(), data.size(), headerFields));
if (result == nullptr) {
return false;
}
auto code = static_cast<rest::ResponseCode>(result->getHttpReturnCode());
if (code == rest::ResponseCode::CONFLICT ||
code == rest::ResponseCode::OK ||
if (code == rest::ResponseCode::CONFLICT || code == rest::ResponseCode::OK ||
code == rest::ResponseCode::CREATED ||
code == rest::ResponseCode::ACCEPTED) {
// collection already exists or was created successfully
return true;
}
LOG_TOPIC(ERR, arangodb::Logger::FIXME) << "unable to create collection '" << _collectionName << "', server returned status code: " << static_cast<int>(code);
LOG_TOPIC(ERR, arangodb::Logger::FIXME)
<< "unable to create collection '" << _collectionName
<< "', server returned status code: " << static_cast<int>(code);
_hasError = true;
return false;
}
bool ImportHelper::truncateCollection() {
if (!_overwrite) {
return true;
}
std::string const url = "/_api/collection/" + _collectionName + "/truncate";
std::string data = "";// never send an completly empty string
std::unordered_map<std::string, std::string> headerFields;
std::unique_ptr<SimpleHttpResult> result(_httpClient->request(
rest::RequestType::PUT, url, data.c_str(), data.size(), headerFields));
if (result == nullptr) {
return false;
}
auto code = static_cast<rest::ResponseCode>(result->getHttpReturnCode());
if (code == rest::ResponseCode::CONFLICT || code == rest::ResponseCode::OK ||
code == rest::ResponseCode::CREATED ||
code == rest::ResponseCode::ACCEPTED) {
// collection already exists or was created successfully
return true;
}
LOG_TOPIC(ERR, arangodb::Logger::FIXME)
<< "unable to truncate collection '" << _collectionName
<< "', server returned status code: " << static_cast<int>(code);
_hasError = true;
_errorMessage = "Unable to overwrite collection";
return false;
}
void ImportHelper::sendCsvBuffer() {
if (_hasError) {
return;
}
if (!checkCreateCollection()) {
return;
}
@ -706,16 +755,15 @@ void ImportHelper::sendCsvBuffer() {
url += "&toPrefix=" + StringUtils::urlEncode(_toCollectionPrefix);
}
if (_firstChunk && _overwrite) {
url += "&overwrite=true";
// url += "&overwrite=true";
truncateCollection();
}
_firstChunk = false;
std::unique_ptr<SimpleHttpResult> result(_client->request(
rest::RequestType::POST, url, _outputBuffer.c_str(),
_outputBuffer.length(), headerFields));
handleResult(result.get());
SenderThread* t = findSender();
if (t != nullptr) {
t->sendData(url, &_outputBuffer);
}
_outputBuffer.reset();
_rowOffset = _rowsRead;
@ -747,70 +795,48 @@ void ImportHelper::sendJsonBuffer(char const* str, size_t len, bool isObject) {
url += "&toPrefix=" + StringUtils::urlEncode(_toCollectionPrefix);
}
if (_firstChunk && _overwrite) {
url += "&overwrite=true";
// url += "&overwrite=true";
truncateCollection();
}
_firstChunk = false;
std::unordered_map<std::string, std::string> headerFields;
std::unique_ptr<SimpleHttpResult> result(_client->request(
rest::RequestType::POST, url, str, len, headerFields));
handleResult(result.get());
SenderThread* t = findSender();
if (t != nullptr) {
StringBuffer buff(TRI_UNKNOWN_MEM_ZONE, len, false);
buff.appendText(str, len);
t->sendData(url, &buff);
}
}
void ImportHelper::handleResult(SimpleHttpResult* result) {
if (result == nullptr) {
return;
}
std::shared_ptr<VPackBuilder> parsedBody;
try {
parsedBody = result->getBodyVelocyPack();
} catch (...) {
// No action required
return;
}
VPackSlice const body = parsedBody->slice();
// error details
VPackSlice const details = body.get("details");
if (details.isArray()) {
for (VPackSlice const& detail : VPackArrayIterator(details)) {
if (detail.isString()) {
LOG_TOPIC(WARN, arangodb::Logger::FIXME) << "" << detail.copyString();
SenderThread* ImportHelper::findSender() {
while (!_senderThreads.empty()) {
for (std::unique_ptr<SenderThread>& t : _senderThreads) {
if (t->hasError()) {
_hasError = true;
_errorMessage = t->errorMessage();
return nullptr;
} else if (t->idle()) {
return t.get();
}
}
usleep(500000);
}
return nullptr;
}
// get the "error" flag. This returns a pointer, not a copy
if (arangodb::basics::VelocyPackHelper::getBooleanValue(body, "error",
false)) {
_hasError = true;
// get the error message
VPackSlice const errorMessage = body.get("errorMessage");
if (errorMessage.isString()) {
_errorMessage = errorMessage.copyString();
void ImportHelper::waitForSenders() {
while (!_senderThreads.empty()) {
uint32_t numIdle = 0;
for (std::unique_ptr<SenderThread>& t : _senderThreads) {
if (t->idle() || t->hasError()) {
numIdle++;
}
}
if (numIdle == _senderThreads.size()) {
return;
}
usleep(100000);
}
// look up the "created" flag
_numberCreated += arangodb::basics::VelocyPackHelper::getNumericValue<size_t>(
body, "created", 0);
// look up the "errors" flag
_numberErrors += arangodb::basics::VelocyPackHelper::getNumericValue<size_t>(
body, "errors", 0);
// look up the "updated" flag
_numberUpdated += arangodb::basics::VelocyPackHelper::getNumericValue<size_t>(
body, "updated", 0);
// look up the "ignored" flag
_numberIgnored += arangodb::basics::VelocyPackHelper::getNumericValue<size_t>(
body, "ignored", 0);
}
}
}

68
arangosh/Import/ImportHelper.h
View File

@ -26,18 +26,21 @@
#define ARANGODB_IMPORT_IMPORT_HELPER_H 1
#include "Basics/Common.h"
#include "Basics/Mutex.h"
#include "Basics/csv.h"
#include "Basics/StringBuffer.h"
#include "Basics/csv.h"
#ifdef _WIN32
#include "Basics/win-utils.h"
#endif
namespace arangodb {
class ClientFeature;
namespace httpclient {
class SimpleHttpClient;
class SimpleHttpResult;
struct SimpleHttpClientParams;
}
}
@ -47,6 +50,16 @@ class SimpleHttpResult;
namespace arangodb {
namespace import {
class SenderThread;
struct ImportStatistics {
size_t _numberCreated = 0;
size_t _numberErrors = 0;
size_t _numberUpdated = 0;
size_t _numberIgnored = 0;
arangodb::Mutex _mutex;
};
class ImportHelper {
public:
@ -61,7 +74,9 @@ class ImportHelper {
ImportHelper& operator=(ImportHelper const&) = delete;
public:
ImportHelper(httpclient::SimpleHttpClient* client, uint64_t maxUploadSize);
ImportHelper(ClientFeature const* client, std::string const& endpoint,
httpclient::SimpleHttpClientParams const& params,
uint64_t maxUploadSize, uint32_t threadCount);
~ImportHelper();
@ -79,8 +94,7 @@ class ImportHelper {
//////////////////////////////////////////////////////////////////////////////
bool importJson(std::string const& collectionName,
std::string const& fileName,
bool assumeLinewise);
std::string const& fileName, bool assumeLinewise);
//////////////////////////////////////////////////////////////////////////////
/// @brief sets the action to carry out on duplicate _key
@ -102,17 +116,13 @@ class ImportHelper {
/// @brief set collection name prefix for _from
//////////////////////////////////////////////////////////////////////////////
void setFrom (std::string const& from) {
_fromCollectionPrefix = from;
}
void setFrom(std::string const& from) { _fromCollectionPrefix = from; }
//////////////////////////////////////////////////////////////////////////////
/// @brief set collection name prefix for _to
//////////////////////////////////////////////////////////////////////////////
void setTo (std::string const& to) {
_toCollectionPrefix = to;
}
void setTo(std::string const& to) { _toCollectionPrefix = to; }
//////////////////////////////////////////////////////////////////////////////
/// @brief whether or not backslashes can be used for escaping quotes
@ -139,7 +149,8 @@ class ImportHelper {
_createCollectionType = value;
}
void setTranslations(std::unordered_map<std::string, std::string> const& translations) {
void setTranslations(
std::unordered_map<std::string, std::string> const& translations) {
_translations = translations;
}
@ -148,19 +159,19 @@ class ImportHelper {
//////////////////////////////////////////////////////////////////////////////
void setOverwrite(bool value) { _overwrite = value; }
//////////////////////////////////////////////////////////////////////////////
/// @brief set the number of rows to skip
//////////////////////////////////////////////////////////////////////////////
void setRowsToSkip(size_t value) { _rowsToSkip = value; }
//////////////////////////////////////////////////////////////////////////////
/// @brief get the number of rows to skip
//////////////////////////////////////////////////////////////////////////////
size_t getRowsToSkip() const { return _rowsToSkip; }
//////////////////////////////////////////////////////////////////////////////
/// @brief whether or not to convert strings that contain "null", "false",
/// "true" or that look like numbers into those types
@ -184,36 +195,36 @@ class ImportHelper {
/// @brief get the number of documents imported
//////////////////////////////////////////////////////////////////////////////
size_t getNumberCreated() { return _numberCreated; }
size_t getNumberCreated() { return _stats._numberCreated; }
//////////////////////////////////////////////////////////////////////////////
/// @brief get the number of errors
//////////////////////////////////////////////////////////////////////////////
size_t getNumberErrors() { return _numberErrors; }
size_t getNumberErrors() { return _stats._numberErrors; }
//////////////////////////////////////////////////////////////////////////////
/// @brief get the number of updated documents
//////////////////////////////////////////////////////////////////////////////
size_t getNumberUpdated() { return _numberUpdated; }
size_t getNumberUpdated() { return _stats._numberUpdated; }
//////////////////////////////////////////////////////////////////////////////
/// @brief get the number of ignored documents
//////////////////////////////////////////////////////////////////////////////
size_t getNumberIgnored() const { return _numberIgnored; }
size_t getNumberIgnored() const { return _stats._numberIgnored; }
//////////////////////////////////////////////////////////////////////////////
/// @brief increase the row counter
//////////////////////////////////////////////////////////////////////////////
void incRowsRead() { ++_rowsRead; }
//////////////////////////////////////////////////////////////////////////////
/// @brief get the number of rows read
//////////////////////////////////////////////////////////////////////////////
size_t getRowsRead() const { return _rowsRead; }
//////////////////////////////////////////////////////////////////////////////
@ -240,13 +251,17 @@ class ImportHelper {
bool escaped);
bool checkCreateCollection();
bool truncateCollection();
void sendCsvBuffer();
void sendJsonBuffer(char const* str, size_t len, bool isObject);
void handleResult(httpclient::SimpleHttpResult* result);
SenderThread* findSender();
void waitForSenders();
private:
httpclient::SimpleHttpClient* _client;
uint64_t _maxUploadSize;
std::unique_ptr<httpclient::SimpleHttpClient> _httpClient;
uint64_t const _maxUploadSize;
std::vector<std::unique_ptr<SenderThread>> _senderThreads;
std::string _separator;
std::string _quote;
@ -259,10 +274,7 @@ class ImportHelper {
bool _firstChunk;
size_t _numberLines;
size_t _numberCreated;
size_t _numberErrors;
size_t _numberUpdated;
size_t _numberIgnored;
ImportStatistics _stats;
size_t _rowsRead;
size_t _rowOffset;

165
arangosh/Import/SenderThread.cpp Normal file
View File

@ -0,0 +1,165 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2017 ArangoDB GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
/// you may not use this file except in compliance with the License.
/// You may obtain a copy of the License at
///
/// http://www.apache.org/licenses/LICENSE-2.0
///
/// Unless required by applicable law or agreed to in writing, software
/// distributed under the License is distributed on an "AS IS" BASIS,
/// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
/// See the License for the specific language governing permissions and
/// limitations under the License.
///
/// Copyright holder is ArangoDB GmbH, Cologne, Germany
///
/// @author Simon Grätzer
////////////////////////////////////////////////////////////////////////////////
#include "SenderThread.h"
#include "Basics/Common.h"
#include "Basics/ConditionLocker.h"
#include "Basics/MutexLocker.h"
#include "Basics/StringBuffer.h"
#include "Basics/StringUtils.h"
#include "Basics/VelocyPackHelper.h"
#include "ImportHelper.h"
#include "SimpleHttpClient/SimpleHttpClient.h"
#include "SimpleHttpClient/SimpleHttpResult.h"
#include <velocypack/Builder.h>
#include <velocypack/Iterator.h>
#include <velocypack/velocypack-aliases.h>
using namespace arangodb;
using namespace arangodb::import;
SenderThread::SenderThread(
std::unique_ptr<httpclient::SimpleHttpClient>&& client,
ImportStatistics* stats)
: Thread("Import Sender"),
_client(client.release()),
_data(TRI_UNKNOWN_MEM_ZONE, false),
_hasError(false),
_idle(true),
_stats(stats) {}
SenderThread::~SenderThread() {
shutdown();
delete _client;
}
void SenderThread::beginShutdown() {
Thread::beginShutdown();
// wake up the thread that may be waiting in run()
CONDITION_LOCKER(guard, _condition);
guard.broadcast();
}
void SenderThread::sendData(std::string const& url,
arangodb::basics::StringBuffer* data) {
TRI_ASSERT(_idle && !_hasError);
_url = url;
_data.swap(data);
// wake up the thread that may be waiting in run()
_idle = false;
CONDITION_LOCKER(guard, _condition);
guard.broadcast();
}
void SenderThread::run() {
while (!isStopping() && !_hasError) {
{
CONDITION_LOCKER(guard, _condition);
guard.wait();
}
if (isStopping()) {
return;
}
try {
if (_data.length() > 0) {
TRI_ASSERT(!_idle && !_url.empty());
std::unordered_map<std::string, std::string> headerFields;
std::unique_ptr<httpclient::SimpleHttpResult> result(
_client->request(rest::RequestType::POST, _url, _data.c_str(),
_data.length(), headerFields));
handleResult(result.get());
_url.clear();
_data.reset();
}
_idle = true;
} catch (...) {
_hasError = true;
_idle = true;
}
}
}
void SenderThread::handleResult(httpclient::SimpleHttpResult* result) {
if (result == nullptr) {
return;
}
std::shared_ptr<VPackBuilder> parsedBody;
try {
parsedBody = result->getBodyVelocyPack();
} catch (...) {
// No action required
return;
}
VPackSlice const body = parsedBody->slice();
// error details
VPackSlice const details = body.get("details");
if (details.isArray()) {
for (VPackSlice const& detail : VPackArrayIterator(details)) {
if (detail.isString()) {
LOG_TOPIC(WARN, arangodb::Logger::FIXME) << "" << detail.copyString();
}
}
}
// get the "error" flag. This returns a pointer, not a copy
if (arangodb::basics::VelocyPackHelper::getBooleanValue(body, "error",
false)) {
_hasError = true;
// get the error message
VPackSlice const errorMessage = body.get("errorMessage");
if (errorMessage.isString()) {
_errorMessage = errorMessage.copyString();
}
}
MUTEX_LOCKER(guard, _stats->_mutex);
// look up the "created" flag
_stats->_numberCreated +=
arangodb::basics::VelocyPackHelper::getNumericValue<size_t>(body,
"created", 0);
// look up the "errors" flag
_stats->_numberErrors +=
arangodb::basics::VelocyPackHelper::getNumericValue<size_t>(body,
"errors", 0);
// look up the "updated" flag
_stats->_numberUpdated +=
arangodb::basics::VelocyPackHelper::getNumericValue<size_t>(body,
"updated", 0);
// look up the "ignored" flag
_stats->_numberIgnored +=
arangodb::basics::VelocyPackHelper::getNumericValue<size_t>(body,
"ignored", 0);
}

85
arangosh/Import/SenderThread.h Normal file
View File

@ -0,0 +1,85 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2017 ArangoDB GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
/// you may not use this file except in compliance with the License.
/// You may obtain a copy of the License at
///
/// http://www.apache.org/licenses/LICENSE-2.0
///
/// Unless required by applicable law or agreed to in writing, software
/// distributed under the License is distributed on an "AS IS" BASIS,
/// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
/// See the License for the specific language governing permissions and
/// limitations under the License.
///
/// Copyright holder is ArangoDB GmbH, Cologne, Germany
///
/// @author Simon Grätzer
////////////////////////////////////////////////////////////////////////////////
#ifndef ARANGODB_IMPORT_SEND_THREAD_H
#define ARANGODB_IMPORT_SEND_THREAD_H 1
#include "Basics/ConditionVariable.h"
#include "Basics/StringBuffer.h"
#include "Basics/Thread.h"
#include "SimpleHttpClient/SimpleHttpClient.h"
namespace arangodb {
namespace basics {
class StringBuffer;
}
namespace httpclient {
class SimpleHttpClient;
class SimpleHttpResult;
}
namespace import {
struct ImportStatistics;
class SenderThread : public arangodb::Thread {
private:
SenderThread(SenderThread const&) = delete;
SenderThread& operator=(SenderThread const&) = delete;
public:
explicit SenderThread(std::unique_ptr<httpclient::SimpleHttpClient>&&,
ImportStatistics* stats);
~SenderThread();
//////////////////////////////////////////////////////////////////////////////
/// @brief imports a delimited file
//////////////////////////////////////////////////////////////////////////////
void sendData(std::string const& url, basics::StringBuffer* sender);
bool idle() { return _idle; }
bool hasError() { return _hasError; }
std::string const& errorMessage() { return _errorMessage; }
void beginShutdown() override;
protected:
void run() override;
private:
basics::ConditionVariable _condition;
httpclient::SimpleHttpClient* _client;
std::string _url;
basics::StringBuffer _data;
bool _hasError = false;
bool _idle = true;
ImportStatistics* _stats;
std::string _errorMessage;
void handleResult(httpclient::SimpleHttpResult* result);
};
}
}
#endif

4
arangosh/Restore/RestoreFeature.cpp
View File

@ -678,9 +678,9 @@ void RestoreFeature::start() {
std::string dbName = client->databaseName();
_httpClient->setLocationRewriter(static_cast<void*>(client),
_httpClient->params().setLocationRewriter(static_cast<void*>(client),
&rewriteLocation);
_httpClient->setUserNamePassword("/", client->username(), client->password());
_httpClient->params().setUserNamePassword("/", client->username(), client->password());
int err = TRI_ERROR_NO_ERROR;
std::string versionString = _httpClient->getServerVersion(&err);

28
arangosh/Shell/ClientFeature.cpp
View File

@ -141,7 +141,7 @@ void ClientFeature::validateOptions(std::shared_ptr<ProgramOptions> options) {
_haveServerPassword = !options->processingResult().touched("server.password");
SimpleHttpClient::setMaxPacketSize(_maxPacketSize);
SimpleHttpClientParams::setDefaultMaxPacketSize(_maxPacketSize);
}
void ClientFeature::prepare() {
@ -189,25 +189,29 @@ std::unique_ptr<GeneralClientConnection> ClientFeature::createConnection(
return connection;
}
std::unique_ptr<SimpleHttpClient> ClientFeature::createHttpClient() {
std::unique_ptr<SimpleHttpClient> ClientFeature::createHttpClient() const {
return createHttpClient(_endpoint);
}
std::unique_ptr<SimpleHttpClient> ClientFeature::createHttpClient(
std::string const& definition) {
std::unique_ptr<Endpoint> endpoint(Endpoint::clientFactory(definition));
std::unique_ptr<SimpleHttpClient> ClientFeature::createHttpClient(std::string const& definition) const {
return createHttpClient(definition, SimpleHttpClientParams(_requestTimeout, _warn));
}
std::unique_ptr<httpclient::SimpleHttpClient> ClientFeature::createHttpClient(
std::string const& definition,
SimpleHttpClientParams const& params) const {
std::unique_ptr<Endpoint> endpoint(Endpoint::clientFactory(definition));
if (endpoint.get() == nullptr) {
LOG_TOPIC(ERR, arangodb::Logger::FIXME) << "invalid value for --server.endpoint ('" << definition << "')";
THROW_ARANGO_EXCEPTION(TRI_ERROR_BAD_PARAMETER);
}
std::unique_ptr<GeneralClientConnection> connection(
GeneralClientConnection::factory(endpoint, _requestTimeout,
_connectionTimeout, _retries,
_sslProtocol));
return std::make_unique<SimpleHttpClient>(connection, _requestTimeout, _warn);
std::unique_ptr<GeneralClientConnection> connection(GeneralClientConnection::factory(endpoint, _requestTimeout,
_connectionTimeout, _retries,
_sslProtocol));
return std::make_unique<SimpleHttpClient>(connection, params);
}
std::vector<std::string> ClientFeature::httpEndpoints() {

9
arangosh/Shell/ClientFeature.h
View File

@ -32,6 +32,7 @@ class Endpoint;
namespace httpclient {
class GeneralClientConnection;
class SimpleHttpClient;
struct SimpleHttpClientParams;
}
class ClientFeature final : public application_features::ApplicationFeature,
@ -70,9 +71,11 @@ class ClientFeature final : public application_features::ApplicationFeature,
std::unique_ptr<httpclient::GeneralClientConnection> createConnection();
std::unique_ptr<httpclient::GeneralClientConnection> createConnection(
std::string const& definition);
std::unique_ptr<httpclient::SimpleHttpClient> createHttpClient();
std::unique_ptr<httpclient::SimpleHttpClient> createHttpClient() const;
std::unique_ptr<httpclient::SimpleHttpClient> createHttpClient(
std::string const& definition);
std::string const& definition) const;
std::unique_ptr<httpclient::SimpleHttpClient> createHttpClient(
std::string const& definition, httpclient::SimpleHttpClientParams const&) const;
std::vector<std::string> httpEndpoints() override;
void setDatabaseName(std::string const& databaseName) {
@ -82,6 +85,8 @@ class ClientFeature final : public application_features::ApplicationFeature,
void setRetries(size_t retries) { _retries = retries; }
void setWarn(bool warn) { _warn = warn; }
bool getWarn() { return _warn; }
static int runMain(int argc, char* argv[],
std::function<int(int argc, char* argv[])> const& mainFunc);

22
arangosh/Shell/V8ClientConnection.cpp
View File

@ -71,9 +71,10 @@ void V8ClientConnection::init(
_password = password;
_databaseName = databaseName;
_client.reset(new SimpleHttpClient(connection, _requestTimeout, false));
_client->setLocationRewriter(this, &rewriteLocation);
_client->setUserNamePassword("/", _username, _password);
SimpleHttpClientParams params(_requestTimeout, false);
params.setLocationRewriter(this, &rewriteLocation);
params.setUserNamePassword("/", _username, _password);
_client.reset(new SimpleHttpClient(connection, params));
// connect to server and get version number
std::unordered_map<std::string, std::string> headerFields;
@ -946,11 +947,9 @@ static void ClientConnection_importCsv(
v8::Local<v8::External> wrap = v8::Local<v8::External>::Cast(args.Data());
ClientFeature* client = static_cast<ClientFeature*>(wrap->Value());
std::unique_ptr<SimpleHttpClient> httpClient =
client->createHttpClient(v8connection->endpointSpecification());
ImportHelper ih(httpClient.get(), DefaultChunkSize);
SimpleHttpClientParams params(client->requestTimeout(), client->getWarn());
ImportHelper ih(client, v8connection->endpointSpecification(), params,
DefaultChunkSize, 1);
ih.setQuote(quote);
ih.setSeparator(separator.c_str());
@ -1016,10 +1015,9 @@ static void ClientConnection_importJson(
v8::Local<v8::External> wrap = v8::Local<v8::External>::Cast(args.Data());
ClientFeature* client = static_cast<ClientFeature*>(wrap->Value());
std::unique_ptr<SimpleHttpClient> httpClient =
client->createHttpClient(v8connection->endpointSpecification());
ImportHelper ih(httpClient.get(), DefaultChunkSize);
SimpleHttpClientParams params(client->requestTimeout(), client->getWarn());
ImportHelper ih(client, v8connection->endpointSpecification(), params,
DefaultChunkSize, 1);
std::string fileName = TRI_ObjectToString(isolate, args[0]);
std::string collectionName = TRI_ObjectToString(isolate, args[1]);

38
js/common/modules/@arangodb/general-graph.js
View File

@ -1757,8 +1757,44 @@ class Graph {
// / @brief was docuBlock JSF_general_graph_connectingEdges
// //////////////////////////////////////////////////////////////////////////////
_getConnectingEdges (vertexExample1, vertexExample2, options) {
_connectingEdges (vertexExample1, vertexExample2, options) {
options = options || {};
if (options.vertex1CollectionRestriction) {
if (!Array.isArray(options.vertex1CollectionRestriction)) {
options.vertex1CollectionRestriction = [ options.vertex1CollectionRestriction ];
}
}
if (options.vertex2CollectionRestriction) {
if (!Array.isArray(options.vertex2CollectionRestriction)) {
options.vertex2CollectionRestriction = [ options.vertex2CollectionRestriction ];
}
}
/*var query = `
${generateWithStatement(this, optionsVertex1.hasOwnProperty('edgeCollectionRestriction') ? optionsVertex1 : optionsVertex2)}
${transformExampleToAQL(vertex1Example, Object.keys(this.__vertexCollections), bindVars, 'left')}
LET leftNeighbors = (FOR v IN ${optionsVertex1.minDepth || 1}..${optionsVertex1.maxDepth || 1} ${optionsVertex1.direction || 'ANY'} left
${buildEdgeCollectionRestriction(optionsVertex1.edgeCollectionRestriction, bindVars, this)}
OPTIONS {bfs: true, uniqueVertices: "global"}
${Array.isArray(optionsVertex1.vertexCollectionRestriction) && optionsVertex1.vertexCollectionRestriction.length > 0 ? buildVertexCollectionRestriction(optionsVertex1.vertexCollectionRestriction, 'v') : ''}
RETURN v)
${transformExampleToAQL(vertex2Example, Object.keys(this.__vertexCollections), bindVars, 'right')}
FILTER right != left
LET rightNeighbors = (FOR v IN ${optionsVertex2.minDepth || 1}..${optionsVertex2.maxDepth || 1} ${optionsVertex2.direction || 'ANY'} right
${buildEdgeCollectionRestriction(optionsVertex2.edgeCollectionRestriction, bindVars, this)}
OPTIONS {bfs: true, uniqueVertices: "global"}
${Array.isArray(optionsVertex2.vertexCollectionRestriction) && optionsVertex2.vertexCollectionRestriction.length > 0 ? buildVertexCollectionRestriction(optionsVertex2.vertexCollectionRestriction, 'v') : ''}
RETURN v)
LET neighbors = INTERSECTION(leftNeighbors, rightNeighbors)
FILTER LENGTH(neighbors) > 0 `;
if (optionsVertex1.includeData === true || optionsVertex2.includeData === true) {
query += `RETURN {left : left, right: right, neighbors: neighbors}`;
} else {
query += `RETURN {left : left._id, right: right._id, neighbors: neighbors[*]._id}`;
}
return db._query(query, bindVars).toArray();*/
// TODO
return [];
}

96
lib/SimpleHttpClient/SimpleHttpClient.cpp
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2017 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
@ -20,6 +20,7 @@
///
/// @author Dr. Frank Celler
/// @author Achim Brandt
/// @author Simon Grätzer
////////////////////////////////////////////////////////////////////////////////
#include "SimpleHttpClient.h"
@ -44,33 +45,23 @@ std::unordered_map<std::string, std::string> const
SimpleHttpClient::NO_HEADERS{};
/// @brief default value for max packet size
size_t SimpleHttpClient::MaxPacketSize = 256 * 1024 * 1024;
size_t SimpleHttpClientParams::MaxPacketSize = 256 * 1024 * 1024;
SimpleHttpClient::SimpleHttpClient(GeneralClientConnection* connection,
double requestTimeout, bool warn)
SimpleHttpClientParams const& params)
: _connection(connection),
_deleteConnectionOnDestruction(false),
_params(params),
_writeBuffer(TRI_UNKNOWN_MEM_ZONE, false),
_readBuffer(TRI_UNKNOWN_MEM_ZONE),
_readBufferOffset(0),
_requestTimeout(requestTimeout),
_state(IN_CONNECT),
_written(0),
_errorMessage(""),
_locationRewriter({nullptr, nullptr}),
_nextChunkedSize(0),
_method(rest::RequestType::GET),
_result(nullptr),
_jwt(""),
_maxPacketSize(MaxPacketSize),
_maxRetries(3),
_retryWaitTime(1 * 1000 * 1000),
_retryMessage(),
_deleteConnectionOnDestruction(false),
_keepConnectionOnDestruction(false),
_warn(warn),
_keepAlive(true),
_exposeArangoDB(true),
_supportDeflate(true) {
_result(nullptr)
{
TRI_ASSERT(connection != nullptr);
if (_connection->isConnected()) {
@ -79,17 +70,17 @@ SimpleHttpClient::SimpleHttpClient(GeneralClientConnection* connection,
}
SimpleHttpClient::SimpleHttpClient(
std::unique_ptr<GeneralClientConnection>& connection, double requestTimeout,
bool warn)
: SimpleHttpClient(connection.get(), requestTimeout, warn) {
_deleteConnectionOnDestruction = true;
connection.release();
std::unique_ptr<GeneralClientConnection>& connection,
SimpleHttpClientParams const& params)
: SimpleHttpClient(connection.get(), params) {
_deleteConnectionOnDestruction = true;
connection.release();
}
SimpleHttpClient::~SimpleHttpClient() {
// connection may have been invalidated by other objects
if (_connection != nullptr) {
if (!_keepConnectionOnDestruction || !_connection->isConnected()) {
if (!_params._keepConnectionOnDestruction || !_connection->isConnected()) {
_connection->disconnect();
}
@ -174,16 +165,16 @@ SimpleHttpResult* SimpleHttpClient::retryRequest(
delete result;
result = nullptr;
if (tries++ >= _maxRetries) {
if (tries++ >= _params._maxRetries) {
break;
}
if (!_retryMessage.empty() && (_maxRetries - tries) > 0) {
LOG_TOPIC(WARN, arangodb::Logger::FIXME) << "" << _retryMessage
<< " - retries left: " << (_maxRetries - tries);
if (!_params._retryMessage.empty() && (_params._maxRetries - tries) > 0) {
LOG_TOPIC(WARN, arangodb::Logger::FIXME) << "" << _params._retryMessage
<< " - retries left: " << (_params._maxRetries - tries);
}
usleep(static_cast<TRI_usleep_t>(_retryWaitTime));
usleep(static_cast<TRI_usleep_t>(_params._retryWaitTime));
}
return result;
@ -240,8 +231,8 @@ SimpleHttpResult* SimpleHttpClient::doRequest(
TRI_ASSERT(_state == IN_CONNECT || _state == IN_WRITE);
// respect timeout
double endTime = TRI_microtime() + _requestTimeout;
double remainingTime = _requestTimeout;
double endTime = TRI_microtime() + _params._requestTimeout;
double remainingTime = _params._requestTimeout;
bool haveSentRequest = false;
@ -441,21 +432,6 @@ void SimpleHttpClient::clearReadBuffer() {
}
}
////////////////////////////////////////////////////////////////////////////////
/// @brief sets username and password
////////////////////////////////////////////////////////////////////////////////
void SimpleHttpClient::setJwt(std::string const& jwt) { _jwt = jwt; }
void SimpleHttpClient::setUserNamePassword(std::string const& prefix,
std::string const& username,
std::string const& password) {
std::string value =
arangodb::basics::StringUtils::encodeBase64(username + ":" + password);
_pathToBasicAuth.push_back(std::make_pair(prefix, value));
}
////////////////////////////////////////////////////////////////////////////////
/// @brief return the result
////////////////////////////////////////////////////////////////////////////////
@ -537,29 +513,29 @@ void SimpleHttpClient::setRequest(
_writeBuffer.appendText(hostname);
_writeBuffer.appendText(TRI_CHAR_LENGTH_PAIR("\r\n"));
if (_keepAlive) {
if (_params._keepAlive) {
_writeBuffer.appendText(TRI_CHAR_LENGTH_PAIR("Connection: Keep-Alive\r\n"));
} else {
_writeBuffer.appendText(TRI_CHAR_LENGTH_PAIR("Connection: Close\r\n"));
}
if (_exposeArangoDB) {
if (_params._exposeArangoDB) {
_writeBuffer.appendText(TRI_CHAR_LENGTH_PAIR("User-Agent: ArangoDB\r\n"));
}
// do not automatically advertise deflate support
if (_supportDeflate) {
if (_params._supportDeflate) {
_writeBuffer.appendText(
TRI_CHAR_LENGTH_PAIR("Accept-Encoding: deflate\r\n"));
}
// do basic authorization
if (!_pathToBasicAuth.empty()) {
if (!_params._pathToBasicAuth.empty()) {
std::string foundPrefix;
std::string foundValue;
auto i = _pathToBasicAuth.begin();
auto i = _params._pathToBasicAuth.begin();
for (; i != _pathToBasicAuth.end(); ++i) {
for (; i != _params._pathToBasicAuth.end(); ++i) {
std::string& f = i->first;
if (l->compare(0, f.size(), f) == 0) {
@ -577,9 +553,9 @@ void SimpleHttpClient::setRequest(
_writeBuffer.appendText(TRI_CHAR_LENGTH_PAIR("\r\n"));
}
}
if (!_jwt.empty()) {
if (!_params._jwt.empty()) {
_writeBuffer.appendText(TRI_CHAR_LENGTH_PAIR("Authorization: bearer "));
_writeBuffer.appendText(_jwt);
_writeBuffer.appendText(_params._jwt);
_writeBuffer.appendText(TRI_CHAR_LENGTH_PAIR("\r\n"));
}
@ -692,7 +668,7 @@ void SimpleHttpClient::processHeader() {
_result->setResultType(SimpleHttpResult::COMPLETE);
_state = FINISHED;
if (!_keepAlive) {
if (!_params._keepAlive) {
_connection->disconnect();
}
return;
@ -700,7 +676,7 @@ void SimpleHttpClient::processHeader() {
// found content-length header in response
else if (_result->hasContentLength() && _result->getContentLength() > 0) {
if (_result->getContentLength() > _maxPacketSize) {
if (_result->getContentLength() > _params._maxPacketSize) {
setErrorMessage("Content-Length > max packet size found", true);
// reset connection
@ -753,7 +729,7 @@ void SimpleHttpClient::processBody() {
_result->setResultType(SimpleHttpResult::COMPLETE);
_state = FINISHED;
if (!_keepAlive) {
if (!_params._keepAlive) {
_connection->disconnect();
}
@ -790,7 +766,7 @@ void SimpleHttpClient::processBody() {
_result->setResultType(SimpleHttpResult::COMPLETE);
_state = FINISHED;
if (!_keepAlive) {
if (!_params._keepAlive) {
_connection->disconnect();
}
}
@ -847,7 +823,7 @@ void SimpleHttpClient::processChunkedHeader() {
}
// failed: too many bytes
if (contentLength > _maxPacketSize) {
if (contentLength > _params._maxPacketSize) {
setErrorMessage("Content-Length > max packet size found!", true);
// reset connection
this->close();
@ -868,7 +844,7 @@ void SimpleHttpClient::processChunkedBody() {
_result->setResultType(SimpleHttpResult::COMPLETE);
_state = FINISHED;
if (!_keepAlive) {
if (!_params._keepAlive) {
_connection->disconnect();
}
@ -882,7 +858,7 @@ void SimpleHttpClient::processChunkedBody() {
_state = FINISHED;
if (!_keepAlive) {
if (!_params._keepAlive) {
_connection->disconnect();
}

296
lib/SimpleHttpClient/SimpleHttpClient.h
View File

@ -1,7 +1,7 @@
////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2014-2017 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
@ -20,6 +20,7 @@
///
/// @author Dr. Frank Celler
/// @author Achim Brandt
/// @author Simon Grätzer
////////////////////////////////////////////////////////////////////////////////
#ifndef ARANGODB_SIMPLE_HTTP_CLIENT_SIMPLE_HTTP_CLIENT_H
@ -37,6 +38,130 @@ namespace httpclient {
class SimpleHttpResult;
class GeneralClientConnection;
struct SimpleHttpClientParams {
friend class SimpleHttpClient;
SimpleHttpClientParams(double requestTimeout, bool warn)
: _requestTimeout(requestTimeout),
_warn(warn),
_locationRewriter({nullptr, nullptr}) {}
//////////////////////////////////////////////////////////////////////////////
/// @brief leave connection open on destruction
//////////////////////////////////////////////////////////////////////////////
void keepConnectionOnDestruction(bool b) { _keepConnectionOnDestruction = b; }
//////////////////////////////////////////////////////////////////////////////
/// @brief enable or disable keep-alive
//////////////////////////////////////////////////////////////////////////////
void setKeepAlive(bool value) { _keepAlive = value; }
//////////////////////////////////////////////////////////////////////////////
/// @brief expose ArangoDB via user-agent?
//////////////////////////////////////////////////////////////////////////////
void setExposeArangoDB(bool value) { _exposeArangoDB = value; }
//////////////////////////////////////////////////////////////////////////////
/// @brief advertise support for deflate?
//////////////////////////////////////////////////////////////////////////////
void setSupportDeflate(bool value) { _supportDeflate = value; }
void setMaxRetries(size_t s) { _maxRetries = s; }
size_t getMaxRetries() { return _maxRetries; }
void setRetryWaitTime(uint64_t wt) { _retryWaitTime = wt; }
uint64_t getRetryWaitTime() { return _retryWaitTime; }
void setRetryMessage(std::string const& m) { _retryMessage = m; }
void setMaxPacketSize(size_t ms) { _maxPacketSize = ms; }
//////////////////////////////////////////////////////////////////////////////
/// @brief sets username and password
///
/// @param prefix prefix for sending username and
/// password
/// @param username username
/// @param password password
//////////////////////////////////////////////////////////////////////////////
void setJwt(std::string const& jwt) { _jwt = jwt; }
////////////////////////////////////////////////////////////////////////////////
/// @brief sets username and password
////////////////////////////////////////////////////////////////////////////////
void setUserNamePassword(std::string const& prefix,
std::string const& username,
std::string const& password) {
std::string value =
arangodb::basics::StringUtils::encodeBase64(username + ":" + password);
_pathToBasicAuth.push_back(std::make_pair(prefix, value));
}
//////////////////////////////////////////////////////////////////////////////
/// @brief allows rewriting locations
//////////////////////////////////////////////////////////////////////////////
void setLocationRewriter(void* data,
std::string (*func)(void*, std::string const&)) {
_locationRewriter.data = data;
_locationRewriter.func = func;
}
//////////////////////////////////////////////////////////////////////////////
/// @brief set the value for max packet size
//////////////////////////////////////////////////////////////////////////////
static void setDefaultMaxPacketSize(size_t value) { MaxPacketSize = value; }
private:
// flag whether or not we keep the connection on destruction
bool _keepConnectionOnDestruction = false;
double _requestTimeout;
bool _warn;
bool _keepAlive = true;
bool _exposeArangoDB = true;
bool _supportDeflate = true;
size_t _maxRetries = 3;
uint64_t _retryWaitTime = 1 * 1000 * 1000;
std::string _retryMessage = "";
size_t _maxPacketSize = SimpleHttpClientParams::MaxPacketSize;
std::vector<std::pair<std::string, std::string>> _pathToBasicAuth;
std::string _jwt = "";
//////////////////////////////////////////////////////////////////////////////
/// @brief struct for rewriting location URLs
//////////////////////////////////////////////////////////////////////////////
struct {
void* data;
std::string (*func)(void*, std::string const&);
} _locationRewriter;
private:
// default value for max packet size
static size_t MaxPacketSize;
};
////////////////////////////////////////////////////////////////////////////////
/// @brief simple http client
////////////////////////////////////////////////////////////////////////////////
@ -63,8 +188,9 @@ class SimpleHttpClient {
};
public:
SimpleHttpClient(std::unique_ptr<GeneralClientConnection>&, double, bool);
SimpleHttpClient(GeneralClientConnection*, double, bool);
SimpleHttpClient(std::unique_ptr<GeneralClientConnection>&,
SimpleHttpClientParams const&);
SimpleHttpClient(GeneralClientConnection*, SimpleHttpClientParams const&);
~SimpleHttpClient();
public:
@ -105,10 +231,17 @@ class SimpleHttpClient {
void close();
//////////////////////////////////////////////////////////////////////////////
/// @brief leave connection open on destruction
/// @brief make an http request, creating a new HttpResult object
/// the caller has to delete the result object
/// this version does not allow specifying custom headers
/// if the request fails because of connection problems, the request will be
/// retried until it either succeeds (at least no connection problem) or there
/// have been _maxRetries retries
//////////////////////////////////////////////////////////////////////////////
void keepConnectionOnDestruction(bool b) { _keepConnectionOnDestruction = b; }
SimpleHttpResult* retryRequest(
rest::RequestType, std::string const&, char const*, size_t,
std::unordered_map<std::string, std::string> const&);
//////////////////////////////////////////////////////////////////////////////
/// @brief make an http request, creating a new HttpResult object
@ -119,21 +252,8 @@ class SimpleHttpClient {
/// have been _maxRetries retries
//////////////////////////////////////////////////////////////////////////////
SimpleHttpResult* retryRequest(rest::RequestType,
std::string const&, char const*, size_t,
std::unordered_map<std::string, std::string> const&);
//////////////////////////////////////////////////////////////////////////////
/// @brief make an http request, creating a new HttpResult object
/// the caller has to delete the result object
/// this version does not allow specifying custom headers
/// if the request fails because of connection problems, the request will be
/// retried until it either succeeds (at least no connection problem) or there
/// have been _maxRetries retries
//////////////////////////////////////////////////////////////////////////////
SimpleHttpResult* retryRequest(rest::RequestType,
std::string const&, char const*, size_t);
SimpleHttpResult* retryRequest(rest::RequestType, std::string const&,
char const*, size_t);
//////////////////////////////////////////////////////////////////////////////
/// @brief make an http request, creating a new HttpResult object
@ -141,8 +261,8 @@ class SimpleHttpClient {
/// this version does not allow specifying custom headers
//////////////////////////////////////////////////////////////////////////////
SimpleHttpResult* request(rest::RequestType,
std::string const&, char const*, size_t);
SimpleHttpResult* request(rest::RequestType, std::string const&, char const*,
size_t);
//////////////////////////////////////////////////////////////////////////////
/// @brief make an http request, actual worker function
@ -150,60 +270,9 @@ class SimpleHttpClient {
/// this version allows specifying custom headers
//////////////////////////////////////////////////////////////////////////////
SimpleHttpResult* request(rest::RequestType,
std::string const&, char const*, size_t,
std::unordered_map<std::string, std::string> const&);
//////////////////////////////////////////////////////////////////////////////
/// @brief sets username and password
///
/// @param prefix prefix for sending username and
/// password
/// @param username username
/// @param password password
//////////////////////////////////////////////////////////////////////////////
void setJwt(std::string const& jwt);
void setUserNamePassword(std::string const& prefix,
std::string const& username,
std::string const& password);
//////////////////////////////////////////////////////////////////////////////
/// @brief allows rewriting locations
//////////////////////////////////////////////////////////////////////////////
void setLocationRewriter(void* data,
std::string (*func)(void*, std::string const&)) {
_locationRewriter.data = data;
_locationRewriter.func = func;
}
//////////////////////////////////////////////////////////////////////////////
/// @brief set the value for max packet size
//////////////////////////////////////////////////////////////////////////////
static void setMaxPacketSize(size_t value) {
MaxPacketSize = value;
}
//////////////////////////////////////////////////////////////////////////////
/// @brief enable or disable keep-alive
//////////////////////////////////////////////////////////////////////////////
void setKeepAlive(bool value) { _keepAlive = value; }
//////////////////////////////////////////////////////////////////////////////
/// @brief expose ArangoDB via user-agent?
//////////////////////////////////////////////////////////////////////////////
void setExposeArangoDB(bool value) { _exposeArangoDB = value; }
//////////////////////////////////////////////////////////////////////////////
/// @brief advertise support for deflate?
//////////////////////////////////////////////////////////////////////////////
void setSupportDeflate(bool value) { _supportDeflate = value; }
SimpleHttpResult* request(
rest::RequestType, std::string const&, char const*, size_t,
std::unordered_map<std::string, std::string> const&);
//////////////////////////////////////////////////////////////////////////////
/// @brief returns the current error message
@ -218,7 +287,7 @@ class SimpleHttpClient {
void setErrorMessage(std::string const& message, bool forceWarn = false) {
_errorMessage = message;
if (_warn || forceWarn) {
if (_params._warn || forceWarn) {
LOG_TOPIC(WARN, arangodb::Logger::FIXME) << "" << _errorMessage;
}
}
@ -251,7 +320,10 @@ class SimpleHttpClient {
/// @brief extract an error message from a response
//////////////////////////////////////////////////////////////////////////////
std::string getHttpErrorMessage(SimpleHttpResult const*, int* errorCode = nullptr);
std::string getHttpErrorMessage(SimpleHttpResult const*,
int* errorCode = nullptr);
SimpleHttpClientParams& params() { return _params; };
private:
//////////////////////////////////////////////////////////////////////////////
@ -260,9 +332,9 @@ class SimpleHttpClient {
/// this version allows specifying custom headers
//////////////////////////////////////////////////////////////////////////////
SimpleHttpResult* doRequest(rest::RequestType,
std::string const&, char const*, size_t,
std::unordered_map<std::string, std::string> const&);
SimpleHttpResult* doRequest(
rest::RequestType, std::string const&, char const*, size_t,
std::unordered_map<std::string, std::string> const&);
//////////////////////////////////////////////////////////////////////////////
/// @brief initialize the connection
@ -281,8 +353,9 @@ class SimpleHttpClient {
//////////////////////////////////////////////////////////////////////////////
std::string rewriteLocation(std::string const& location) {
if (_locationRewriter.func != nullptr) {
return _locationRewriter.func(_locationRewriter.data, location);
if (_params._locationRewriter.func != nullptr) {
return _params._locationRewriter.func(_params._locationRewriter.data,
location);
}
return location;
@ -305,10 +378,10 @@ class SimpleHttpClient {
/// @param headerFields list of header fields
//////////////////////////////////////////////////////////////////////////////
void setRequest(rest::RequestType method,
std::string const& location, char const* body,
size_t bodyLength,
std::unordered_map<std::string, std::string> const& headerFields);
void setRequest(
rest::RequestType method, std::string const& location, char const* body,
size_t bodyLength,
std::unordered_map<std::string, std::string> const& headerFields);
//////////////////////////////////////////////////////////////////////////////
/// @brief process (a part of) the http header, the data is
@ -360,6 +433,14 @@ class SimpleHttpClient {
GeneralClientConnection* _connection;
// flag whether or not to delete the connection on destruction
bool _deleteConnectionOnDestruction = false;
//////////////////////////////////////////////////////////////////////////////
/// @brief connection parameters
//////////////////////////////////////////////////////////////////////////////
SimpleHttpClientParams _params;
//////////////////////////////////////////////////////////////////////////////
/// @brief write buffer
//////////////////////////////////////////////////////////////////////////////
@ -393,61 +474,20 @@ class SimpleHttpClient {
size_t _readBufferOffset;
double _requestTimeout;
request_state _state;
size_t _written;
std::string _errorMessage;
//////////////////////////////////////////////////////////////////////////////
/// @brief struct for rewriting location URLs
//////////////////////////////////////////////////////////////////////////////
struct {
void* data;
std::string (*func)(void*, std::string const&);
} _locationRewriter;
uint32_t _nextChunkedSize;
rest::RequestType _method;
SimpleHttpResult* _result;
std::vector<std::pair<std::string, std::string>> _pathToBasicAuth;
std::string _jwt;
size_t _maxPacketSize;
public:
size_t _maxRetries;
uint64_t _retryWaitTime;
std::string _retryMessage;
private:
// flag whether or not to delete the connection on destruction
bool _deleteConnectionOnDestruction;
// flag whether or not we keep the connection on destruction
bool _keepConnectionOnDestruction;
bool _warn;
bool _keepAlive;
bool _exposeArangoDB;
bool _supportDeflate;
// empty map, used for headers
static std::unordered_map<std::string, std::string> const NO_HEADERS;
// default value for max packet size
static size_t MaxPacketSize;
};
}
}

11
lib/V8/v8-utils.cpp
View File

@ -854,12 +854,13 @@ void JS_Download(v8::FunctionCallbackInfo<v8::Value> const& args) {
if (connection == nullptr) {
TRI_V8_THROW_EXCEPTION_MEMORY();
}
SimpleHttpClient client(connection.get(), timeout, false);
client.setSupportDeflate(false);
SimpleHttpClientParams params(timeout, false);
params.setSupportDeflate(false);
// security by obscurity won't work. Github requires a useragent nowadays.
client.setExposeArangoDB(true);
params.setExposeArangoDB(true);
SimpleHttpClient client(connection.get(), params);
v8::Handle<v8::Object> result = v8::Object::New(isolate);
if (numRedirects > 0) {