/**
* Copyright (C) 2017 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.
*/
#include "mongo/platform/basic.h"
#include "mongo/db/sessions_collection_sharded.h"
#include "mongo/db/matcher/extensions_callback_noop.h"
#include "mongo/db/operation_context.h"
#include "mongo/db/query/canonical_query.h"
#include "mongo/db/query/query_request.h"
#include "mongo/db/sessions_collection_rs.h"
#include "mongo/s/commands/cluster_write.h"
#include "mongo/s/query/cluster_find.h"
#include "mongo/s/write_ops/batch_write_exec.h"
#include "mongo/s/write_ops/batched_command_request.h"
#include "mongo/s/write_ops/batched_command_response.h"
#include "mongo/util/net/op_msg.h"
namespace mongo {
namespace {
BSONObj lsidQuery(const LogicalSessionId& lsid) {
return BSON(LogicalSessionRecord::kIdFieldName << lsid.toBSON());
}
} // namespace
Status SessionsCollectionSharded::refreshSessions(OperationContext* opCtx,
const LogicalSessionRecordSet& sessions,
Date_t refreshTime) {
auto send = [&](BSONObj toSend) {
auto opMsg = OpMsgRequest::fromDBAndBody(SessionsCollection::kSessionsDb, toSend);
auto request = BatchedCommandRequest::parseUpdate(opMsg);
BatchedCommandResponse response;
BatchWriteExecStats stats;
ClusterWriter::write(opCtx, request, &stats, &response);
if (response.getOk()) {
return Status::OK();
}
auto error = response.isErrCodeSet() ? ErrorCodes::fromInt(response.getErrCode())
: ErrorCodes::UnknownError;
return Status(error, response.getErrMessage());
};
return doRefresh(kSessionsNamespaceString, sessions, refreshTime, send);
}
Status SessionsCollectionSharded::removeRecords(OperationContext* opCtx,
const LogicalSessionIdSet& sessions) {
auto send = [&](BSONObj toSend) {
auto opMsg = OpMsgRequest::fromDBAndBody(SessionsCollection::kSessionsDb, toSend);
auto request = BatchedCommandRequest::parseDelete(opMsg);
BatchedCommandResponse response;
BatchWriteExecStats stats;
ClusterWriter::write(opCtx, request, &stats, &response);
if (response.getOk()) {
return Status::OK();
}
auto error = response.isErrCodeSet() ? ErrorCodes::fromInt(response.getErrCode())
: ErrorCodes::UnknownError;
return Status(error, response.getErrMessage());
};
return doRemove(kSessionsNamespaceString, sessions, send);
}
StatusWith SessionsCollectionSharded::findRemovedSessions(
OperationContext* opCtx, const LogicalSessionIdSet& sessions) {
auto send = [&](BSONObj toSend) -> StatusWith {
const NamespaceString nss(SessionsCollection::kSessionsFullNS);
auto qr = QueryRequest::makeFromFindCommand(nss, toSend, false);
if (!qr.isOK()) {
return qr.getStatus();
}
const boost::intrusive_ptr expCtx;
auto cq = CanonicalQuery::canonicalize(opCtx,
std::move(qr.getValue()),
expCtx,
ExtensionsCallbackNoop(),
MatchExpressionParser::kAllowAllSpecialFeatures &
~MatchExpressionParser::AllowedFeatures::kExpr);
if (!cq.isOK()) {
return cq.getStatus();
}
// Do the work to generate the first batch of results. This blocks waiting to get responses
// from the shard(s).
std::vector batch;
BSONObj viewDefinition;
auto cursorId = ClusterFind::runQuery(
opCtx, *cq.getValue(), ReadPreferenceSetting::get(opCtx), &batch, &viewDefinition);
if (!cursorId.isOK()) {
return cursorId.getStatus();
}
BSONObjBuilder result;
CursorResponseBuilder firstBatch(/*firstBatch*/ true, &result);
for (const auto& obj : batch) {
firstBatch.append(obj);
}
firstBatch.done(cursorId.getValue(), nss.ns());
return result.obj();
};
return doFetch(kSessionsNamespaceString, sessions, send);
}
Status SessionsCollectionSharded::removeTransactionRecords(OperationContext* opCtx,
const LogicalSessionIdSet& sessions) {
return SessionsCollectionRS::removeTransactionRecordsHelper(opCtx, sessions);
}
} // namespace mongo