addresses, TypeDBCredential credential) {
return new TypeDBDriverImpl(addresses, credential);
}
+
+ /**
+ * Open a TypeDB Driver to TypeDB Cloud server(s), using provided address translation, with
+ * the provided credential.
+ *
+ * Examples
+ *
+ * TypeDB.cloudDriver(addressTranslation, credential);
+ *
+ *
+ * @param addressTranslation Translation map from addresses received from the TypeDB server(s)
+ * to addresses to be used by the driver for connection
+ * @param credential The credential to connect with
+ */
+ public static TypeDBDriver cloudDriver(Map addressTranslation, TypeDBCredential credential) {
+ return new TypeDBDriverImpl(addressTranslation, credential);
+ }
}
diff --git a/java/api/database/Database.java b/java/api/database/Database.java
index 542d9039e0..a8b5c81fae 100644
--- a/java/api/database/Database.java
+++ b/java/api/database/Database.java
@@ -116,10 +116,10 @@ public interface Database {
interface Replica {
/**
- * Retrieves the address of the server hosting this replica
+ * The server hosting this replica
*/
@CheckReturnValue
- String address();
+ String server();
/**
* Checks whether this is the primary replica of the raft cluster.
diff --git a/java/connection/TypeDBDatabaseImpl.java b/java/connection/TypeDBDatabaseImpl.java
index 61fd7fd16a..5b7cf56d8e 100644
--- a/java/connection/TypeDBDatabaseImpl.java
+++ b/java/connection/TypeDBDatabaseImpl.java
@@ -37,7 +37,7 @@
import static com.vaticle.typedb.driver.jni.typedb_driver.database_rule_schema;
import static com.vaticle.typedb.driver.jni.typedb_driver.database_schema;
import static com.vaticle.typedb.driver.jni.typedb_driver.database_type_schema;
-import static com.vaticle.typedb.driver.jni.typedb_driver.replica_info_get_address;
+import static com.vaticle.typedb.driver.jni.typedb_driver.replica_info_get_server;
import static com.vaticle.typedb.driver.jni.typedb_driver.replica_info_get_term;
import static com.vaticle.typedb.driver.jni.typedb_driver.replica_info_is_preferred;
import static com.vaticle.typedb.driver.jni.typedb_driver.replica_info_is_primary;
@@ -127,8 +127,8 @@ public static class Replica extends NativeObject implements TypeDBDriver {
@@ -49,6 +53,10 @@ public TypeDBDriverImpl(Set initAddresses, TypeDBCredential credential)
this(openCloud(initAddresses, credential));
}
+ public TypeDBDriverImpl(Map addressTranslation, TypeDBCredential credential) throws TypeDBDriverException {
+ this(openCloud(addressTranslation, credential));
+ }
+
private TypeDBDriverImpl(com.vaticle.typedb.driver.jni.Connection connection) {
super(connection);
databaseMgr = new TypeDBDatabaseManagerImpl(this.nativeObject);
@@ -71,6 +79,24 @@ private static com.vaticle.typedb.driver.jni.Connection openCloud(Set in
}
}
+ private static com.vaticle.typedb.driver.jni.Connection openCloud(Map addressTranslation, TypeDBCredential credential) {
+ try {
+ List advertised = new ArrayList();
+ List translated = new ArrayList();
+ for (Map.Entry entry: addressTranslation.entrySet()) {
+ advertised.add(entry.getKey());
+ translated.add(entry.getValue());
+ }
+ return connection_open_cloud_translated(
+ advertised.toArray(new String[0]),
+ translated.toArray(new String[0]),
+ credential.nativeObject
+ );
+ } catch (com.vaticle.typedb.driver.jni.Error e) {
+ throw new TypeDBDriverException(e);
+ }
+ }
+
@Override
public boolean isOpen() {
return connection_is_open(nativeObject);
diff --git a/java/docs/connection/Database.Replica.adoc b/java/docs/connection/Database.Replica.adoc
index 115a27b252..3f5426cf30 100644
--- a/java/docs/connection/Database.Replica.adoc
+++ b/java/docs/connection/Database.Replica.adoc
@@ -6,50 +6,50 @@
The metadata and state of an individual raft replica of a database.
// tag::methods[]
-[#_Database_Replica_address__]
-==== address
+[#_Database_Replica_isPreferred__]
+==== isPreferred
[source,java]
----
@CheckReturnValue
-java.lang.String address()
+boolean isPreferred()
----
-Retrieves the address of the server hosting this replica
+Checks whether this is the preferred replica of the raft cluster. If true, Operations which can be run on any replica will prefer to use this replica.
[caption=""]
.Returns
-`java.lang.String`
+`boolean`
-[#_Database_Replica_isPreferred__]
-==== isPreferred
+[#_Database_Replica_isPrimary__]
+==== isPrimary
[source,java]
----
@CheckReturnValue
-boolean isPreferred()
+boolean isPrimary()
----
-Checks whether this is the preferred replica of the raft cluster. If true, Operations which can be run on any replica will prefer to use this replica.
+Checks whether this is the primary replica of the raft cluster.
[caption=""]
.Returns
`boolean`
-[#_Database_Replica_isPrimary__]
-==== isPrimary
+[#_Database_Replica_server__]
+==== server
[source,java]
----
@CheckReturnValue
-boolean isPrimary()
+java.lang.String server()
----
-Checks whether this is the primary replica of the raft cluster.
+The server hosting this replica
[caption=""]
.Returns
-`boolean`
+`java.lang.String`
[#_Database_Replica_term__]
==== term
diff --git a/java/docs/connection/TypeDB.adoc b/java/docs/connection/TypeDB.adoc
index 3608520d07..6411f63d1d 100644
--- a/java/docs/connection/TypeDB.adoc
+++ b/java/docs/connection/TypeDB.adoc
@@ -95,6 +95,39 @@ a| `credential` a| The credential to connect with a| `TypeDBCredential`
TypeDB.cloudDriver(addresses, credential);
----
+[#_TypeDB_cloudDriver__java_util_Map_java_lang_String_java_lang_String___TypeDBCredential]
+==== cloudDriver
+
+[source,java]
+----
+public static TypeDBDriver cloudDriver(java.util.Map addressTranslation,
+ TypeDBCredential credential)
+----
+
+Open a TypeDB Driver to TypeDB Cloud server(s), using provided address translation, with the provided credential.
+
+
+[caption=""]
+.Input parameters
+[cols=",,"]
+[options="header"]
+|===
+|Name |Description |Type
+a| `addressTranslation` a| Translation map from addresses received from the TypeDB server(s) to addresses to be used by the driver for connection a| `java.util.Map`
+a| `credential` a| The credential to connect with a| `TypeDBCredential`
+|===
+
+[caption=""]
+.Returns
+`public static TypeDBDriver`
+
+[caption=""]
+.Code examples
+[source,java]
+----
+TypeDB.cloudDriver(addressTranslation, credential);
+----
+
[#_TypeDB_coreDriver__java_lang_String]
==== coreDriver
diff --git a/java/test/integration/AddressTranslationTest.java b/java/test/integration/AddressTranslationTest.java
new file mode 100644
index 0000000000..f25da6bbae
--- /dev/null
+++ b/java/test/integration/AddressTranslationTest.java
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.vaticle.typedb.driver.test.integration;
+
+import com.vaticle.typedb.cloud.tool.runner.TypeDBCloudRunner;
+import com.vaticle.typedb.common.collection.Pair;
+import com.vaticle.typedb.driver.TypeDB;
+import com.vaticle.typedb.driver.api.TypeDBCredential;
+import com.vaticle.typedb.driver.api.TypeDBDriver;
+import com.vaticle.typedb.driver.api.TypeDBSession;
+import com.vaticle.typedb.driver.api.TypeDBTransaction;
+import com.vaticle.typedb.driver.api.concept.type.EntityType;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.file.Paths;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static com.vaticle.typedb.common.collection.Collections.map;
+import static com.vaticle.typedb.common.collection.Collections.pair;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.fail;
+
+@SuppressWarnings("Duplicates")
+public class AddressTranslationTest {
+ private static final Logger LOG = LoggerFactory.getLogger(AddressTranslationTest.class);
+
+ private static final Map serverOptions = map(
+ pair("--diagnostics.reporting.errors", "false"),
+ pair("--diagnostics.reporting.statistics", "false"),
+ pair("--diagnostics.monitoring.enable", "false")
+ );
+ private static final TypeDBCredential credential = new TypeDBCredential("admin", "password", false);
+
+ @Test
+ public void testAddressTranslation() {
+ TypeDBCloudRunner typedb = TypeDBCloudRunner.create(Paths.get("."), 3, serverOptions);
+ typedb.start();
+ Map addresses = typedb.externalAddresses().stream().map(address -> pair(address, address))
+ .collect(Collectors.toMap(Pair::first, Pair::second));
+ TypeDBDriver driver = TypeDB.cloudDriver(addresses, credential);
+ driver.databases().create("typedb");
+ TypeDBSession session = driver.session("typedb", TypeDBSession.Type.DATA);
+ TypeDBTransaction tx = session.transaction(TypeDBTransaction.Type.WRITE);
+ EntityType root = tx.concepts().getRootEntityType();
+ assertNotNull(root);
+ assertEquals(1, root.getSubtypes(tx).collect(Collectors.toList()).size());
+ tx.close();
+ session.close();
+ driver.close();
+ }
+}
diff --git a/java/test/integration/BUILD b/java/test/integration/BUILD
index 5c7dacd507..14f6b44e56 100644
--- a/java/test/integration/BUILD
+++ b/java/test/integration/BUILD
@@ -44,6 +44,29 @@ typedb_java_test(
],
)
+typedb_java_test(
+ name = "test-address-translation",
+ srcs = ["AddressTranslationTest.java"],
+ server_artifacts = {
+ "@vaticle_bazel_distribution//platform:is_linux_arm64": "@vaticle_typedb_cloud_artifact_linux-arm64//file",
+ "@vaticle_bazel_distribution//platform:is_linux_x86_64": "@vaticle_typedb_cloud_artifact_linux-x86_64//file",
+ "@vaticle_bazel_distribution//platform:is_mac_arm64": "@vaticle_typedb_cloud_artifact_mac-arm64//file",
+ "@vaticle_bazel_distribution//platform:is_mac_x86_64": "@vaticle_typedb_cloud_artifact_mac-x86_64//file",
+ "@vaticle_bazel_distribution//platform:is_windows_x86_64": "@vaticle_typedb_cloud_artifact_windows-x86_64//file",
+ },
+ test_class = "com.vaticle.typedb.driver.test.integration.AddressTranslationTest",
+ deps = [
+ # Internal dependencies
+ "//java:driver-java",
+ "//java/api",
+
+ # External dependencies from @vaticle
+ "@vaticle_typeql//common/java:common",
+ "@maven//:org_slf4j_slf4j_api",
+ "@maven//:com_vaticle_typedb_typedb_cloud_runner",
+ ],
+)
+
checkstyle_test(
name = "checkstyle",
include = glob(["*"]),
diff --git a/nodejs/TypeDB.ts b/nodejs/TypeDB.ts
index 633ea5925e..4b6372ef80 100644
--- a/nodejs/TypeDB.ts
+++ b/nodejs/TypeDB.ts
@@ -50,7 +50,7 @@ export namespace TypeDB {
* const driver = TypeDB.cloudDriver(["127.0.0.1:11729"], new TypeDBCredential(username, password));
* ```
*/
- export function cloudDriver(addresses: string | string[], credential: TypeDBCredential): Promise {
+ export function cloudDriver(addresses: string | string[] | Record, credential: TypeDBCredential): Promise {
if (typeof addresses === 'string') addresses = [addresses];
return new TypeDBDriverImpl(addresses, credential).open();
}
diff --git a/nodejs/api/connection/database/Database.ts b/nodejs/api/connection/database/Database.ts
index d063d1aa58..035d62f045 100644
--- a/nodejs/api/connection/database/Database.ts
+++ b/nodejs/api/connection/database/Database.ts
@@ -69,7 +69,7 @@ export namespace Database {
* If true, Operations which can be run on any replica will prefer to use this replica.
*/
readonly preferred: boolean;
- /** The address of the server hosting this replica */
- readonly address: string;
+ /** The server hosting this replica */
+ readonly server: string;
}
}
diff --git a/nodejs/common/errors/ErrorMessage.ts b/nodejs/common/errors/ErrorMessage.ts
index 6cf02d49aa..63b651449f 100644
--- a/nodejs/common/errors/ErrorMessage.ts
+++ b/nodejs/common/errors/ErrorMessage.ts
@@ -102,6 +102,7 @@ export namespace ErrorMessage {
export const CLOUD_INVALID_ROOT_CA_PATH = new Driver(21, (args: Stringable[]) => `The provided Root CA path '${args[0]}' does not exist`);
export const UNRECOGNISED_SESSION_TYPE = new Driver(22, (args: Stringable[]) => `Session type '${args[1]}' was not recognised.`);
export const MISSING_PORT = new Driver(23, (args: Stringable[]) => `Invalid URL '${args[1]}': missing port.`);
+ export const ADDRESS_TRANSLATION_MISMATCH = new Driver(24, (args: Stringable[]) => `Address translation map does not match the server's advertised address list. User-provided servers not in the advertised list: {${args[0]}}. Advertised servers not mapped by user: {${args[1]}}.`);
}
export class Concept extends ErrorMessage {
diff --git a/nodejs/connection/TypeDBDatabaseImpl.ts b/nodejs/connection/TypeDBDatabaseImpl.ts
index e3d3df21e8..afa17a4548 100644
--- a/nodejs/connection/TypeDBDatabaseImpl.ts
+++ b/nodejs/connection/TypeDBDatabaseImpl.ts
@@ -128,7 +128,7 @@ export class TypeDBDatabaseImpl implements Database {
async runOnAnyReplica(task: (serverDriver: ServerDriver, serverDatabase: ServerDatabase) => Promise): Promise {
for (const replica of this.replicas) {
try {
- return await task(this._driver.serverDrivers.get(replica.address), replica.database);
+ return await task(this._driver.serverDrivers.get(replica.server), replica.database);
} catch (e) {
if (e instanceof TypeDBDriverError && UNABLE_TO_CONNECT === e.messageTemplate) {
// TODO log
@@ -144,7 +144,7 @@ export class TypeDBDatabaseImpl implements Database {
}
for (const _ of Array(PRIMARY_REPLICA_TASK_MAX_RETRIES)) {
try {
- return await task(this._driver.serverDrivers.get(this.primaryReplica.address), this.primaryReplica.database);
+ return await task(this._driver.serverDrivers.get(this.primaryReplica.server), this.primaryReplica.database);
} catch (e) {
if (e instanceof TypeDBDriverError &&
(UNABLE_TO_CONNECT === e.messageTemplate || CLOUD_REPLICA_NOT_PRIMARY === e.messageTemplate)
@@ -176,14 +176,14 @@ export class TypeDBDatabaseImpl implements Database {
}
export class Replica implements Database.Replica {
- private readonly _address: string;
+ private readonly _server: string;
private readonly _database: ServerDatabase;
private readonly _term: number;
private readonly _isPrimary: boolean;
private readonly _isPreferred: boolean;
- private constructor(database: ServerDatabase, address: string, term: number, isPrimary: boolean, isPreferred: boolean) {
- this._address = address;
+ private constructor(database: ServerDatabase, server: string, term: number, isPrimary: boolean, isPreferred: boolean) {
+ this._server = server;
this._database = database;
this._term = term;
this._isPrimary = isPrimary;
@@ -198,8 +198,8 @@ export class Replica implements Database.Replica {
return this._database;
}
- get address(): string {
- return this._address;
+ get server(): string {
+ return this._server;
}
get databaseName(): string {
@@ -219,7 +219,7 @@ export class Replica implements Database.Replica {
}
toString(): string {
- return `${this._address}/${this.databaseName}:${this._isPrimary ? "P" : "S"}:${this._term}`;
+ return `${this._server}/${this.databaseName}:${this._isPrimary ? "P" : "S"}:${this._term}`;
}
}
diff --git a/nodejs/connection/TypeDBDriverImpl.ts b/nodejs/connection/TypeDBDriverImpl.ts
index fc6dcc1ddc..c2f227bd97 100644
--- a/nodejs/connection/TypeDBDriverImpl.ts
+++ b/nodejs/connection/TypeDBDriverImpl.ts
@@ -37,12 +37,13 @@ import CLOUD_UNABLE_TO_CONNECT = ErrorMessage.Driver.CLOUD_UNABLE_TO_CONNECT;
import SESSION_ID_EXISTS = ErrorMessage.Driver.SESSION_ID_EXISTS;
import UNABLE_TO_CONNECT = ErrorMessage.Driver.UNABLE_TO_CONNECT;
import MISSING_PORT = ErrorMessage.Driver.MISSING_PORT;
+import ADDRESS_TRANSLATION_MISMATCH = ErrorMessage.Driver.ADDRESS_TRANSLATION_MISMATCH;
export class TypeDBDriverImpl implements TypeDBDriver {
private _isOpen: boolean;
private readonly _isCloud: boolean;
- private readonly _initAddresses: string[];
+ private readonly _initAddresses: string[] | Record;
private readonly _credential: TypeDBCredential;
private _userManager: UserManagerImpl;
@@ -53,10 +54,10 @@ export class TypeDBDriverImpl implements TypeDBDriver {
private readonly _sessions: { [id: string]: TypeDBSessionImpl };
- constructor(addresses: string | string[], credential?: TypeDBCredential) {
+ constructor(addresses: string | string[] | Record, credential?: TypeDBCredential) {
if (typeof addresses === 'string') addresses = [addresses];
- for (const address of addresses)
+ for (const [_, address] of Object.entries(addresses))
if (!/:\d+/.test(address))
throw new TypeDBDriverError(MISSING_PORT.message(address));
@@ -77,7 +78,7 @@ export class TypeDBDriverImpl implements TypeDBDriver {
}
private async openCore(): Promise {
- const serverAddress = this._initAddresses[0];
+ const serverAddress = (this._initAddresses as string[])[0];
const serverStub = new TypeDBStubImpl(serverAddress, this._credential);
await serverStub.open();
const advertisedAddress = (await serverStub.serversAll(RequestBuilder.ServerManager.allReq())).servers[0].address;
@@ -89,10 +90,38 @@ export class TypeDBDriverImpl implements TypeDBDriver {
private async openCloud(): Promise {
const serverAddresses = await this.fetchCloudServerAddresses();
const openReqs: Promise[] = []
- for (const addr of serverAddresses) {
- const serverStub = new TypeDBStubImpl(addr, this._credential);
+
+ let addressTranslation: Record;
+ if (Array.isArray(this._initAddresses)) {
+ addressTranslation = {};
+ for (const address of serverAddresses) {
+ addressTranslation[address] = address;
+ }
+ } else {
+ addressTranslation = this._initAddresses;
+ const unknown = [];
+ for (const [advertised, _] of Object.entries(addressTranslation)) {
+ if (serverAddresses.indexOf(advertised) === -1) {
+ unknown.push(advertised);
+ }
+ }
+ const unmapped = [];
+ for (const advertisedAddress of serverAddresses) {
+ if (!(advertisedAddress in addressTranslation)) {
+ unmapped.push(advertisedAddress);
+ }
+ }
+ if (unknown.length > 0 || unmapped.length > 0) {
+ throw new TypeDBDriverError(
+ ADDRESS_TRANSLATION_MISMATCH.message(unknown.join(", "), unmapped.join(", "))
+ );
+ }
+ }
+
+ for (const [serverID, address] of Object.entries(addressTranslation)) {
+ const serverStub = new TypeDBStubImpl(address, this._credential);
openReqs.push(serverStub.open());
- this.serverDrivers.set(addr, new ServerDriver(addr, serverStub));
+ this.serverDrivers.set(serverID, new ServerDriver(address, serverStub));
}
try {
await Promise.any(openReqs);
@@ -105,7 +134,7 @@ export class TypeDBDriverImpl implements TypeDBDriver {
}
private async fetchCloudServerAddresses(): Promise {
- for (const address of this._initAddresses) {
+ for (const [_, address] of Object.entries(this._initAddresses)) {
try {
const stub = new TypeDBStubImpl(address, this._credential);
await stub.open();
@@ -116,7 +145,7 @@ export class TypeDBDriverImpl implements TypeDBDriver {
console.error(`Fetching cloud servers from ${address} failed.`, e);
}
}
- throw new TypeDBDriverError(CLOUD_UNABLE_TO_CONNECT.message(this._initAddresses.join(",")));
+ throw new TypeDBDriverError(CLOUD_UNABLE_TO_CONNECT.message(Object.values(this._initAddresses).join(",")));
}
isOpen(): boolean {
diff --git a/nodejs/docs/connection/Replica.adoc b/nodejs/docs/connection/Replica.adoc
index 3464de5cd9..e42d26308b 100644
--- a/nodejs/docs/connection/Replica.adoc
+++ b/nodejs/docs/connection/Replica.adoc
@@ -10,10 +10,10 @@ The metadata and state of an individual raft replica of a database.
[options="header"]
|===
|Name |Type |Description
-a| `address` a| `string` a| The address of the server hosting this replica
a| `databaseName` a| `string` a| The database for which this is a replica.
a| `preferred` a| `boolean` a| Checks whether this is the preferred replica of the raft cluster. If true, Operations which can be run on any replica will prefer to use this replica.
a| `primary` a| `boolean` a| Checks whether this is the primary replica of the raft cluster.
+a| `server` a| `string` a| The server hosting this replica
a| `term` a| `number` a| The raft protocol ‘term’ of this replica.
|===
// end::properties[]
diff --git a/nodejs/docs/connection/TypeDB.adoc b/nodejs/docs/connection/TypeDB.adoc
index 705cc9ad41..23e1f06596 100644
--- a/nodejs/docs/connection/TypeDB.adoc
+++ b/nodejs/docs/connection/TypeDB.adoc
@@ -13,7 +13,7 @@ a| `DEFAULT_ADDRESS`
// end::enum_constants[]
// tag::methods[]
-[#_TypeDB_cloudDriver__addresses_string__string____credential_TypeDBCredential]
+[#_TypeDB_cloudDriver__addresses_string__string____Record_string__string___credential_TypeDBCredential]
==== cloudDriver
[source,nodejs]
@@ -29,7 +29,7 @@ Creates a connection to TypeDB Cloud, authenticating with the provided credentia
[options="header"]
|===
|Name |Description |Type
-a| `addresses` a| List of addresses of the individual TypeDB Cloud servers. As long one specified address is provided, the driver will discover the other addresses from that server. a| `string \| string[]`
+a| `addresses` a| List of addresses of the individual TypeDB Cloud servers. As long one specified address is provided, the driver will discover the other addresses from that server. a| `string \| string[] \| Record`
a| `credential` a| The credentials to log in, and encryption settings. See ``TypeDBCredential``
Examples
``const driver = TypeDB.cloudDriver(["127.0.0.1:11729"], new TypeDBCredential(username, password));
diff --git a/nodejs/test/integration/test-cloud-failover.js b/nodejs/test/integration/test-cloud-failover.js
index 320d6a18a1..d766f7bb8a 100644
--- a/nodejs/test/integration/test-cloud-failover.js
+++ b/nodejs/test/integration/test-cloud-failover.js
@@ -115,7 +115,7 @@ async function run() {
);
primaryReplica = await seekPrimaryReplica(driver.databases);
console.info(`Stopping primary replica (test ${iteration}/10)...`);
- const port = primaryReplica.address.substring(10,15);
+ const port = primaryReplica.server.substring(10,15);
const primaryReplicaServerPID = getServerPID(port);
console.info(`Primary replica is hosted by server with PID ${primaryReplicaServerPID}`);
spawnSync("kill", ["-9", primaryReplicaServerPID]);
@@ -129,7 +129,7 @@ async function run() {
console.info(`Retrieved entity type with label '${person.label.scopedName}' from new primary replica`);
assert(person.label.scopedName === "person");
await session.close();
- const idx = primaryReplica.address[10];
+ const idx = primaryReplica.server[10];
serverStart(idx);
await new Promise(resolve => setTimeout(resolve, 20000));
const spawned = getServerPID(`${idx}1729`);
diff --git a/nodejs/test/integration/test-connection-cloud.js b/nodejs/test/integration/test-connection-cloud.js
new file mode 100644
index 0000000000..5d589c5f57
--- /dev/null
+++ b/nodejs/test/integration/test-connection-cloud.js
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+const { TypeDB, SessionType, TransactionType, TypeDBCredential } = require("../../dist");
+const assert = require("assert");
+
+async function run() {
+ try {
+ const driver = await TypeDB.cloudDriver(
+ {
+ "localhost:11729": "localhost:11729",
+ "localhost:21729": "localhost:21729",
+ "localhost:31729": "localhost:31729",
+ },
+ new TypeDBCredential("admin", "password", process.env.ROOT_CA)
+ );
+
+ const dbs = await driver.databases.all();
+ const typedb = dbs.find(x => x.name === "typedb");
+ if (typedb) {
+ await typedb.delete();
+ }
+ await driver.databases.create("typedb");
+ const session = await driver.session("typedb", SessionType.DATA);
+ const tx = await session.transaction(TransactionType.WRITE);
+
+ const root = await tx.concepts.getRootEntityType();
+ const subtypes = await root.getSubtypes(tx).collect();
+ assert(subtypes.length === 1);
+
+ await tx.close();
+ await session.close();
+ await driver.close();
+ } catch (err) {
+ console.error(`ERROR: ${err.stack || err}`);
+ process.exit(1);
+ }
+}
+
+run();
+
diff --git a/nodejs/test/integration/test-connection.js b/nodejs/test/integration/test-connection-core.js
similarity index 100%
rename from nodejs/test/integration/test-connection.js
rename to nodejs/test/integration/test-connection-core.js
diff --git a/python/docs/connection/Replica.adoc b/python/docs/connection/Replica.adoc
index 72e8c9568d..d686bd4f8a 100644
--- a/python/docs/connection/Replica.adoc
+++ b/python/docs/connection/Replica.adoc
@@ -4,20 +4,6 @@
The metadata and state of an individual raft replica of a database.
// tag::methods[]
-[#_Replica_address__]
-==== address
-
-[source,python]
-----
-address() -> str
-----
-
-Retrieves address of the server hosting this replica
-
-[caption=""]
-.Returns
-`str`
-
[#_Replica_database__]
==== database
@@ -60,6 +46,20 @@ Checks whether this is the primary replica of the raft cluster.
.Returns
`bool`
+[#_Replica_server__]
+==== server
+
+[source,python]
+----
+server() -> str
+----
+
+The server hosting this replica
+
+[caption=""]
+.Returns
+`str`
+
[#_Replica_term__]
==== term
diff --git a/python/docs/connection/TypeDB.adoc b/python/docs/connection/TypeDB.adoc
index 2c6a39f781..49c8a01fb4 100644
--- a/python/docs/connection/TypeDB.adoc
+++ b/python/docs/connection/TypeDB.adoc
@@ -2,12 +2,12 @@
=== TypeDB
// tag::methods[]
-[#_TypeDB_cloud_driver__addresses_Iterable_str___str__credential_TypeDBCredential]
+[#_TypeDB_cloud_driver__addresses_Mapping_str__str___Iterable_str___str__credential_TypeDBCredential]
==== cloud_driver
[source,python]
----
-static cloud_driver(addresses: Iterable[str] | str, credential: TypeDBCredential) -> TypeDBDriver
+static cloud_driver(addresses: Mapping[str, str] | Iterable[str] | str, credential: TypeDBCredential) -> TypeDBDriver
----
Creates a connection to TypeDB Cloud, authenticating with the provided credentials.
@@ -18,7 +18,7 @@ Creates a connection to TypeDB Cloud, authenticating with the provided credentia
[options="header"]
|===
|Name |Description |Type |Default Value
-a| `addresses` a| – a| `Iterable[str] \| str` a|
+a| `addresses` a| – a| `Mapping[str, str] \| Iterable[str] \| str` a|
a| `credential` a| – a| `TypeDBCredential` a|
|===
diff --git a/python/tests/integration/test_connection.py b/python/tests/integration/test_connection.py
index bf9720ca17..0ddff99154 100644
--- a/python/tests/integration/test_connection.py
+++ b/python/tests/integration/test_connection.py
@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
+import os
import unittest
from unittest import TestCase
@@ -22,6 +23,10 @@
from typedb.driver import *
+TYPEDB = "typedb"
+DATA = SessionType.DATA
+WRITE = TransactionType.WRITE
+
class TestDebug(TestCase):
@@ -29,6 +34,20 @@ def test_missing_port(self):
assert_that(calling(lambda: TypeDB.core_driver("localhost")), raises(TypeDBDriverException))
+ def test_address_translation(self):
+ address_translation = {
+ "localhost:11729": "localhost:11729",
+ "localhost:21729": "localhost:21729",
+ "localhost:31729": "localhost:31729"
+ }
+ credential = TypeDBCredential("admin", "password", tls_enabled=True, tls_root_ca_path=os.environ["ROOT_CA"])
+ with TypeDB.cloud_driver(address_translation, credential) as driver:
+ if TYPEDB not in [db.name for db in driver.databases.all()]:
+ driver.databases.create(TYPEDB)
+ with driver.session(TYPEDB, DATA) as session, session.transaction(WRITE) as tx:
+ root = tx.concepts.get_root_entity_type()
+ assert_that(len(list(root.get_subtypes(tx))), equal_to(1))
+
if __name__ == "__main__":
unittest.main(verbosity=2)
diff --git a/python/typedb/api/connection/database.py b/python/typedb/api/connection/database.py
index ee16b60ec8..923edaae81 100644
--- a/python/typedb/api/connection/database.py
+++ b/python/typedb/api/connection/database.py
@@ -154,9 +154,9 @@ def database(self) -> Database:
pass
@abstractmethod
- def address(self) -> str:
+ def server(self) -> str:
"""
- Retrieves address of the server hosting this replica
+ The server hosting this replica
:return:
"""
diff --git a/python/typedb/connection/database.py b/python/typedb/connection/database.py
index e37e29cce5..b71d84d231 100644
--- a/python/typedb/connection/database.py
+++ b/python/typedb/connection/database.py
@@ -20,7 +20,7 @@
from typing import Optional
from typedb.native_driver_wrapper import database_get_name, database_schema, database_delete, database_rule_schema, \
- database_type_schema, ReplicaInfo, replica_info_get_address, replica_info_is_primary, replica_info_is_preferred, \
+ database_type_schema, ReplicaInfo, replica_info_get_server, replica_info_is_primary, replica_info_is_preferred, \
replica_info_get_term, database_get_replicas_info, database_get_primary_replica_info, \
database_get_preferred_replica_info, replica_info_iterator_next, Database as NativeDatabase, \
TypeDBDriverExceptionNative
@@ -105,8 +105,8 @@ def __init__(self, replica_info: ReplicaInfo):
def database(self) -> Database:
pass
- def address(self) -> str:
- return replica_info_get_address(self._info)
+ def server(self) -> str:
+ return replica_info_get_server(self._info)
def is_primary(self) -> bool:
return replica_info_is_primary(self._info)
diff --git a/python/typedb/connection/driver.py b/python/typedb/connection/driver.py
index f3ae5e243f..bbaa52eb98 100644
--- a/python/typedb/connection/driver.py
+++ b/python/typedb/connection/driver.py
@@ -19,8 +19,8 @@
from typing import Optional, TYPE_CHECKING
-from typedb.native_driver_wrapper import connection_open_core, connection_open_cloud, connection_is_open, \
- connection_force_close, Connection as NativeConnection, TypeDBDriverExceptionNative
+from typedb.native_driver_wrapper import connection_open_core, connection_open_cloud, connection_open_cloud_translated, \
+ connection_is_open, connection_force_close, Connection as NativeConnection, TypeDBDriverExceptionNative
from typedb.api.connection.driver import TypeDBDriver
from typedb.api.connection.options import TypeDBOptions
@@ -38,10 +38,16 @@
class _Driver(TypeDBDriver, NativeWrapper[NativeConnection]):
- def __init__(self, addresses: list[str], credential: Optional[TypeDBCredential] = None):
+ def __init__(self, addresses: list[str] | dict[str], credential: Optional[TypeDBCredential] = None):
if credential:
try:
- native_connection = connection_open_cloud(addresses, credential.native_object)
+ if isinstance(addresses, list):
+ native_connection = connection_open_cloud(addresses, credential.native_object)
+ else:
+ advertised_addresses = list(addresses.keys())
+ translated_addresses = [addresses[advertised] for advertised in advertised_addresses]
+ native_connection = connection_open_cloud_translated(
+ advertised_addresses, translated_addresses, credential.native_object)
except TypeDBDriverExceptionNative as e:
raise TypeDBDriverException.of(e)
else:
diff --git a/python/typedb/driver.py b/python/typedb/driver.py
index 1acaaf5ff5..45a00413f1 100644
--- a/python/typedb/driver.py
+++ b/python/typedb/driver.py
@@ -15,7 +15,8 @@
# specific language governing permissions and limitations
# under the License.
-from typing import Iterable, Union
+from collections.abc import Iterable, Mapping
+from typing import Union
from typedb.api.answer.concept_map import * # noqa # pylint: disable=unused-import
from typedb.api.answer.concept_map_group import * # noqa # pylint: disable=unused-import
@@ -67,7 +68,7 @@ def core_driver(address: str) -> TypeDBDriver:
return _Driver([address])
@staticmethod
- def cloud_driver(addresses: Union[Iterable[str], str], credential: TypeDBCredential) -> TypeDBDriver:
+ def cloud_driver(addresses: Union[Mapping[str, str], Iterable[str], str], credential: TypeDBCredential) -> TypeDBDriver:
"""
Creates a connection to TypeDB Cloud, authenticating with the provided credentials.
@@ -77,5 +78,7 @@ def cloud_driver(addresses: Union[Iterable[str], str], credential: TypeDBCredent
"""
if isinstance(addresses, str):
return _Driver([addresses], credential)
+ elif isinstance(addresses, Mapping):
+ return _Driver(dict(addresses), credential)
else:
return _Driver(list(addresses), credential)
diff --git a/rust/docs/connection/Connection.adoc b/rust/docs/connection/Connection.adoc
index 4a63a18e27..639d3222a1 100644
--- a/rust/docs/connection/Connection.adoc
+++ b/rust/docs/connection/Connection.adoc
@@ -128,6 +128,54 @@ Connection::new_cloud(
)
----
+[#_struct_Connection_new_cloud_with_translation__address_translation_HashMap_T__credential_Credential]
+==== new_cloud_with_translation
+
+[source,rust]
+----
+pub fn new_cloud_with_translation(
+ address_translation: HashMap,
+ credential: Credential
+) -> Resultwhere
+ T: AsRef + Sync,
+ U: AsRef + Sync,
+----
+
+Creates a new TypeDB Cloud connection.
+
+[caption=""]
+.Input parameters
+[cols=",,"]
+[options="header"]
+|===
+|Name |Description |Type
+a| `address_translation` a| Translation map from addresses received from the TypeDB server(s) to addresses to be used by the driver for connection a| `HashMapwhere
+ T: AsRef + Sync,
+ U: AsRef + Sync,
+----
+
+[caption=""]
+.Code examples
+[source,rust]
+----
+Connection::new_cloud_with_translation(
+ [
+ ("typedb-cloud.ext:11729", "localhost:11729"),
+ ("typedb-cloud.ext:21729", "localhost:21729"),
+ ("typedb-cloud.ext:31729", "localhost:31729"),
+ ].into(),
+ credential,
+)
+----
+
[#_struct_Connection_new_core__address_impl_AsRef_str_]
==== new_core
diff --git a/rust/docs/connection/ReplicaInfo.adoc b/rust/docs/connection/ReplicaInfo.adoc
index 4afca870f3..f84cf2439f 100644
--- a/rust/docs/connection/ReplicaInfo.adoc
+++ b/rust/docs/connection/ReplicaInfo.adoc
@@ -14,9 +14,9 @@ The metadata and state of an individual raft replica of a database.
[options="header"]
|===
|Name |Type |Description
-a| `address` a| `Address` a| The address of the server hosting this replica
a| `is_preferred` a| `bool` a| Whether this is the preferred replica of the raft cluster. If true, Operations which can be run on any replica will prefer to use this replica.
a| `is_primary` a| `bool` a| Whether this is the primary replica of the raft cluster.
+a| `server` a| `String` a| The server hosting this replica
a| `term` a| `i64` a| The raft protocol ‘term’ of this replica.
|===
// end::properties[]
diff --git a/rust/docs/errors/ConnectionError.adoc b/rust/docs/errors/ConnectionError.adoc
index a054d17d61..4d243d8d39 100644
--- a/rust/docs/errors/ConnectionError.adoc
+++ b/rust/docs/errors/ConnectionError.adoc
@@ -8,6 +8,7 @@
[options="header"]
|===
|Variant
+a| `AddressTranslationMismatch`
a| `BrokenPipe`
a| `CloudAllNodesFailed`
a| `CloudEndpointEncrypted`
diff --git a/rust/docs/errors/InternalError.adoc b/rust/docs/errors/InternalError.adoc
index a59d2cc829..78f7c480b5 100644
--- a/rust/docs/errors/InternalError.adoc
+++ b/rust/docs/errors/InternalError.adoc
@@ -13,7 +13,7 @@ a| `RecvError`
a| `SendError`
a| `UnexpectedRequestType`
a| `UnexpectedResponseType`
-a| `UnknownConnectionAddress`
+a| `UnknownServer`
|===
// end::enum_constants[]
diff --git a/rust/src/common/error.rs b/rust/src/common/error.rs
index d141e3d23a..359fab7ac0 100644
--- a/rust/src/common/error.rs
+++ b/rust/src/common/error.rs
@@ -17,12 +17,12 @@
* under the License.
*/
-use std::{error::Error as StdError, fmt};
+use std::{collections::HashSet, error::Error as StdError, fmt};
use tonic::{Code, Status};
use typeql::error_messages;
-use super::{address::Address, RequestID};
+use super::RequestID;
error_messages! { ConnectionError
code: "CXN", type: "Connection Error",
@@ -72,6 +72,8 @@ error_messages! { ConnectionError
22: "Connection failed. Please check the server is running and the address is accessible. Encrypted Cloud endpoints may also have misconfigured SSL certificates.",
MissingPort { address: String } =
23: "Invalid URL '{address}': missing port.",
+ AddressTranslationMismatch { unknown: HashSet, unmapped: HashSet } =
+ 24: "Address translation map does not match the server's advertised address list. User-provided servers not in the advertised list: {unknown:?}. Advertised servers not mapped by user: {unmapped:?}.",
}
error_messages! { InternalError
@@ -84,8 +86,8 @@ error_messages! { InternalError
3: "Unexpected request type for remote procedure call: {request_type}.",
UnexpectedResponseType { response_type: String } =
4: "Unexpected response type for remote procedure call: {response_type}.",
- UnknownConnectionAddress { address: Address } =
- 5: "Received unrecognized address from the server: {address}.",
+ UnknownServer { server: String } =
+ 5: "Received replica at unrecognized server: {server}.",
EnumOutOfBounds { value: i32, enum_name: &'static str } =
6: "Value '{value}' is out of bounds for enum '{enum_name}'.",
}
diff --git a/rust/src/common/info.rs b/rust/src/common/info.rs
index 2ce6b74f33..3122fa050a 100644
--- a/rust/src/common/info.rs
+++ b/rust/src/common/info.rs
@@ -21,12 +21,10 @@ use std::time::Duration;
use tokio::sync::mpsc::UnboundedSender;
-use super::{address::Address, SessionID};
-use crate::common::Callback;
+use super::{Callback, SessionID};
#[derive(Clone, Debug)]
pub(crate) struct SessionInfo {
- pub(crate) address: Address,
pub(crate) session_id: SessionID,
pub(crate) network_latency: Duration,
pub(crate) on_close_register_sink: UnboundedSender,
@@ -41,8 +39,8 @@ pub(crate) struct DatabaseInfo {
/// The metadata and state of an individual raft replica of a database.
#[derive(Debug)]
pub struct ReplicaInfo {
- /// The address of the server hosting this replica
- pub address: Address,
+ /// The server hosting this replica
+ pub server: String,
/// Whether this is the primary replica of the raft cluster.
pub is_primary: bool,
/// Whether this is the preferred replica of the raft cluster.
diff --git a/rust/src/connection/connection.rs b/rust/src/connection/connection.rs
index 29f636d1db..d664e98aef 100644
--- a/rust/src/connection/connection.rs
+++ b/rust/src/connection/connection.rs
@@ -57,7 +57,7 @@ use crate::{
/// A connection to a TypeDB server which serves as the starting point for all interaction.
#[derive(Clone)]
pub struct Connection {
- server_connections: HashMap,
+ server_connections: HashMap,
background_runtime: Arc,
username: Option,
is_cloud: bool,
@@ -76,18 +76,21 @@ impl Connection {
/// Connection::new_core("127.0.0.1:1729")
/// ```
pub fn new_core(address: impl AsRef) -> Result {
- let address: Address = address.as_ref().parse()?;
+ let id = address.as_ref().to_string();
+ let address: Address = id.parse()?;
let background_runtime = Arc::new(BackgroundRuntime::new()?);
- let mut server_connection = ServerConnection::new_core(background_runtime.clone(), address)?;
- let address = server_connection
+ let server_connection = ServerConnection::new_core(background_runtime.clone(), address)?;
+
+ let advertised_id = server_connection
.servers_all()?
.into_iter()
.exactly_one()
- .map_err(|e| ConnectionError::ServerConnectionFailedStatusError { error: e.to_string() })?;
- server_connection.set_address(address.clone());
+ .map_err(|e| ConnectionError::ServerConnectionFailedStatusError { error: e.to_string() })?
+ .to_string();
+
match server_connection.validate() {
Ok(()) => Ok(Self {
- server_connections: [(address, server_connection)].into(),
+ server_connections: [(advertised_id, server_connection)].into(),
background_runtime,
username: None,
is_cloud: false,
@@ -121,14 +124,79 @@ impl Connection {
pub fn new_cloud + Sync>(init_addresses: &[T], credential: Credential) -> Result {
let background_runtime = Arc::new(BackgroundRuntime::new()?);
- let init_addresses = init_addresses.iter().map(|addr| addr.as_ref().parse()).try_collect()?;
- let addresses = Self::fetch_current_addresses(background_runtime.clone(), init_addresses, credential.clone())?;
+ let servers = Self::fetch_server_list(background_runtime.clone(), init_addresses, credential.clone())?;
- let server_connections: HashMap = addresses
+ let server_to_address: HashMap = servers
.into_iter()
.map(|address| {
- ServerConnection::new_cloud(background_runtime.clone(), address.clone(), credential.clone())
- .map(|server_connection| (address, server_connection))
+ let id = address.clone();
+ address.parse().map(|address| (id, address))
+ })
+ .try_collect()?;
+
+ Self::new_cloud_impl(server_to_address, background_runtime, credential)
+ }
+
+ /// Creates a new TypeDB Cloud connection.
+ ///
+ /// # Arguments
+ ///
+ /// * `address_translation` -- Translation map from addresses received from the TypeDB server(s)
+ /// to addresses to be used by the driver for connection
+ /// * `credential` -- User credential and TLS encryption setting
+ ///
+ /// # Examples
+ ///
+ /// ```rust
+ /// Connection::new_cloud_with_translation(
+ /// [
+ /// ("typedb-cloud.ext:11729", "localhost:11729"),
+ /// ("typedb-cloud.ext:21729", "localhost:21729"),
+ /// ("typedb-cloud.ext:31729", "localhost:31729"),
+ /// ].into(),
+ /// credential,
+ /// )
+ /// ```
+ pub fn new_cloud_with_translation(address_translation: HashMap, credential: Credential) -> Result
+ where
+ T: AsRef + Sync,
+ U: AsRef + Sync,
+ {
+ let background_runtime = Arc::new(BackgroundRuntime::new()?);
+
+ let servers =
+ Self::fetch_server_list(background_runtime.clone(), address_translation.values(), credential.clone())?;
+
+ let server_to_address: HashMap = address_translation
+ .into_iter()
+ .map(|(id, address)| {
+ let id = id.as_ref().to_owned();
+ address.as_ref().parse().map(|address| (id, address))
+ })
+ .try_collect()?;
+
+ let translated: HashSet = server_to_address.keys().cloned().collect();
+ let unknown = &translated - &servers;
+ let unmapped = &servers - &translated;
+ if !unknown.is_empty() || !unmapped.is_empty() {
+ return Err(ConnectionError::AddressTranslationMismatch { unknown, unmapped }.into());
+ }
+
+ debug_assert_eq!(servers, translated);
+
+ Self::new_cloud_impl(server_to_address, background_runtime, credential)
+ }
+
+ fn new_cloud_impl(
+ server_to_address: HashMap,
+ background_runtime: Arc,
+ credential: Credential,
+ ) -> Result {
+ let server_connections: HashMap = server_to_address
+ .into_iter()
+ .map(|(server_id, address)| {
+ ServerConnection::new_cloud(background_runtime.clone(), address, credential.clone())
+ .map(|server_connection| (server_id, server_connection))
})
.try_collect()?;
@@ -142,20 +210,20 @@ impl Connection {
Ok(Self {
server_connections,
background_runtime,
- username: Some(credential.username().to_string()),
+ username: Some(credential.username().to_owned()),
is_cloud: true,
})
}
}
- fn fetch_current_addresses(
+ fn fetch_server_list(
background_runtime: Arc,
- addresses: Vec,
+ addresses: impl IntoIterator- > + Clone,
credential: Credential,
- ) -> Result> {
- for address in &addresses {
+ ) -> Result> {
+ for address in addresses.clone() {
let server_connection =
- ServerConnection::new_cloud(background_runtime.clone(), address.clone(), credential.clone());
+ ServerConnection::new_cloud(background_runtime.clone(), address.as_ref().parse()?, credential.clone());
match server_connection {
Ok(server_connection) => match server_connection.servers_all() {
Ok(servers) => return Ok(servers.into_iter().collect()),
@@ -171,7 +239,7 @@ impl Connection {
}
}
Err(ConnectionError::ServerConnectionFailed {
- addresses: addresses.into_iter().map(|a| a.to_string()).collect::>().join(","),
+ addresses: addresses.into_iter().map(|addr| addr.as_ref().to_owned()).join(", "),
}
.into())
}
@@ -215,27 +283,25 @@ impl Connection {
self.server_connections.len()
}
- pub(crate) fn addresses(&self) -> impl Iterator
- {
- self.server_connections.keys()
+ pub(crate) fn servers(&self) -> impl Iterator
- {
+ self.server_connections.keys().map(String::as_str)
}
- pub(crate) fn connection(&self, address: &Address) -> Result<&ServerConnection> {
- self.server_connections
- .get(address)
- .ok_or_else(|| InternalError::UnknownConnectionAddress { address: address.clone() }.into())
+ pub(crate) fn connection(&self, id: &str) -> Option<&ServerConnection> {
+ self.server_connections.get(id)
}
- pub(crate) fn connections(&self) -> impl Iterator
- + '_ {
- self.server_connections.values()
+ pub(crate) fn connections(&self) -> impl Iterator
- + '_ {
+ self.server_connections.iter().map(|(id, conn)| (id.as_str(), conn))
}
pub(crate) fn username(&self) -> Option<&str> {
- self.username.as_ref().map(|s| s.as_ref())
+ self.username.as_deref()
}
pub(crate) fn unable_to_connect_error(&self) -> Error {
Error::Connection(ConnectionError::ServerConnectionFailedStatusError {
- error: self.addresses().map(Address::to_string).collect::>().join(", "),
+ error: self.servers().map(str::to_owned).collect_vec().join(", "),
})
}
}
@@ -248,7 +314,6 @@ impl fmt::Debug for Connection {
#[derive(Clone)]
pub(crate) struct ServerConnection {
- address: Address,
background_runtime: Arc,
open_sessions: Arc>>>,
request_transmitter: Arc,
@@ -256,14 +321,13 @@ pub(crate) struct ServerConnection {
impl ServerConnection {
fn new_core(background_runtime: Arc, address: Address) -> Result {
- let request_transmitter = Arc::new(RPCTransmitter::start_core(address.clone(), &background_runtime)?);
- Ok(Self { address, background_runtime, open_sessions: Default::default(), request_transmitter })
+ let request_transmitter = Arc::new(RPCTransmitter::start_core(address, &background_runtime)?);
+ Ok(Self { background_runtime, open_sessions: Default::default(), request_transmitter })
}
fn new_cloud(background_runtime: Arc, address: Address, credential: Credential) -> Result {
- let request_transmitter =
- Arc::new(RPCTransmitter::start_cloud(address.clone(), credential, &background_runtime)?);
- Ok(Self { address, background_runtime, open_sessions: Default::default(), request_transmitter })
+ let request_transmitter = Arc::new(RPCTransmitter::start_cloud(address, credential, &background_runtime)?);
+ Ok(Self { background_runtime, open_sessions: Default::default(), request_transmitter })
}
pub(crate) fn validate(&self) -> Result {
@@ -273,14 +337,6 @@ impl ServerConnection {
}
}
- fn set_address(&mut self, address: Address) {
- self.address = address;
- }
-
- pub(crate) fn address(&self) -> &Address {
- &self.address
- }
-
#[cfg_attr(feature = "sync", maybe_async::must_be_sync)]
async fn request(&self, request: Request) -> Result {
if !self.background_runtime.is_open() {
@@ -304,7 +360,7 @@ impl ServerConnection {
self.request_transmitter.force_close()
}
- pub(crate) fn servers_all(&self) -> Result> {
+ pub(crate) fn servers_all(&self) -> Result> {
match self.request_blocking(Request::ServersAll)? {
Response::ServersAll { servers } => Ok(servers),
other => Err(InternalError::UnexpectedResponseType { response_type: format!("{other:?}") }.into()),
@@ -392,7 +448,6 @@ impl ServerConnection {
pulse_shutdown_source,
));
Ok(SessionInfo {
- address: self.address.clone(),
session_id,
network_latency: start.elapsed().saturating_sub(server_duration),
on_close_register_sink,
@@ -506,10 +561,7 @@ impl ServerConnection {
impl fmt::Debug for ServerConnection {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
- f.debug_struct("ServerConnection")
- .field("address", &self.address)
- .field("open_sessions", &self.open_sessions)
- .finish()
+ f.debug_struct("ServerConnection").field("open_sessions", &self.open_sessions).finish()
}
}
diff --git a/rust/src/connection/message.rs b/rust/src/connection/message.rs
index a710dd748f..03bf77f2d7 100644
--- a/rust/src/connection/message.rs
+++ b/rust/src/connection/message.rs
@@ -26,7 +26,7 @@ use typeql::pattern::{Conjunction, Statement};
use crate::{
answer::{readable_concept, ConceptMap, ConceptMapGroup, ValueGroup},
- common::{address::Address, info::DatabaseInfo, RequestID, SessionID, IID},
+ common::{info::DatabaseInfo, RequestID, SessionID, IID},
concept::{
Annotation, Attribute, AttributeType, Entity, EntityType, Relation, RelationType, RoleType, SchemaException,
Thing, ThingType, Transitivity, Value, ValueType,
@@ -73,7 +73,7 @@ pub(super) enum Response {
ConnectionOpen,
ServersAll {
- servers: Vec,
+ servers: Vec,
},
DatabasesContains {
diff --git a/rust/src/connection/network/proto/database.rs b/rust/src/connection/network/proto/database.rs
index 471d116619..c9d9f0734a 100644
--- a/rust/src/connection/network/proto/database.rs
+++ b/rust/src/connection/network/proto/database.rs
@@ -17,31 +17,24 @@
* under the License.
*/
-use itertools::Itertools;
use typedb_protocol::{database_replicas::Replica as ReplicaProto, DatabaseReplicas as DatabaseProto};
-use super::TryFromProto;
-use crate::{
- common::info::{DatabaseInfo, ReplicaInfo},
- Result,
-};
+use super::FromProto;
+use crate::common::info::{DatabaseInfo, ReplicaInfo};
-impl TryFromProto for DatabaseInfo {
- fn try_from_proto(proto: DatabaseProto) -> Result {
- Ok(Self {
- name: proto.name,
- replicas: proto.replicas.into_iter().map(ReplicaInfo::try_from_proto).try_collect()?,
- })
+impl FromProto for DatabaseInfo {
+ fn from_proto(proto: DatabaseProto) -> Self {
+ Self { name: proto.name, replicas: proto.replicas.into_iter().map(ReplicaInfo::from_proto).collect() }
}
}
-impl TryFromProto for ReplicaInfo {
- fn try_from_proto(proto: ReplicaProto) -> Result {
- Ok(Self {
- address: proto.address.parse()?,
+impl FromProto for ReplicaInfo {
+ fn from_proto(proto: ReplicaProto) -> Self {
+ Self {
+ server: proto.address, // TODO should be eventually replaced by "server_id" or "server_name" in protocol
is_primary: proto.primary,
is_preferred: proto.preferred,
term: proto.term,
- })
+ }
}
}
diff --git a/rust/src/connection/network/proto/message.rs b/rust/src/connection/network/proto/message.rs
index 7adad840f5..a94cc152bd 100644
--- a/rust/src/connection/network/proto/message.rs
+++ b/rust/src/connection/network/proto/message.rs
@@ -245,10 +245,10 @@ impl FromProto for Response {
}
}
-impl TryFromProto for Response {
- fn try_from_proto(proto: server_manager::all::Res) -> Result {
- let servers = proto.servers.into_iter().map(|server| server.address.parse()).try_collect()?;
- Ok(Self::ServersAll { servers })
+impl FromProto for Response {
+ fn from_proto(proto: server_manager::all::Res) -> Self {
+ let servers = proto.servers.into_iter().map(|server| server.address).collect();
+ Self::ServersAll { servers }
}
}
@@ -267,18 +267,16 @@ impl FromProto for Response {
impl TryFromProto for Response {
fn try_from_proto(proto: database_manager::get::Res) -> Result {
Ok(Self::DatabaseGet {
- database: DatabaseInfo::try_from_proto(
+ database: DatabaseInfo::from_proto(
proto.database.ok_or(ConnectionError::MissingResponseField { field: "database" })?,
- )?,
+ ),
})
}
}
-impl TryFromProto for Response {
- fn try_from_proto(proto: database_manager::all::Res) -> Result {
- Ok(Self::DatabasesAll {
- databases: proto.databases.into_iter().map(DatabaseInfo::try_from_proto).try_collect()?,
- })
+impl FromProto for Response {
+ fn from_proto(proto: database_manager::all::Res) -> Self {
+ Self::DatabasesAll { databases: proto.databases.into_iter().map(DatabaseInfo::from_proto).collect() }
}
}
diff --git a/rust/src/connection/network/transmitter/rpc.rs b/rust/src/connection/network/transmitter/rpc.rs
index a867d305f3..2d66aa7ea0 100644
--- a/rust/src/connection/network/transmitter/rpc.rs
+++ b/rust/src/connection/network/transmitter/rpc.rs
@@ -121,7 +121,7 @@ impl RPCTransmitter {
match request {
Request::ConnectionOpen => rpc.connection_open(request.try_into_proto()?).await.map(Response::from_proto),
- Request::ServersAll => rpc.servers_all(request.try_into_proto()?).await.and_then(Response::try_from_proto),
+ Request::ServersAll => rpc.servers_all(request.try_into_proto()?).await.map(Response::from_proto),
Request::DatabasesContains { .. } => {
rpc.databases_contains(request.try_into_proto()?).await.map(Response::from_proto)
@@ -132,9 +132,7 @@ impl RPCTransmitter {
Request::DatabaseGet { .. } => {
rpc.databases_get(request.try_into_proto()?).await.and_then(Response::try_from_proto)
}
- Request::DatabasesAll => {
- rpc.databases_all(request.try_into_proto()?).await.and_then(Response::try_from_proto)
- }
+ Request::DatabasesAll => rpc.databases_all(request.try_into_proto()?).await.map(Response::from_proto),
Request::DatabaseDelete { .. } => {
rpc.database_delete(request.try_into_proto()?).await.map(Response::from_proto)
diff --git a/rust/src/database/database.rs b/rust/src/database/database.rs
index a8d8e8a178..8cca837e89 100644
--- a/rust/src/database/database.rs
+++ b/rust/src/database/database.rs
@@ -21,17 +21,16 @@
use std::future::Future;
use std::{fmt, sync::RwLock, thread::sleep, time::Duration};
-use itertools::Itertools;
use log::{debug, error};
use crate::{
common::{
- address::Address,
error::ConnectionError,
info::{DatabaseInfo, ReplicaInfo},
Error, Result,
},
connection::ServerConnection,
+ error::InternalError,
Connection,
};
@@ -104,10 +103,6 @@ impl Database {
self.preferred_replica().map(|replica| replica.to_info())
}
- pub(super) fn connection(&self) -> &Connection {
- &self.connection
- }
-
/// Deletes this database.
///
/// # Examples
@@ -118,7 +113,7 @@ impl Database {
/// ```
#[cfg_attr(feature = "sync", maybe_async::must_be_sync)]
pub async fn delete(self) -> Result {
- self.run_on_primary_replica(|database, _| database.delete()).await
+ self.run_on_primary_replica(|database| database.delete()).await
}
/// Returns a full schema text as a valid TypeQL define query string.
@@ -131,7 +126,7 @@ impl Database {
/// ```
#[cfg_attr(feature = "sync", maybe_async::must_be_sync)]
pub async fn schema(&self) -> Result {
- self.run_failsafe(|database, _| async move { database.schema().await }).await
+ self.run_failsafe(|database| async move { database.schema().await }).await
}
/// Returns the types in the schema as a valid TypeQL define query string.
@@ -144,7 +139,7 @@ impl Database {
/// ```
#[cfg_attr(feature = "sync", maybe_async::must_be_sync)]
pub async fn type_schema(&self) -> Result {
- self.run_failsafe(|database, _| async move { database.type_schema().await }).await
+ self.run_failsafe(|database| async move { database.type_schema().await }).await
}
/// Returns the rules in the schema as a valid TypeQL define query string.
@@ -157,13 +152,13 @@ impl Database {
/// ```
#[cfg_attr(feature = "sync", maybe_async::must_be_sync)]
pub async fn rule_schema(&self) -> Result {
- self.run_failsafe(|database, _| async move { database.rule_schema().await }).await
+ self.run_failsafe(|database| async move { database.rule_schema().await }).await
}
#[cfg_attr(feature = "sync", maybe_async::must_be_sync)]
pub(crate) async fn run_failsafe(&self, task: F) -> Result
where
- F: Fn(ServerDatabase, ServerConnection) -> P,
+ F: Fn(ServerDatabase) -> P,
P: Future