1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
|
// @file strategy.cpp
/*
* Copyright (C) 2010 10gen 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 <http://www.gnu.org/licenses/>.
*/
#include "pch.h"
#include "../client/connpool.h"
#include "../db/commands.h"
#include "grid.h"
#include "request.h"
#include "server.h"
#include "writeback_listener.h"
#include "strategy.h"
namespace mongo {
// ----- Strategy ------
void Strategy::doWrite( int op , Request& r , const Shard& shard , bool checkVersion ) {
ShardConnection conn( shard , r.getns() );
if ( ! checkVersion )
conn.donotCheckVersion();
else if ( conn.setVersion() ) {
conn.done();
throw StaleConfigException( r.getns() , "doWrite" , true );
}
conn->say( r.m() );
conn.done();
}
void Strategy::doQuery( Request& r , const Shard& shard ) {
r.checkAuth();
ShardConnection dbcon( shard , r.getns() );
DBClientBase &c = dbcon.conn();
string actualServer;
Message response;
bool ok = c.call( r.m(), response, true , &actualServer );
uassert( 10200 , "mongos: error calling db", ok );
{
QueryResult *qr = (QueryResult *) response.singleData();
if ( qr->resultFlags() & ResultFlag_ShardConfigStale ) {
dbcon.done();
throw StaleConfigException( r.getns() , "Strategy::doQuery" );
}
}
r.reply( response , actualServer.size() ? actualServer : c.getServerAddress() );
dbcon.done();
}
void Strategy::insert( const Shard& shard , const char * ns , const BSONObj& obj , int flags, bool safe ) {
ShardConnection dbcon( shard , ns );
if ( dbcon.setVersion() ) {
dbcon.done();
throw StaleConfigException( ns , "for insert" );
}
dbcon->insert( ns , obj , flags);
if (safe)
dbcon->getLastError();
dbcon.done();
}
void Strategy::update( const Shard& shard , const char * ns , const BSONObj& query , const BSONObj& toupdate , int flags, bool safe ) {
bool upsert = flags & UpdateOption_Upsert;
bool multi = flags & UpdateOption_Multi;
ShardConnection dbcon( shard , ns );
if ( dbcon.setVersion() ) {
dbcon.done();
throw StaleConfigException( ns , "for insert" );
}
dbcon->update( ns , query , toupdate, upsert, multi);
if (safe)
dbcon->getLastError();
dbcon.done();
}
}
|
