Skip to content

Commit fc3f3e6

Browse files
mattijauhiainenmattijauhiainen
authored
Resolve SRV urls (#198)
1 parent a9ccb29 commit fc3f3e6

9 files changed

Lines changed: 569 additions & 156 deletions

File tree

‎src/client.ts‎

Lines changed: 20 additions & 92 deletions
Original file line numberDiff line numberDiff line change
@@ -1,97 +1,29 @@
1-
import { assert } from "../deps.ts";
21
import { Database } from "./database.ts";
3-
import { WireProtocol } from "./protocol/mod.ts";
4-
import {
5-
ConnectOptions,
6-
Credential,
7-
Document,
8-
ListDatabaseInfo,
9-
} from "./types.ts";
2+
import { ConnectOptions, Document, ListDatabaseInfo } from "./types.ts";
103
import { parse } from "./utils/uri.ts";
11-
import { AuthContext, ScramAuthPlugin, X509AuthPlugin } from "./auth/mod.ts";
124
import { MongoError } from "./error.ts";
5+
import { Cluster } from "./cluster.ts";
6+
import { assert } from "../deps.ts";
137

148
const DENO_DRIVER_VERSION = "0.0.1";
159

16-
export interface DenoConnectOptions {
17-
hostname: string;
18-
port: number;
19-
certFile?: string;
20-
}
21-
2210
export class MongoClient {
23-
#protocol?: WireProtocol;
24-
#conn?: Deno.Conn;
11+
#cluster?: Cluster;
2512

2613
async connect(
2714
options: ConnectOptions | string,
28-
serverIndex: number = 0,
2915
): Promise<Database> {
3016
try {
31-
if (typeof options === "string") {
32-
options = parse(options);
33-
}
34-
let conn;
35-
const denoConnectOps: DenoConnectOptions = {
36-
hostname: options.servers[serverIndex].host,
37-
port: options.servers[serverIndex].port,
38-
};
39-
if (options.tls) {
40-
if (options.certFile) {
41-
denoConnectOps.certFile = options.certFile;
42-
}
43-
if (options.keyFile) {
44-
if (options.keyFilePassword) {
45-
throw new MongoError(
46-
`Tls keyFilePassword not implemented in Deno driver`,
47-
);
48-
//TODO, need something like const key = decrypt(options.keyFile) ...
49-
}
50-
throw new MongoError(`Tls keyFile not implemented in Deno driver`);
51-
//TODO, need Deno.connectTls with something like key or keyFile option.
52-
}
53-
conn = await Deno.connectTls(denoConnectOps);
54-
} else {
55-
conn = await Deno.connect(denoConnectOps);
56-
}
57-
58-
this.#conn = conn;
59-
this.#protocol = new WireProtocol(conn);
60-
61-
if ((options as ConnectOptions).credential) {
62-
const authContext = new AuthContext(
63-
this.#protocol,
64-
(options as ConnectOptions).credential,
65-
options as ConnectOptions,
66-
);
67-
const mechanism = (options as ConnectOptions).credential!.mechanism;
68-
let authPlugin;
69-
if (mechanism === "SCRAM-SHA-256") {
70-
authPlugin = new ScramAuthPlugin("sha256"); //TODO AJUST sha256
71-
} else if (mechanism === "SCRAM-SHA-1") {
72-
authPlugin = new ScramAuthPlugin("sha1");
73-
} else if (mechanism === "MONGODB-X509") {
74-
authPlugin = new X509AuthPlugin();
75-
} else {
76-
throw new MongoError(
77-
`Auth mechanism not implemented in Deno driver: ${mechanism}`,
78-
);
79-
}
80-
const request = authPlugin.prepare(authContext);
81-
authContext.response = await this.#protocol.commandSingle(
82-
"admin",
83-
request,
84-
);
85-
await authPlugin.auth(authContext);
86-
} else {
87-
await this.#protocol.connect();
88-
}
17+
const parsedOptions = typeof options === "string"
18+
? await parse(options)
19+
: options;
20+
const cluster = new Cluster(parsedOptions);
21+
await cluster.connect();
22+
await cluster.authenticate();
23+
await cluster.updateMaster();
24+
this.#cluster = cluster;
8925
} catch (e) {
90-
if (serverIndex < (options as ConnectOptions).servers.length - 1) {
91-
return await this.connect(options, serverIndex + 1);
92-
} else {
93-
throw new MongoError(`Connection failed: ${e.message || e}`);
94-
}
26+
throw new MongoError(`Connection failed: ${e.message || e}`);
9527
}
9628
return this.database((options as ConnectOptions).db);
9729
}
@@ -102,11 +34,11 @@ export class MongoClient {
10234
authorizedCollections?: boolean;
10335
comment?: Document;
10436
}): Promise<ListDatabaseInfo[]> {
105-
assert(this.#protocol);
10637
if (!options) {
10738
options = {};
10839
}
109-
const { databases } = await this.#protocol.commandSingle("admin", {
40+
assert(this.#cluster);
41+
const { databases } = await this.#cluster.protocol.commandSingle("admin", {
11042
listDatabases: 1,
11143
...options,
11244
});
@@ -115,20 +47,16 @@ export class MongoClient {
11547

11648
// TODO: add test cases
11749
async runCommand<T = any>(db: string, body: Document): Promise<T> {
118-
assert(this.#protocol);
119-
return await this.#protocol.commandSingle(db, body);
50+
assert(this.#cluster);
51+
return await this.#cluster.protocol.commandSingle(db, body);
12052
}
12153

12254
database(name: string): Database {
123-
assert(this.#protocol);
124-
return new Database(this.#protocol, name);
55+
assert(this.#cluster);
56+
return new Database(this.#cluster, name);
12557
}
12658

12759
close() {
128-
if (this.#conn) {
129-
Deno.close(this.#conn.rid);
130-
this.#conn = undefined;
131-
this.#protocol = undefined;
132-
}
60+
if (this.#cluster) this.#cluster.close();
13361
}
13462
}

‎src/cluster.ts‎

Lines changed: 133 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,133 @@
1+
import { WireProtocol } from "./protocol/mod.ts";
2+
import { ConnectOptions } from "./types.ts";
3+
import { AuthContext, ScramAuthPlugin, X509AuthPlugin } from "./auth/mod.ts";
4+
import { MongoError } from "./error.ts";
5+
import { assert } from "../deps.ts";
6+
import { Server } from "./types.ts";
7+
8+
export interface DenoConnectOptions {
9+
hostname: string;
10+
port: number;
11+
certFile?: string;
12+
}
13+
14+
export class Cluster {
15+
#options: ConnectOptions;
16+
#connections: Deno.Conn[];
17+
#protocols: WireProtocol[];
18+
#masterIndex: number;
19+
20+
constructor(options: ConnectOptions) {
21+
this.#options = options;
22+
this.#connections = [];
23+
this.#protocols = [];
24+
this.#masterIndex = -1;
25+
}
26+
27+
async connect() {
28+
const options = this.#options;
29+
this.#connections = await Promise.all(
30+
options.servers.map((server) => this.connectToServer(server, options)),
31+
);
32+
}
33+
34+
async connectToServer(server: Server, options: ConnectOptions) {
35+
const denoConnectOps: DenoConnectOptions = {
36+
hostname: server.host,
37+
port: server.port,
38+
};
39+
if (options.tls) {
40+
if (options.certFile) {
41+
denoConnectOps.certFile = options.certFile;
42+
}
43+
if (options.keyFile) {
44+
if (options.keyFilePassword) {
45+
throw new MongoError(
46+
`Tls keyFilePassword not implemented in Deno driver`,
47+
);
48+
//TODO, need something like const key = decrypt(options.keyFile) ...
49+
}
50+
throw new MongoError(`Tls keyFile not implemented in Deno driver`);
51+
//TODO, need Deno.connectTls with something like key or keyFile option.
52+
}
53+
return await Deno.connectTls(denoConnectOps);
54+
} else {
55+
return await Deno.connect(denoConnectOps);
56+
}
57+
}
58+
59+
async authenticate() {
60+
const options = this.#options;
61+
this.#protocols = await Promise.all(
62+
this.#connections.map((conn) => this.authenticateToServer(conn, options)),
63+
);
64+
}
65+
66+
async authenticateToServer(conn: Deno.Conn, options: ConnectOptions) {
67+
const protocol = new WireProtocol(conn);
68+
if (options.credential) {
69+
const authContext = new AuthContext(
70+
protocol,
71+
options.credential,
72+
options,
73+
);
74+
const mechanism = options.credential!.mechanism;
75+
let authPlugin;
76+
if (mechanism === "SCRAM-SHA-256") {
77+
authPlugin = new ScramAuthPlugin("sha256"); //TODO AJUST sha256
78+
} else if (mechanism === "SCRAM-SHA-1") {
79+
authPlugin = new ScramAuthPlugin("sha1");
80+
} else if (mechanism === "MONGODB-X509") {
81+
authPlugin = new X509AuthPlugin();
82+
} else {
83+
throw new MongoError(
84+
`Auth mechanism not implemented in Deno driver: ${mechanism}`,
85+
);
86+
}
87+
const request = authPlugin.prepare(authContext);
88+
authContext.response = await protocol.commandSingle(
89+
"admin", // TODO: Should get the auth db from connectionOptions?
90+
request,
91+
);
92+
await authPlugin.auth(authContext);
93+
} else {
94+
await protocol.connect();
95+
}
96+
return protocol;
97+
}
98+
99+
async updateMaster() {
100+
const results = await Promise.all(this.#protocols.map((protocol) => {
101+
return protocol.commandSingle(
102+
"admin",
103+
{ hello: 1 },
104+
);
105+
}));
106+
const masterIndex = results.findIndex((result) => result.isWritablePrimary);
107+
if (masterIndex === -1) throw new Error(`Could not find a master node`);
108+
this.#masterIndex = masterIndex;
109+
}
110+
111+
private getMaster() {
112+
return {
113+
protocol: this.#protocols[this.#masterIndex],
114+
conn: this.#connections[this.#masterIndex],
115+
};
116+
}
117+
118+
get protocol() {
119+
const protocol = this.getMaster().protocol;
120+
assert(protocol);
121+
return protocol;
122+
}
123+
124+
close() {
125+
this.#connections.forEach((connection) => {
126+
try {
127+
Deno.close(connection.rid);
128+
} catch (error) {
129+
console.error(`Error closing connection: ${error}`);
130+
}
131+
});
132+
}
133+
}

‎src/database.ts‎

Lines changed: 10 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
import { Collection } from "./collection/mod.ts";
2-
import { CommandCursor, WireProtocol } from "./protocol/mod.ts";
2+
import { CommandCursor } from "./protocol/mod.ts";
33
import { CreateUserOptions, Document } from "./types.ts";
4+
import { Cluster } from "./cluster.ts";
45

56
interface ListCollectionsReponse {
67
cursor: {
@@ -22,14 +23,14 @@ export interface ListCollectionsResult {
2223
}
2324

2425
export class Database {
25-
#protocol: WireProtocol;
26+
#cluster: Cluster;
2627

27-
constructor(protocol: WireProtocol, readonly name: string) {
28-
this.#protocol = protocol;
28+
constructor(cluster: Cluster, readonly name: string) {
29+
this.#cluster = cluster;
2930
}
3031

3132
collection<T>(name: string): Collection<T> {
32-
return new Collection(this.#protocol, this.name, name);
33+
return new Collection(this.#cluster.protocol, this.name, name);
3334
}
3435

3536
listCollections(options?: {
@@ -42,9 +43,9 @@ export class Database {
4243
options = {};
4344
}
4445
return new CommandCursor<ListCollectionsResult>(
45-
this.#protocol,
46+
this.#cluster.protocol,
4647
async () => {
47-
const { cursor } = await this.#protocol.commandSingle<
48+
const { cursor } = await this.#cluster.protocol.commandSingle<
4849
ListCollectionsReponse
4950
>(this.name, {
5051
listCollections: 1,
@@ -82,7 +83,7 @@ export class Database {
8283
password: string,
8384
options?: CreateUserOptions,
8485
) {
85-
await this.#protocol.commandSingle(this.name, {
86+
await this.#cluster.protocol.commandSingle(this.name, {
8687
createUser: options?.username ?? username,
8788
pwd: options?.password ?? password,
8889
customData: options?.customData,
@@ -99,7 +100,7 @@ export class Database {
99100
writeConcern?: Document;
100101
comment?: Document;
101102
}) {
102-
await this.#protocol.commandSingle(this.name, {
103+
await this.#cluster.protocol.commandSingle(this.name, {
103104
dropUser: username,
104105
writeConcern: options?.writeConcern,
105106
comment: options?.comment,

0 commit comments

Comments
 (0)