/**
* Copyright (C) 2014 MongoDB Inc.
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License, version 3,
* as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see .
*
* As a special exception, the copyright holders give permission to link the
* code of portions of this program with the OpenSSL library under certain
* conditions as described in each individual source file and distribute
* linked combinations including the program with the OpenSSL library. You
* must comply with the GNU Affero General Public License in all respects for
* all of the code used other than as permitted herein. If you modify file(s)
* with this exception, you may extend this exception to your version of the
* file(s), but you are not obligated to do so. If you do not wish to do so,
* delete this exception statement from your version. If you delete this
* exception statement from all source files in the program, then also delete
* it in the license file.
*/
#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kReplication
#include "mongo/platform/basic.h"
#include "mongo/db/repl/replication_coordinator_external_state_mock.h"
#include "mongo/base/status_with.h"
#include "mongo/bson/oid.h"
#include "mongo/db/client.h"
#include "mongo/db/jsobj.h"
#include "mongo/db/repl/oplog_buffer_blocking_queue.h"
#include "mongo/db/storage/snapshot_name.h"
#include "mongo/stdx/memory.h"
#include "mongo/util/log.h"
#include "mongo/util/net/hostandport.h"
#include "mongo/util/sequence_util.h"
namespace mongo {
namespace repl {
ReplicationCoordinatorExternalStateMock::ReplicationCoordinatorExternalStateMock()
: _localRsConfigDocument(ErrorCodes::NoMatchingDocument, "No local config document"),
_localRsLastVoteDocument(ErrorCodes::NoMatchingDocument, "No local lastVote document"),
_lastOpTime(ErrorCodes::NoMatchingDocument, "No last oplog entry"),
_canAcquireGlobalSharedLock(true),
_storeLocalConfigDocumentStatus(Status::OK()),
_storeLocalLastVoteDocumentStatus(Status::OK()),
_storeLocalConfigDocumentShouldHang(false),
_storeLocalLastVoteDocumentShouldHang(false),
_connectionsClosed(false),
_threadsStarted(false) {}
ReplicationCoordinatorExternalStateMock::~ReplicationCoordinatorExternalStateMock() {}
void ReplicationCoordinatorExternalStateMock::runOnInitialSyncThread(
stdx::function run) {
log() << "not running initial sync during test.";
}
void ReplicationCoordinatorExternalStateMock::startThreads(const ReplSettings& settings) {
_threadsStarted = true;
}
bool ReplicationCoordinatorExternalStateMock::isInitialSyncFlagSet(OperationContext*) {
return false;
}
void ReplicationCoordinatorExternalStateMock::startInitialSync(OnInitialSyncFinishedFn) {}
void ReplicationCoordinatorExternalStateMock::startSteadyStateReplication(OperationContext*,
ReplicationCoordinator*) {
}
void ReplicationCoordinatorExternalStateMock::stopDataReplication(OperationContext*) {}
void ReplicationCoordinatorExternalStateMock::startMasterSlave(OperationContext*) {}
Status ReplicationCoordinatorExternalStateMock::runRepairOnLocalDB(OperationContext* opCtx) {
return Status::OK();
}
Status ReplicationCoordinatorExternalStateMock::initializeReplSetStorage(OperationContext* opCtx,
const BSONObj& config) {
return storeLocalConfigDocument(opCtx, config);
}
void ReplicationCoordinatorExternalStateMock::shutdown(OperationContext*) {}
executor::TaskExecutor* ReplicationCoordinatorExternalStateMock::getTaskExecutor() const {
return nullptr;
}
OldThreadPool* ReplicationCoordinatorExternalStateMock::getDbWorkThreadPool() const {
return nullptr;
}
void ReplicationCoordinatorExternalStateMock::forwardSlaveProgress() {}
OID ReplicationCoordinatorExternalStateMock::ensureMe(OperationContext*) {
return OID::gen();
}
bool ReplicationCoordinatorExternalStateMock::isSelf(const HostAndPort& host,
ServiceContext* const service) {
return sequenceContains(_selfHosts, host);
}
void ReplicationCoordinatorExternalStateMock::addSelf(const HostAndPort& host) {
_selfHosts.push_back(host);
}
HostAndPort ReplicationCoordinatorExternalStateMock::getClientHostAndPort(
const OperationContext* opCtx) {
return _clientHostAndPort;
}
void ReplicationCoordinatorExternalStateMock::setClientHostAndPort(
const HostAndPort& clientHostAndPort) {
_clientHostAndPort = clientHostAndPort;
}
StatusWith ReplicationCoordinatorExternalStateMock::loadLocalConfigDocument(
OperationContext* opCtx) {
return _localRsConfigDocument;
}
Status ReplicationCoordinatorExternalStateMock::storeLocalConfigDocument(OperationContext* opCtx,
const BSONObj& config) {
{
stdx::unique_lock lock(_shouldHangConfigMutex);
while (_storeLocalConfigDocumentShouldHang) {
_shouldHangConfigCondVar.wait(lock);
}
}
if (_storeLocalConfigDocumentStatus.isOK()) {
setLocalConfigDocument(StatusWith(config));
return Status::OK();
}
return _storeLocalConfigDocumentStatus;
}
void ReplicationCoordinatorExternalStateMock::setLocalConfigDocument(
const StatusWith& localConfigDocument) {
_localRsConfigDocument = localConfigDocument;
}
StatusWith ReplicationCoordinatorExternalStateMock::loadLocalLastVoteDocument(
OperationContext* opCtx) {
return _localRsLastVoteDocument;
}
Status ReplicationCoordinatorExternalStateMock::storeLocalLastVoteDocument(
OperationContext* opCtx, const LastVote& lastVote) {
{
stdx::unique_lock lock(_shouldHangLastVoteMutex);
while (_storeLocalLastVoteDocumentShouldHang) {
_shouldHangLastVoteCondVar.wait(lock);
}
}
if (_storeLocalLastVoteDocumentStatus.isOK()) {
setLocalLastVoteDocument(StatusWith(lastVote));
return Status::OK();
}
return _storeLocalLastVoteDocumentStatus;
}
void ReplicationCoordinatorExternalStateMock::setLocalLastVoteDocument(
const StatusWith& localLastVoteDocument) {
_localRsLastVoteDocument = localLastVoteDocument;
}
void ReplicationCoordinatorExternalStateMock::setGlobalTimestamp(ServiceContext* service,
const Timestamp& newTime) {}
void ReplicationCoordinatorExternalStateMock::cleanUpLastApplyBatch(OperationContext* opCtx) {}
StatusWith ReplicationCoordinatorExternalStateMock::loadLastOpTime(
OperationContext* opCtx) {
return _lastOpTime;
}
void ReplicationCoordinatorExternalStateMock::setLastOpTime(const StatusWith& lastApplied) {
_lastOpTime = lastApplied;
}
void ReplicationCoordinatorExternalStateMock::setStoreLocalConfigDocumentStatus(Status status) {
_storeLocalConfigDocumentStatus = status;
}
void ReplicationCoordinatorExternalStateMock::setStoreLocalConfigDocumentToHang(bool hang) {
stdx::unique_lock lock(_shouldHangConfigMutex);
_storeLocalConfigDocumentShouldHang = hang;
if (!hang) {
_shouldHangConfigCondVar.notify_all();
}
}
bool ReplicationCoordinatorExternalStateMock::threadsStarted() const {
return _threadsStarted;
}
void ReplicationCoordinatorExternalStateMock::setStoreLocalLastVoteDocumentStatus(Status status) {
_storeLocalLastVoteDocumentStatus = status;
}
void ReplicationCoordinatorExternalStateMock::setStoreLocalLastVoteDocumentToHang(bool hang) {
stdx::unique_lock lock(_shouldHangLastVoteMutex);
_storeLocalLastVoteDocumentShouldHang = hang;
if (!hang) {
_shouldHangLastVoteCondVar.notify_all();
}
}
void ReplicationCoordinatorExternalStateMock::closeConnections() {
_connectionsClosed = true;
}
void ReplicationCoordinatorExternalStateMock::killAllUserOperations(OperationContext* opCtx) {}
void ReplicationCoordinatorExternalStateMock::shardingOnStepDownHook() {}
void ReplicationCoordinatorExternalStateMock::signalApplierToChooseNewSyncSource() {}
void ReplicationCoordinatorExternalStateMock::stopProducer() {}
void ReplicationCoordinatorExternalStateMock::startProducerIfStopped() {}
void ReplicationCoordinatorExternalStateMock::dropAllSnapshots() {}
void ReplicationCoordinatorExternalStateMock::updateCommittedSnapshot(SnapshotName newCommitPoint) {
}
void ReplicationCoordinatorExternalStateMock::createSnapshot(OperationContext* opCtx,
SnapshotName name) {}
void ReplicationCoordinatorExternalStateMock::forceSnapshotCreation() {}
bool ReplicationCoordinatorExternalStateMock::snapshotsEnabled() const {
return _areSnapshotsEnabled;
}
void ReplicationCoordinatorExternalStateMock::setAreSnapshotsEnabled(bool val) {
_areSnapshotsEnabled = val;
}
void ReplicationCoordinatorExternalStateMock::notifyOplogMetadataWaiters() {}
double ReplicationCoordinatorExternalStateMock::getElectionTimeoutOffsetLimitFraction() const {
return 0.15;
}
bool ReplicationCoordinatorExternalStateMock::isReadCommittedSupportedByStorageEngine(
OperationContext* opCtx) const {
return _isReadCommittedSupported;
}
StatusWith ReplicationCoordinatorExternalStateMock::multiApply(
OperationContext*, MultiApplier::Operations, MultiApplier::ApplyOperationFn) {
return {ErrorCodes::InternalError, "Method not implemented"};
}
Status ReplicationCoordinatorExternalStateMock::multiSyncApply(MultiApplier::OperationPtrs* ops) {
return Status::OK();
}
Status ReplicationCoordinatorExternalStateMock::multiInitialSyncApply(
MultiApplier::OperationPtrs* ops, const HostAndPort& source, AtomicUInt32* fetchCount) {
return Status::OK();
}
std::unique_ptr ReplicationCoordinatorExternalStateMock::makeInitialSyncOplogBuffer(
OperationContext* opCtx) const {
return stdx::make_unique();
}
std::unique_ptr ReplicationCoordinatorExternalStateMock::makeSteadyStateOplogBuffer(
OperationContext* opCtx) const {
return stdx::make_unique();
}
bool ReplicationCoordinatorExternalStateMock::shouldUseDataReplicatorInitialSync() const {
return true;
}
std::size_t ReplicationCoordinatorExternalStateMock::getOplogFetcherMaxFetcherRestarts() const {
return 0;
}
void ReplicationCoordinatorExternalStateMock::setIsReadCommittedEnabled(bool val) {
_isReadCommittedSupported = val;
}
void ReplicationCoordinatorExternalStateMock::onDrainComplete(OperationContext* opCtx) {}
OpTime ReplicationCoordinatorExternalStateMock::onTransitionToPrimary(OperationContext* opCtx,
bool isV1ElectionProtocol) {
if (isV1ElectionProtocol) {
_lastOpTime = OpTime(Timestamp(1, 0), 1);
}
return fassertStatusOK(40297, _lastOpTime);
}
void ReplicationCoordinatorExternalStateMock::startNoopWriter(OpTime opTime) {}
void ReplicationCoordinatorExternalStateMock::stopNoopWriter() {}
void ReplicationCoordinatorExternalStateMock::setupNoopWriter(Seconds waitTime) {}
} // namespace repl
} // namespace mongo