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

include replication tests

This commit is contained in:
Jan Steemann 2013-07-23 11:19:41 +02:00
parent f193015b5f
commit f49ae77bc5
10 changed files with 562 additions and 390 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
UnitTests/Makefile.unittests
View File

@ -269,7 +269,7 @@ unittests-boost:
@echo "================================================================================"
@echo
@echo "to enable unit-testing, configure with --enable-maintainer-mode"
@echo "to enable boost unit-tests, install Boost test and configure with --enable-maintainer-mode"
@echo
endif
@ -279,7 +279,7 @@ endif
################################################################################
SHELL_COMMON = \
@top_srcdir@/js/common/tests/shell-require.js \
@top_srcdir@/js/common/tests/shell-require.js \
@top_srcdir@/js/common/tests/shell-transactions.js \
@top_srcdir@/js/common/tests/shell-aqlfunctions.js \
@top_srcdir@/js/common/tests/shell-attributes.js \
@ -304,6 +304,13 @@ SHELL_COMMON = \
@top_srcdir@/js/common/tests/shell-fulltext.js \
@top_srcdir@/js/common/tests/shell-graph.js
if ENABLE_REPLICATION
SHELL_COMMON += \
@top_srcdir@/js/common/tests/shell-replication.js
endif
################################################################################
### @brief SHELL SERVER TESTS (BASICS)
################################################################################

118
arangod/Replication/ReplicationFetcher.cpp
View File

@ -221,7 +221,7 @@ int ReplicationFetcher::run () {
TRI_SetErrorReplicationApplier(_applier, res, errorMsg.c_str());
// stop ourselves
TRI_StopReplicationApplier(_applier);
TRI_StopReplicationApplier(_applier, false);
return res;
}
@ -1264,42 +1264,39 @@ int ReplicationFetcher::getMasterState (string& errorMsg) {
0,
headers);
if (response == 0) {
errorMsg = "could not connect to master at " + string(_masterInfo._endpoint);
if (response == 0 || ! response->isComplete()) {
errorMsg = "could not connect to master at " + string(_masterInfo._endpoint) +
": " + _client->getErrorMessage();
if (response != 0) {
delete response;
}
return TRI_ERROR_REPLICATION_NO_RESPONSE;
}
int res = TRI_ERROR_NO_ERROR;
if (! response->isComplete()) {
res = TRI_ERROR_REPLICATION_NO_RESPONSE;
if (response->wasHttpError()) {
res = TRI_ERROR_REPLICATION_MASTER_ERROR;
errorMsg = "got invalid response from master at " + string(_masterInfo._endpoint) +
": " + _client->getErrorMessage();
": HTTP " + StringUtils::itoa(response->getHttpReturnCode()) +
": " + response->getHttpReturnMessage();
}
else {
if (response->wasHttpError()) {
res = TRI_ERROR_REPLICATION_MASTER_ERROR;
errorMsg = "got invalid response from master at " + string(_masterInfo._endpoint) +
": HTTP " + StringUtils::itoa(response->getHttpReturnCode()) +
": " + response->getHttpReturnMessage();
TRI_json_t* json = TRI_JsonString(TRI_UNKNOWN_MEM_ZONE, response->getBody().str().c_str());
if (JsonHelper::isArray(json)) {
res = handleStateResponse(json, errorMsg);
TRI_FreeJson(TRI_UNKNOWN_MEM_ZONE, json);
}
else {
TRI_json_t* json = TRI_JsonString(TRI_UNKNOWN_MEM_ZONE, response->getBody().str().c_str());
res = TRI_ERROR_REPLICATION_INVALID_RESPONSE;
if (JsonHelper::isArray(json)) {
res = handleStateResponse(json, errorMsg);
TRI_FreeJson(TRI_UNKNOWN_MEM_ZONE, json);
}
else {
res = TRI_ERROR_REPLICATION_INVALID_RESPONSE;
errorMsg = "got invalid response from master at " + string(_masterInfo._endpoint) +
": invalid JSON";
}
errorMsg = "got invalid response from master at " + string(_masterInfo._endpoint) +
": invalid JSON";
}
}
@ -1328,42 +1325,39 @@ int ReplicationFetcher::performInitialSync (string& errorMsg) {
0,
headers);
if (response == 0) {
errorMsg = "could not connect to master at " + string(_masterInfo._endpoint);
if (response == 0 || ! response->isComplete()) {
errorMsg = "could not connect to master at " + string(_masterInfo._endpoint) +
": " + _client->getErrorMessage();
if (response != 0) {
delete response;
}
return TRI_ERROR_REPLICATION_NO_RESPONSE;
}
int res = TRI_ERROR_NO_ERROR;
if (! response->isComplete()) {
res = TRI_ERROR_REPLICATION_NO_RESPONSE;
if (response->wasHttpError()) {
res = TRI_ERROR_REPLICATION_MASTER_ERROR;
errorMsg = "got invalid response from master at " + string(_masterInfo._endpoint) +
": " + _client->getErrorMessage();
": HTTP " + StringUtils::itoa(response->getHttpReturnCode()) +
": " + response->getHttpReturnMessage();
}
else {
if (response->wasHttpError()) {
res = TRI_ERROR_REPLICATION_MASTER_ERROR;
errorMsg = "got invalid response from master at " + string(_masterInfo._endpoint) +
": HTTP " + StringUtils::itoa(response->getHttpReturnCode()) +
": " + response->getHttpReturnMessage();
TRI_json_t* json = TRI_JsonString(TRI_UNKNOWN_MEM_ZONE, response->getBody().str().c_str());
if (JsonHelper::isArray(json)) {
res = handleInventoryResponse(json, errorMsg);
TRI_FreeJson(TRI_UNKNOWN_MEM_ZONE, json);
}
else {
TRI_json_t* json = TRI_JsonString(TRI_UNKNOWN_MEM_ZONE, response->getBody().str().c_str());
res = TRI_ERROR_REPLICATION_INVALID_RESPONSE;
if (JsonHelper::isArray(json)) {
res = handleInventoryResponse(json, errorMsg);
TRI_FreeJson(TRI_UNKNOWN_MEM_ZONE, json);
}
else {
res = TRI_ERROR_REPLICATION_INVALID_RESPONSE;
errorMsg = "got invalid response from master at " + string(_masterInfo._endpoint) +
": invalid JSON";
}
errorMsg = "got invalid response from master at " + string(_masterInfo._endpoint) +
": invalid JSON";
}
}
@ -1487,17 +1481,13 @@ int ReplicationFetcher::handleCollectionDump (TRI_transaction_collection_t* trxC
0,
headers);
if (response == 0) {
errorMsg = "could not connect to master at " + string(_masterInfo._endpoint);
return TRI_ERROR_REPLICATION_NO_RESPONSE;
}
if (! response->isComplete()) {
errorMsg = "got invalid response from master at " + string(_masterInfo._endpoint) +
if (response == 0 || ! response->isComplete()) {
errorMsg = "could not connect to master at " + string(_masterInfo._endpoint) +
": " + _client->getErrorMessage();
delete response;
if (response != 0) {
delete response;
}
return TRI_ERROR_REPLICATION_NO_RESPONSE;
}
@ -2040,17 +2030,13 @@ int ReplicationFetcher::followMasterLog (string& errorMsg,
0,
headers);
if (response == 0) {
errorMsg = "could not connect to master at " + string(_masterInfo._endpoint);
return TRI_ERROR_REPLICATION_NO_RESPONSE;
}
if (! response->isComplete()) {
if (response == 0 || ! response->isComplete()) {
errorMsg = "got invalid response from master at " + string(_masterInfo._endpoint) +
": " + _client->getErrorMessage();
delete response;
if (response != 0) {
delete response;
}
return TRI_ERROR_REPLICATION_NO_RESPONSE;
}

65
arangod/RestHandler/RestReplicationHandler.cpp
View File

@ -157,10 +157,15 @@ Handler::status_e RestReplicationHandler::execute() {
handleCommandDump();
}
else if (command == "apply-config") {
if (type != HttpRequest::HTTP_REQUEST_PUT) {
goto BAD_CALL;
if (type == HttpRequest::HTTP_REQUEST_GET) {
handleCommandApplyGetConfig();
}
else {
if (type != HttpRequest::HTTP_REQUEST_PUT) {
goto BAD_CALL;
}
handleCommandApplySetConfig();
}
handleCommandApplyConfig();
}
else if (command == "apply-start") {
if (type != HttpRequest::HTTP_REQUEST_PUT) {
@ -618,36 +623,64 @@ void RestReplicationHandler::handleCommandDump () {
TRI_FreeStringBuffer(TRI_CORE_MEM_ZONE, dump._buffer);
}
////////////////////////////////////////////////////////////////////////////////
/// @brief return the configuration of the replication applier
////////////////////////////////////////////////////////////////////////////////
void RestReplicationHandler::handleCommandApplyGetConfig () {
assert(_vocbase->_replicationApplier != 0);
TRI_replication_apply_configuration_t config;
TRI_InitApplyConfigurationReplicationApplier(&config);
TRI_ReadLockReadWriteLock(&_vocbase->_replicationApplier->_statusLock);
TRI_CopyApplyConfigurationReplicationApplier(&_vocbase->_replicationApplier->_configuration, &config);
TRI_ReadUnlockReadWriteLock(&_vocbase->_replicationApplier->_statusLock);
TRI_json_t* result = TRI_JsonApplyConfigurationReplicationApplier(&config);
TRI_DestroyApplyConfigurationReplicationApplier(&config);
if (result == 0) {
generateError(HttpResponse::SERVER_ERROR, TRI_ERROR_OUT_OF_MEMORY);
}
else {
generateResult(result);
TRI_FreeJson(TRI_CORE_MEM_ZONE, result);
}
}
////////////////////////////////////////////////////////////////////////////////
/// @brief configure the replication applier
////////////////////////////////////////////////////////////////////////////////
void RestReplicationHandler::handleCommandApplyConfig () {
void RestReplicationHandler::handleCommandApplySetConfig () {
assert(_vocbase->_replicationApplier != 0);
TRI_replication_apply_configuration_t configuration;
TRI_InitApplyConfigurationReplicationApplier(&configuration);
TRI_replication_apply_configuration_t config;
TRI_InitApplyConfigurationReplicationApplier(&config);
TRI_json_t* json = parseJsonBody();
if (json == 0) {
return;
}
const string endpoint = JsonHelper::getStringValue(json, "endpoint", "");
configuration._endpoint = TRI_DuplicateString2Z(TRI_CORE_MEM_ZONE, endpoint.c_str(), endpoint.size());
configuration._requestTimeout = JsonHelper::getDoubleValue(json, "requestTimeout", configuration._requestTimeout);
configuration._connectTimeout = JsonHelper::getDoubleValue(json, "connectTimeout", configuration._connectTimeout);
configuration._ignoreErrors = JsonHelper::getUInt64Value(json, "ignoreErrors", configuration._ignoreErrors);
configuration._maxConnectRetries = JsonHelper::getIntValue(json, "maxConnectRetries", configuration._maxConnectRetries);
configuration._autoStart = JsonHelper::getBooleanValue(json, "autoStart", configuration._autoStart);
configuration._adaptivePolling = JsonHelper::getBooleanValue(json, "adaptivePolling", configuration._adaptivePolling);
config._endpoint = TRI_DuplicateString2Z(TRI_CORE_MEM_ZONE, endpoint.c_str(), endpoint.size());
config._requestTimeout = JsonHelper::getDoubleValue(json, "requestTimeout", config._requestTimeout);
config._connectTimeout = JsonHelper::getDoubleValue(json, "connectTimeout", config._connectTimeout);
config._ignoreErrors = JsonHelper::getUInt64Value(json, "ignoreErrors", config._ignoreErrors);
config._maxConnectRetries = JsonHelper::getIntValue(json, "maxConnectRetries", config._maxConnectRetries);
config._autoStart = JsonHelper::getBooleanValue(json, "autoStart", config._autoStart);
config._adaptivePolling = JsonHelper::getBooleanValue(json, "adaptivePolling", config._adaptivePolling);
TRI_Free(TRI_UNKNOWN_MEM_ZONE, json);
int res = TRI_ConfigureReplicationApplier(_vocbase->_replicationApplier, &configuration);
int res = TRI_ConfigureReplicationApplier(_vocbase->_replicationApplier, &config);
TRI_DestroyApplyConfigurationReplicationApplier(&configuration);
TRI_DestroyApplyConfigurationReplicationApplier(&config);
if (res != TRI_ERROR_NO_ERROR) {
generateError(HttpResponse::SERVER_ERROR, res);
@ -692,7 +725,7 @@ void RestReplicationHandler::handleCommandApplyStart () {
void RestReplicationHandler::handleCommandApplyStop () {
assert(_vocbase->_replicationApplier != 0);
int res = TRI_StopReplicationApplier(_vocbase->_replicationApplier);
int res = TRI_StopReplicationApplier(_vocbase->_replicationApplier, true);
if (res != TRI_ERROR_NO_ERROR) {
generateError(HttpResponse::SERVER_ERROR, res);

8
arangod/RestHandler/RestReplicationHandler.h
View File

@ -195,11 +195,17 @@ namespace triagens {
void handleCommandDump ();
////////////////////////////////////////////////////////////////////////////////
/// @brief return the configuration of the the replication applier
////////////////////////////////////////////////////////////////////////////////
void handleCommandApplyGetConfig ();
////////////////////////////////////////////////////////////////////////////////
/// @brief configure the replication applier
////////////////////////////////////////////////////////////////////////////////
void handleCommandApplyConfig ();
void handleCommandApplySetConfig ();
////////////////////////////////////////////////////////////////////////////////
/// @brief start the replication applier

148
arangod/V8Server/v8-vocbase.cpp
View File

@ -3128,80 +3128,104 @@ static v8::Handle<v8::Value> JS_ConfigureApplierReplication (v8::Arguments const
TRI_vocbase_t* vocbase = GetContextVocBase();
if (argv.Length() != 1 || ! argv[0]->IsObject()) {
TRI_V8_EXCEPTION_USAGE(scope, "REPLICATION_APPLIER_CONFIGURE(<configuration>)");
}
if (vocbase == 0 || vocbase->_replicationApplier == 0) {
TRI_V8_EXCEPTION_INTERNAL(scope, "cannot extract vocbase");
}
TRI_replication_apply_configuration_t configuration;
if (argv.Length() == 0) {
// no argument: return the current configuration
TRI_replication_apply_configuration_t config;
TRI_InitApplyConfigurationReplicationApplier(&config);
TRI_ReadLockReadWriteLock(&vocbase->_replicationApplier->_statusLock);
TRI_CopyApplyConfigurationReplicationApplier(&vocbase->_replicationApplier->_configuration, &config);
TRI_ReadUnlockReadWriteLock(&vocbase->_replicationApplier->_statusLock);
TRI_json_t* json = TRI_JsonApplyConfigurationReplicationApplier(&config);
TRI_DestroyApplyConfigurationReplicationApplier(&config);
TRI_InitApplyConfigurationReplicationApplier(&configuration);
v8::Handle<v8::Value> result = TRI_ObjectJson(json);
TRI_FreeJson(TRI_CORE_MEM_ZONE, json);
// treat the argument as an object from now on
v8::Handle<v8::Object> object = v8::Handle<v8::Object>::Cast(argv[0]);
string endpoint = "";
if (object->Has(TRI_V8_SYMBOL("endpoint"))) {
if (object->Get(TRI_V8_SYMBOL("endpoint"))->IsString()) {
endpoint = TRI_ObjectToString(object->Get(TRI_V8_SYMBOL("endpoint")));
}
return scope.Close(result);
}
configuration._endpoint = TRI_DuplicateString2Z(TRI_CORE_MEM_ZONE, endpoint.c_str(), endpoint.size());
if (object->Has(TRI_V8_SYMBOL("requestTimeout"))) {
if (object->Get(TRI_V8_SYMBOL("requestTimeout"))->IsNumber()) {
configuration._requestTimeout = TRI_ObjectToDouble(object->Get(TRI_V8_SYMBOL("requestTimeout")));
}
}
if (object->Has(TRI_V8_SYMBOL("connectTimeout"))) {
if (object->Get(TRI_V8_SYMBOL("connectTimeout"))->IsNumber()) {
configuration._connectTimeout = TRI_ObjectToDouble(object->Get(TRI_V8_SYMBOL("connectTimeout")));
}
}
if (object->Has(TRI_V8_SYMBOL("ignoreErrors"))) {
if (object->Get(TRI_V8_SYMBOL("ignoreErrors"))->IsNumber()) {
configuration._ignoreErrors = TRI_ObjectToUInt64(object->Get(TRI_V8_SYMBOL("ignoreErrors")), false);
}
}
if (object->Has(TRI_V8_SYMBOL("maxConnectRetries"))) {
if (object->Get(TRI_V8_SYMBOL("maxConnectRetries"))->IsNumber()) {
configuration._maxConnectRetries = (int) TRI_ObjectToInt64(object->Get(TRI_V8_SYMBOL("maxConnectRetries")));
}
}
if (object->Has(TRI_V8_SYMBOL("autoStart"))) {
if (object->Get(TRI_V8_SYMBOL("autoStart"))->IsBoolean()) {
configuration._autoStart = TRI_ObjectToBoolean(object->Get(TRI_V8_SYMBOL("autoStart")));
}
}
if (object->Has(TRI_V8_SYMBOL("adaptivePolling"))) {
if (object->Get(TRI_V8_SYMBOL("adaptivePolling"))->IsBoolean()) {
configuration._autoStart = TRI_ObjectToBoolean(object->Get(TRI_V8_SYMBOL("adaptivePolling")));
}
}
else {
// set the configuration
int res = TRI_ConfigureReplicationApplier(vocbase->_replicationApplier, &configuration);
TRI_DestroyApplyConfigurationReplicationApplier(&configuration);
if (res != TRI_ERROR_NO_ERROR) {
TRI_V8_EXCEPTION(scope, res);
if (argv.Length() != 1 || ! argv[0]->IsObject()) {
TRI_V8_EXCEPTION_USAGE(scope, "REPLICATION_APPLIER_CONFIGURE(<configuration>)");
}
TRI_replication_apply_configuration_t config;
TRI_InitApplyConfigurationReplicationApplier(&config);
// treat the argument as an object from now on
v8::Handle<v8::Object> object = v8::Handle<v8::Object>::Cast(argv[0]);
string endpoint = "";
if (object->Has(TRI_V8_SYMBOL("endpoint"))) {
if (object->Get(TRI_V8_SYMBOL("endpoint"))->IsString()) {
endpoint = TRI_ObjectToString(object->Get(TRI_V8_SYMBOL("endpoint")));
}
}
config._endpoint = TRI_DuplicateString2Z(TRI_CORE_MEM_ZONE, endpoint.c_str(), endpoint.size());
if (object->Has(TRI_V8_SYMBOL("requestTimeout"))) {
if (object->Get(TRI_V8_SYMBOL("requestTimeout"))->IsNumber()) {
config._requestTimeout = TRI_ObjectToDouble(object->Get(TRI_V8_SYMBOL("requestTimeout")));
}
}
if (object->Has(TRI_V8_SYMBOL("connectTimeout"))) {
if (object->Get(TRI_V8_SYMBOL("connectTimeout"))->IsNumber()) {
config._connectTimeout = TRI_ObjectToDouble(object->Get(TRI_V8_SYMBOL("connectTimeout")));
}
}
if (object->Has(TRI_V8_SYMBOL("ignoreErrors"))) {
if (object->Get(TRI_V8_SYMBOL("ignoreErrors"))->IsNumber()) {
config._ignoreErrors = TRI_ObjectToUInt64(object->Get(TRI_V8_SYMBOL("ignoreErrors")), false);
}
}
if (object->Has(TRI_V8_SYMBOL("maxConnectRetries"))) {
if (object->Get(TRI_V8_SYMBOL("maxConnectRetries"))->IsNumber()) {
config._maxConnectRetries = (int) TRI_ObjectToInt64(object->Get(TRI_V8_SYMBOL("maxConnectRetries")));
}
}
if (object->Has(TRI_V8_SYMBOL("autoStart"))) {
if (object->Get(TRI_V8_SYMBOL("autoStart"))->IsBoolean()) {
config._autoStart = TRI_ObjectToBoolean(object->Get(TRI_V8_SYMBOL("autoStart")));
}
}
if (object->Has(TRI_V8_SYMBOL("adaptivePolling"))) {
if (object->Get(TRI_V8_SYMBOL("adaptivePolling"))->IsBoolean()) {
config._autoStart = TRI_ObjectToBoolean(object->Get(TRI_V8_SYMBOL("adaptivePolling")));
}
}
int res = TRI_ConfigureReplicationApplier(vocbase->_replicationApplier, &config);
TRI_DestroyApplyConfigurationReplicationApplier(&config);
if (res != TRI_ERROR_NO_ERROR) {
TRI_V8_EXCEPTION(scope, res);
}
return scope.Close(v8::True());
}
return scope.Close(v8::True());
}
#endif
////////////////////////////////////////////////////////////////////////////////
/// @brief start the replication applier manually
////////////////////////////////////////////////////////////////////////////////
@ -3256,7 +3280,7 @@ static v8::Handle<v8::Value> JS_StopApplierReplication (v8::Arguments const& arg
TRI_V8_EXCEPTION_INTERNAL(scope, "cannot extract vocbase");
}
int res = TRI_StopReplicationApplier(vocbase->_replicationApplier);
int res = TRI_StopReplicationApplier(vocbase->_replicationApplier, true);
if (res != TRI_ERROR_NO_ERROR) {
TRI_V8_EXCEPTION_MESSAGE(scope, res, "cannot stop replication applier");

67
arangod/VocBase/replication-applier.c
View File

@ -142,12 +142,13 @@ static char* GetApplyConfigurationFilename (TRI_vocbase_t* vocbase) {
/// @brief get a JSON representation of the replication apply configuration
////////////////////////////////////////////////////////////////////////////////
static TRI_json_t* JsonApplyConfiguration (TRI_replication_apply_configuration_t const* config) {
static TRI_json_t* JsonApplyConfiguration (TRI_replication_apply_configuration_t const* config,
bool includePassword) {
TRI_json_t* json;
assert(config->_endpoint != NULL);
json = TRI_CreateArray2Json(TRI_CORE_MEM_ZONE, 8);
json = TRI_CreateArray2Json(TRI_CORE_MEM_ZONE, 9);
if (json == NULL) {
return NULL;
@ -164,6 +165,13 @@ static TRI_json_t* JsonApplyConfiguration (TRI_replication_apply_configuration_t
"username",
TRI_CreateStringCopyJson(TRI_CORE_MEM_ZONE, config->_username));
}
if (config->_password != NULL && includePassword) {
TRI_Insert3ArrayJson(TRI_CORE_MEM_ZONE,
json,
"password",
TRI_CreateStringCopyJson(TRI_CORE_MEM_ZONE, config->_password));
}
TRI_Insert3ArrayJson(TRI_CORE_MEM_ZONE,
json,
@ -367,7 +375,8 @@ static int StartApplier (TRI_replication_applier_t* applier,
/// note: must hold the lock when calling this
////////////////////////////////////////////////////////////////////////////////
static int StopApplier (TRI_replication_applier_t* applier) {
static int StopApplier (TRI_replication_applier_t* applier,
bool resetError) {
TRI_replication_apply_state_t* state;
state = &applier->_state;
@ -380,17 +389,21 @@ static int StopApplier (TRI_replication_applier_t* applier) {
SetTerminateFlag(applier, true);
state->_phase = PHASE_NONE;
state->_phase = PHASE_NONE;
TRI_SetProgressReplicationApplier(applier, "applier stopped", false);
if (state->_lastError._msg != NULL) {
TRI_FreeString(TRI_CORE_MEM_ZONE, state->_lastError._msg);
state->_lastError._msg = NULL;
}
if (resetError) {
if (state->_lastError._msg != NULL) {
TRI_FreeString(TRI_CORE_MEM_ZONE, state->_lastError._msg);
state->_lastError._msg = NULL;
}
state->_lastError._code = TRI_ERROR_NO_ERROR;
TRI_GetTimeStampReplication(state->_lastError._time, sizeof(state->_lastError._time) - 1);
state->_lastError._code = TRI_ERROR_NO_ERROR;
TRI_GetTimeStampReplication(state->_lastError._time, sizeof(state->_lastError._time) - 1);
}
TRI_LockCondition(&applier->_runStateChangeCondition);
TRI_SignalCondition(&applier->_runStateChangeCondition);
@ -472,7 +485,7 @@ TRI_replication_applier_t* TRI_CreateReplicationApplier (TRI_vocbase_t* vocbase)
////////////////////////////////////////////////////////////////////////////////
void TRI_DestroyReplicationApplier (TRI_replication_applier_t* applier) {
TRI_StopReplicationApplier(applier);
TRI_StopReplicationApplier(applier, true);
TRI_DestroyApplyStateReplicationApplier(&applier->_state);
TRI_DestroyApplyConfigurationReplicationApplier(&applier->_configuration);
@ -503,6 +516,10 @@ void TRI_FreeReplicationApplier (TRI_replication_applier_t* applier) {
/// @addtogroup Replication
/// @{
////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////
/// @brief checks whether the apply thread should terminate
////////////////////////////////////////////////////////////////////////////////
bool TRI_WaitReplicationApplier (TRI_replication_applier_t* applier,
uint64_t sleepTime) {
@ -525,6 +542,14 @@ bool TRI_WaitReplicationApplier (TRI_replication_applier_t* applier,
return true;
}
////////////////////////////////////////////////////////////////////////////////
/// @brief get a JSON representation of the replication apply configuration
////////////////////////////////////////////////////////////////////////////////
TRI_json_t* TRI_JsonApplyConfigurationReplicationApplier (TRI_replication_apply_configuration_t const* config) {
return JsonApplyConfiguration(config, false);
}
////////////////////////////////////////////////////////////////////////////////
/// @brief start the replication applier
////////////////////////////////////////////////////////////////////////////////
@ -555,7 +580,8 @@ int TRI_StartReplicationApplier (TRI_replication_applier_t* applier,
/// @brief stop the replication applier
////////////////////////////////////////////////////////////////////////////////
int TRI_StopReplicationApplier (TRI_replication_applier_t* applier) {
int TRI_StopReplicationApplier (TRI_replication_applier_t* applier,
bool resetError) {
int res;
res = TRI_ERROR_NO_ERROR;
@ -570,7 +596,7 @@ int TRI_StopReplicationApplier (TRI_replication_applier_t* applier) {
return res;
}
res = StopApplier(applier);
res = StopApplier(applier, resetError);
TRI_WriteUnlockReadWriteLock(&applier->_statusLock);
// join the thread without the status lock (otherwise it would propbably not join)
@ -792,7 +818,6 @@ void TRI_SetProgressReplicationApplier (TRI_replication_applier_t* applier,
char const* msg,
bool lock) {
char* copy;
char timeString[24];
copy = TRI_DuplicateStringZ(TRI_CORE_MEM_ZONE, msg);
@ -800,8 +825,6 @@ void TRI_SetProgressReplicationApplier (TRI_replication_applier_t* applier,
return;
}
TRI_GetTimeStampReplication(timeString, sizeof(timeString) - 1);
if (lock) {
TRI_WriteLockReadWriteLock(&applier->_statusLock);
}
@ -813,7 +836,7 @@ void TRI_SetProgressReplicationApplier (TRI_replication_applier_t* applier,
applier->_state._progressMsg = copy;
// write time in buffer
memcpy(&applier->_state._progressTime, &timeString, sizeof(applier->_state._progressTime));
TRI_GetTimeStampReplication(applier->_state._progressTime, sizeof(applier->_state._progressTime) - 1);
if (lock) {
TRI_WriteUnlockReadWriteLock(&applier->_statusLock);
@ -1019,21 +1042,21 @@ void TRI_DestroyApplyConfigurationReplicationApplier (TRI_replication_apply_conf
void TRI_CopyApplyConfigurationReplicationApplier (TRI_replication_apply_configuration_t const* src,
TRI_replication_apply_configuration_t* dst) {
if (src->_endpoint != NULL) {
dst->_endpoint = TRI_DuplicateStringZ(TRI_CORE_MEM_ZONE, src->_endpoint);
dst->_endpoint = TRI_DuplicateStringZ(TRI_CORE_MEM_ZONE, src->_endpoint);
}
else {
dst->_endpoint = NULL;
}
if (src->_username != NULL) {
dst->_username = TRI_DuplicateStringZ(TRI_CORE_MEM_ZONE, src->_username);
dst->_username = TRI_DuplicateStringZ(TRI_CORE_MEM_ZONE, src->_username);
}
else {
dst->_username = NULL;
}
if (src->_password != NULL) {
dst->_password = TRI_DuplicateStringZ(TRI_CORE_MEM_ZONE, src->_password);
dst->_password = TRI_DuplicateStringZ(TRI_CORE_MEM_ZONE, src->_password);
}
else {
dst->_password = NULL;
@ -1084,7 +1107,7 @@ int TRI_SaveConfigurationFileReplicationApplier (TRI_vocbase_t* vocbase,
char* filename;
int res;
json = JsonApplyConfiguration(config);
json = JsonApplyConfiguration(config, true);
if (json == NULL) {
return TRI_ERROR_OUT_OF_MEMORY;

9
arangod/VocBase/replication-applier.h
View File

@ -190,6 +190,12 @@ void TRI_FreeReplicationApplier (TRI_replication_applier_t*);
bool TRI_WaitReplicationApplier (TRI_replication_applier_t*,
uint64_t);
////////////////////////////////////////////////////////////////////////////////
/// @brief get a JSON representation of the replication apply configuration
////////////////////////////////////////////////////////////////////////////////
struct TRI_json_s* TRI_JsonApplyConfigurationReplicationApplier (TRI_replication_apply_configuration_t const*);
////////////////////////////////////////////////////////////////////////////////
/// @brief start the replication applier
////////////////////////////////////////////////////////////////////////////////
@ -201,7 +207,8 @@ int TRI_StartReplicationApplier (TRI_replication_applier_t*,
/// @brief stop the replication applier
////////////////////////////////////////////////////////////////////////////////
int TRI_StopReplicationApplier (TRI_replication_applier_t*);
int TRI_StopReplicationApplier (TRI_replication_applier_t*,
bool);
////////////////////////////////////////////////////////////////////////////////
/// @brief configure the replication applier

74
js/client/modules/org/arangodb/replication.js
View File

@ -36,7 +36,7 @@ var arangosh = require("org/arangodb/arangosh");
// -----------------------------------------------------------------------------
// -----------------------------------------------------------------------------
// --SECTION-- public functions
// --SECTION-- private functions
// -----------------------------------------------------------------------------
////////////////////////////////////////////////////////////////////////////////
@ -44,64 +44,53 @@ var arangosh = require("org/arangodb/arangosh");
/// @{
////////////////////////////////////////////////////////////////////////////////
var logger = { };
var applier = { };
////////////////////////////////////////////////////////////////////////////////
/// @brief starts the replication logger
////////////////////////////////////////////////////////////////////////////////
exports.startLogger = function () {
logger.start = function () {
var db = internal.db;
var requestResult = db._connection.PUT("_api/replication/log-start", "");
arangosh.checkRequestResult(requestResult);
return requestResult;
return true;
};
////////////////////////////////////////////////////////////////////////////////
/// @brief stops the replication logger
////////////////////////////////////////////////////////////////////////////////
exports.stopLogger = function () {
logger.stop = function () {
var db = internal.db;
var requestResult = db._connection.PUT("_api/replication/log-stop", "");
arangosh.checkRequestResult(requestResult);
return requestResult;
return true;
};
////////////////////////////////////////////////////////////////////////////////
/// @brief return the replication logger state
////////////////////////////////////////////////////////////////////////////////
exports.getLoggerState = function () {
logger.state = function () {
var db = internal.db;
var requestResult = db._connection.GET("_api/replication/log-state");
arangosh.checkRequestResult(requestResult);
return requestResult;
};
////////////////////////////////////////////////////////////////////////////////
/// @brief configures the replication applier
////////////////////////////////////////////////////////////////////////////////
exports.configureApplier = function (config) {
var db = internal.db;
var requestResult = db._connection.PUT("_api/replication/apply-config",
JSON.stringify(config));
arangosh.checkRequestResult(requestResult);
return requestResult;
return requestResult.state;
};
////////////////////////////////////////////////////////////////////////////////
/// @brief starts the replication applier
////////////////////////////////////////////////////////////////////////////////
exports.startApplier = function (forceFullSynchronisation) {
applier.start = function (forceFullSynchronisation) {
var db = internal.db;
var append = (forceFullSynchronisation ? "?fullSync=true" : "");
@ -115,7 +104,7 @@ exports.startApplier = function (forceFullSynchronisation) {
/// @brief stops the replication applier
////////////////////////////////////////////////////////////////////////////////
exports.stopApplier = function () {
applier.stop = function () {
var db = internal.db;
var requestResult = db._connection.PUT("_api/replication/apply-stop", "");
@ -128,7 +117,7 @@ exports.stopApplier = function () {
/// @brief return the replication applier state
////////////////////////////////////////////////////////////////////////////////
exports.getApplierState = function () {
applier.state = function () {
var db = internal.db;
var requestResult = db._connection.GET("_api/replication/apply-state");
@ -137,6 +126,43 @@ exports.getApplierState = function () {
return requestResult;
};
////////////////////////////////////////////////////////////////////////////////
/// @brief configures the replication applier
////////////////////////////////////////////////////////////////////////////////
applier.properties = function (config) {
var db = internal.db;
var requestResult;
if (config === undefined) {
requestResult = db._connection.GET("_api/replication/apply-config");
}
else {
requestResult = db._connection.PUT("_api/replication/apply-config",
JSON.stringify(config));
}
arangosh.checkRequestResult(requestResult);
return requestResult;
};
////////////////////////////////////////////////////////////////////////////////
/// @}
////////////////////////////////////////////////////////////////////////////////
// -----------------------------------------------------------------------------
// --SECTION-- module exports
// -----------------------------------------------------------------------------
////////////////////////////////////////////////////////////////////////////////
/// @addtogroup ArangoShell
/// @{
////////////////////////////////////////////////////////////////////////////////
exports.logger = logger;
exports.applier = applier;
////////////////////////////////////////////////////////////////////////////////
/// @}
////////////////////////////////////////////////////////////////////////////////

389
js/common/tests/shell-replication.js
View File

File diff suppressed because it is too large Load Diff

63
js/server/modules/org/arangodb/replication.js
View File

@ -35,7 +35,7 @@ var internal = require("internal");
// -----------------------------------------------------------------------------
// -----------------------------------------------------------------------------
// --SECTION-- public functions
// --SECTION-- private functions
// -----------------------------------------------------------------------------
////////////////////////////////////////////////////////////////////////////////
@ -43,51 +43,46 @@ var internal = require("internal");
/// @{
////////////////////////////////////////////////////////////////////////////////
var logger = { };
var applier = { };
////////////////////////////////////////////////////////////////////////////////
/// @brief starts the replication logger
////////////////////////////////////////////////////////////////////////////////
exports.startLogger = function () {
logger.start = function () {
return internal.startReplicationLogger();
};
////////////////////////////////////////////////////////////////////////////////
/// @brief stops the replication logger
////////////////////////////////////////////////////////////////////////////////
exports.stopLogger = function () {
logger.stop = function () {
return internal.stopReplicationLogger();
};
////////////////////////////////////////////////////////////////////////////////
/// @brief return the replication logger state
////////////////////////////////////////////////////////////////////////////////
exports.getLoggerState = function () {
return internal.getStateReplicationLogger();
};
////////////////////////////////////////////////////////////////////////////////
/// @brief configures the replication applier
////////////////////////////////////////////////////////////////////////////////
exports.configureApplier = function (config) {
return internal.configureReplicationApplier(config);
logger.state = function () {
return internal.getStateReplicationLogger();
};
////////////////////////////////////////////////////////////////////////////////
/// @brief starts the replication applier
////////////////////////////////////////////////////////////////////////////////
exports.startApplier = function (forceFullSynchronisation) {
return internal.startReplicationApplier(forceFullSynchronisation);
applier.start = function (forceFullSynchronisation) {
return internal.startReplicationApplier(forceFullSynchronisation || false);
};
////////////////////////////////////////////////////////////////////////////////
/// @brief stops the replication applier
////////////////////////////////////////////////////////////////////////////////
exports.stopApplier = function () {
applier.stop = function () {
return internal.stopReplicationApplier();
};
@ -95,10 +90,38 @@ exports.stopApplier = function () {
/// @brief return the replication applier state
////////////////////////////////////////////////////////////////////////////////
exports.getApplierState = function () {
applier.state = function () {
return internal.getStateReplicationApplier();
};
////////////////////////////////////////////////////////////////////////////////
/// @brief returns the configuration of the replication applier
////////////////////////////////////////////////////////////////////////////////
applier.properties = function (config) {
if (config === undefined) {
return internal.configureReplicationApplier();
}
return internal.configureReplicationApplier(config);
};
////////////////////////////////////////////////////////////////////////////////
/// @}
////////////////////////////////////////////////////////////////////////////////
// -----------------------------------------------------------------------------
// --SECTION-- module exports
// -----------------------------------------------------------------------------
////////////////////////////////////////////////////////////////////////////////
/// @addtogroup ArangoShell
/// @{
////////////////////////////////////////////////////////////////////////////////
exports.logger = logger;
exports.applier = applier;
////////////////////////////////////////////////////////////////////////////////
/// @}
////////////////////////////////////////////////////////////////////////////////