-
Notifications
You must be signed in to change notification settings - Fork 1.8k
/
Copy pathoperation.ts
143 lines (114 loc) · 4.3 KB
/
operation.ts
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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
import { type BSONSerializeOptions, type Document, resolveBSONOptions } from '../bson';
import { type Abortable } from '../mongo_types';
import { ReadPreference, type ReadPreferenceLike } from '../read_preference';
import type { Server } from '../sdam/server';
import type { ClientSession } from '../sessions';
import { type TimeoutContext } from '../timeout';
import type { MongoDBNamespace } from '../utils';
export const Aspect = {
READ_OPERATION: Symbol('READ_OPERATION'),
WRITE_OPERATION: Symbol('WRITE_OPERATION'),
RETRYABLE: Symbol('RETRYABLE'),
EXPLAINABLE: Symbol('EXPLAINABLE'),
SKIP_COLLATION: Symbol('SKIP_COLLATION'),
CURSOR_CREATING: Symbol('CURSOR_CREATING'),
MUST_SELECT_SAME_SERVER: Symbol('MUST_SELECT_SAME_SERVER'),
COMMAND_BATCHING: Symbol('COMMAND_BATCHING')
} as const;
/** @public */
export type Hint = string | Document;
/** @public */
export interface OperationOptions extends BSONSerializeOptions {
/** Specify ClientSession for this command */
session?: ClientSession;
willRetryWrite?: boolean;
/** The preferred read preference (ReadPreference.primary, ReadPreference.primary_preferred, ReadPreference.secondary, ReadPreference.secondary_preferred, ReadPreference.nearest). */
readPreference?: ReadPreferenceLike;
/** @internal Hints to `executeOperation` that this operation should not unpin on an ended transaction */
bypassPinningCheck?: boolean;
omitReadPreference?: boolean;
/** @internal Hint to `executeOperation` to omit maxTimeMS */
omitMaxTimeMS?: boolean;
/**
* @experimental
* Specifies the time an operation will run until it throws a timeout error
*/
timeoutMS?: number;
}
/**
* This class acts as a parent class for any operation and is responsible for setting this.options,
* as well as setting and getting a session.
* Additionally, this class implements `hasAspect`, which determines whether an operation has
* a specific aspect.
* @internal
*/
export abstract class AbstractOperation<TResult = any> {
ns!: MongoDBNamespace;
readPreference: ReadPreference;
server!: Server;
bypassPinningCheck: boolean;
trySecondaryWrite: boolean;
// BSON serialization options
bsonOptions?: BSONSerializeOptions;
options: OperationOptions & Abortable;
/** Specifies the time an operation will run until it throws a timeout error. */
timeoutMS?: number;
private _session: ClientSession | undefined;
static aspects?: Set<symbol>;
constructor(options: OperationOptions & Abortable = {}) {
this.readPreference = this.hasAspect(Aspect.WRITE_OPERATION)
? ReadPreference.primary
: (ReadPreference.fromOptions(options) ?? ReadPreference.primary);
// Pull the BSON serialize options from the already-resolved options
this.bsonOptions = resolveBSONOptions(options);
this._session = options.session != null ? options.session : undefined;
this.options = options;
this.bypassPinningCheck = !!options.bypassPinningCheck;
this.trySecondaryWrite = false;
}
/** Must match the first key of the command object sent to the server.
Command name should be stateless (should not use 'this' keyword) */
abstract get commandName(): string;
abstract execute(
server: Server,
session: ClientSession | undefined,
timeoutContext: TimeoutContext
): Promise<TResult>;
hasAspect(aspect: symbol): boolean {
const ctor = this.constructor as { aspects?: Set<symbol> };
if (ctor.aspects == null) {
return false;
}
return ctor.aspects.has(aspect);
}
// Make sure the session is not writable from outside this class.
get session(): ClientSession | undefined {
return this._session;
}
clearSession() {
this._session = undefined;
}
resetBatch(): boolean {
return true;
}
get canRetryRead(): boolean {
return this.hasAspect(Aspect.RETRYABLE) && this.hasAspect(Aspect.READ_OPERATION);
}
get canRetryWrite(): boolean {
return this.hasAspect(Aspect.RETRYABLE) && this.hasAspect(Aspect.WRITE_OPERATION);
}
}
export function defineAspects(
operation: { aspects?: Set<symbol> },
aspects: symbol | symbol[] | Set<symbol>
): Set<symbol> {
if (!Array.isArray(aspects) && !(aspects instanceof Set)) {
aspects = [aspects];
}
aspects = new Set(aspects);
Object.defineProperty(operation, 'aspects', {
value: aspects,
writable: false
});
return aspects;
}