resolver

Author: rlubke

etc/docker-compose-2-members.yaml

@@ -19,6 +19,7 @@ services:
       - "9612:9612"
       - "8080:8080"
       - "6676:6676"
+      - "7574:7574"
     volumes:
       - .:/args
       - ./cert:/certs

src/nslookup.ts

@@ -0,0 +1,354 @@
+/*
+ * Copyright (c) 2025, Oracle and/or its affiliates.
+ *
+ * Licensed under the Universal Permissive License v 1.0 as shown at
+ * https://oss.oracle.com/licenses/upl.
+ */
+
+import * as net from 'net';
+import { promisify } from 'util';
+import { randomBytes } from 'crypto';
+
+// Constants
+const DEFAULT_PORT = 7574;
+const DEFAULT_HOST = 'localhost';
+const CLUSTER_NAME_LOOKUP = 'Cluster/name';
+const CLUSTER_INFO_LOOKUP = 'Cluster/info';
+const CLUSTER_FOREIGN_LOOKUP = 'Cluster/foreign';
+const MANAGEMENT_LOOKUP = 'management/HTTPManagementURL';
+const JMX_LOOKUP = 'management/JMXServiceURL';
+const METRICS_LOOKUP = 'metrics/HTTPMetricsURL';
+const GRPC_PROXY_LOOKUP = '$GRPC:GrpcProxy';
+const NS_PREFIX = 'NameService/string/';
+const NS_LOCAL_PORT = '/NameService/localPort';
+const DEFAULT_TIMEOUT = 10;
+
+const MULTIPLEXED_SOCKET = Buffer.from([90, 193, 224, 0]);
+const NAME_SERVICE_SUB_PORT = Buffer.from([0, 0, 0, 3]);
+const CONNECTION_OPEN = Buffer.from([
+    0, 1, 2, 0, 66, 0, 1, 14, 0, 0, 66, 166, 182, 159, 222, 178, 81,
+    1, 65, 227, 243, 228, 221, 15, 2, 65, 143, 246, 186, 153, 1, 3,
+    65, 248, 180, 229, 242, 4, 4, 65, 196, 254, 220, 245, 5, 5, 65, 215,
+    206, 195, 141, 7, 6, 65, 219, 137, 220, 213, 10, 64, 2, 110, 3,
+    93, 78, 87, 2, 17, 77, 101, 115, 115, 97, 103, 105, 110, 103, 80,
+    114, 111, 116, 111, 99, 111, 108, 2, 65, 2, 65, 2, 19, 78, 97, 109,
+    101, 83, 101, 114, 118, 105, 99, 101, 80, 114, 111, 116, 111, 99,
+    111, 108, 2, 65, 1, 65, 1, 5, 160, 2, 0, 0, 14, 0, 0, 66, 174, 137,
+    158, 222, 178, 81, 1, 65, 129, 128, 128, 240, 15, 5, 65, 152, 159,
+    129, 128, 8, 6, 65, 147, 158, 1, 64, 1, 106, 2, 110, 3, 106, 4, 113,
+    5, 113, 6, 78, 8, 67, 108, 117, 115, 116, 101, 114, 66, 9, 78, 9, 108,
+    111, 99, 97, 108, 104, 111, 115, 116, 10, 78, 5, 50, 48, 50, 51, 51, 12,
+    78, 16, 67, 111, 104, 101, 114, 101, 110, 99, 101, 67, 111, 110, 115,
+    111, 108, 101, 64, 64,
+]);
+
+const CHANNEL_OPEN = Buffer.from([
+    0, 11, 2, 0, 66, 1, 1, 78, 19, 78, 97, 109, 101, 83, 101, 114, 118,
+    105, 99, 101, 80, 114, 111, 116, 111, 99, 111, 108, 2, 78, 11, 78,
+    97, 109, 101, 83, 101, 114, 118, 105, 99, 101, 64,
+]);
+
+const NS_LOOKUP_REQ_ID = Buffer.from([1, 1, 0, 66, 0, 1, 78]);
+const REQ_END_MARKER = Buffer.from([64]);
+
+function writePackedInt(n: number): Buffer {
+    let result: Buffer = Buffer.alloc(0);
+    let b = 0;
+
+    if (n < 0) {
+        b = 0x40;
+        n = ~n; // bitwise negation
+    }
+
+    b |= n & 0x3F;
+    n >>= 6;
+
+    while (n !== 0) {
+        result = Buffer.concat([result, Buffer.from([b | 0x80])]);
+        b = n & 0x7F;
+        n >>= 7;
+    }
+
+    result = Buffer.concat([result, Buffer.from([b])]);
+    return result;
+}
+
+
+// Helper function to parse results from a string response
+function parseResults(results: string): string[] {
+    if (results.charAt(0) == '[' && results.charAt(results.length -1 ) == ']')
+    results = results.trim().slice(1, -1); // Remove brackets
+    return results ? results.split(', ') : [];
+}
+
+function sleep(ms: number) {
+    return new Promise((resolve) => {
+        setTimeout(resolve, ms);
+    });
+}
+
+// Class representing a discovered cluster
+class DiscoveredCluster {
+    cluster_name: string = '';
+    connection_name: string = '';
+    ns_port: number = 0;
+    host: string = '';
+    management_urls: string[] = [];
+    selected_url: string = '';
+    metrics_urls: string[] = [];
+    jmx_urls: string[] = [];
+    grpc_proxy_endpoints: string[] = [];
+}
+
+// Class representing a ClusterNSPort
+class ClusterNSPort {
+    host_name: string = '';
+    cluster_name: string = '';
+    port: number = 0;
+    is_local: boolean = false;
+
+    constructor(cluster_name = '', port = 0, host_name = '', is_local = false) {
+        this.host_name = host_name;
+        this.cluster_name = cluster_name;
+        this.port = port;
+        this.is_local = is_local;
+    }
+}
+
+// Class handling the asynchronous network operations
+class AsyncNSLookup {
+    host: string;
+    port: number;
+    channel: Buffer;
+    socket: net.Socket | null;
+    reader: net.Socket | null;
+    writer: net.Socket | null;
+
+    constructor(host: string = DEFAULT_HOST, port: number = DEFAULT_PORT) {
+        this.host = host;
+        this.port = port;
+        this.channel = Buffer.alloc(0);
+        this.socket = null;
+        this.reader = null;
+        this.writer = null;
+    }
+
+    // Static function to open a connection
+    static async open(hostPort: string = '', timeout: number = DEFAULT_TIMEOUT): Promise<AsyncNSLookup> {
+        const nslookup = new AsyncNSLookup();
+        if (hostPort) {
+            const [host, port] = hostPort.split(':');
+            nslookup.host = host;
+            nslookup.port = port ? parseInt(port) : DEFAULT_PORT;
+        }
+        await nslookup.connect();
+        return nslookup;
+    }
+
+    // Connect to the server
+    async connect(): Promise<void> {
+        this.socket = new net.Socket();
+        this.socket.setNoDelay(true)
+        this.socket.setTimeout(DEFAULT_TIMEOUT * 1000); // Set timeout
+        this.socket.connect(this.port, this.host)
+
+        this.writer = this.socket;
+        this.reader = this.socket;
+
+        this.writer.write(MULTIPLEXED_SOCKET);
+        this.writer.write(NAME_SERVICE_SUB_PORT);
+        this.writer.write(writePackedInt(CONNECTION_OPEN.length));
+        this.writer.write(CONNECTION_OPEN);
+        this.writer.write(writePackedInt(CHANNEL_OPEN.length));
+        this.writer.write(CHANNEL_OPEN);
+
+        await sleep(1000)
+
+        await this.readResponse()
+        const data = await this.readResponse();
+
+        this.channel = data.slice(8, 8 + data.length - 9);
+    }
+
+    // Perform a lookup
+    async lookup(name: string): Promise<string> {
+        const response = await this.lookupInternal(name);
+        if (response.length <= 7) {
+            return '';
+        }
+        return this.readString(response);
+    }
+
+    // Internal lookup function
+    private async lookupInternal(name: string): Promise<Buffer> {
+        const request = Buffer.concat([
+            this.channel,
+            NS_LOOKUP_REQ_ID,
+            writePackedInt(name.length),
+            Buffer.from(name),
+            REQ_END_MARKER,
+        ]);
+
+        this.writer?.write(writePackedInt(request.length));
+        this.writer?.write(request);
+        //await this.writer?.flush();
+
+        await sleep(1000)
+
+        const response = await this.readResponse();
+        console.log("### DEBUG DATA -> " + response.toString())
+        return response.slice(this.channel.length + 1);
+    }
+
+    // Read response from the socket
+    private async readResponse(): Promise<Buffer> {
+        const length = await this.readPackedInt();
+        const data: any = this.reader?.read(length)
+        //const data = await promisify(this.reader?.read.bind(this.reader))(length);
+        return data!;
+    }
+
+    // Read packed integer
+    private async readPackedInt(): Promise<number> {
+        let bits = 6;
+
+        let byte = await this.reader?.read(1);
+        let byteValue = byte ? byte[0] : 0;
+        let negative: boolean = (byteValue & 0x40) !== 0
+        let result = (byteValue & 0x3F)
+
+        while ((byteValue & 0x80) !== 0) {
+            byte = await this.reader?.read(1);
+            byteValue = byte ? byte[0] : 0;
+            result |= (byteValue & 0x7F) << bits;
+            bits += 7;
+        }
+
+        if (negative) {
+            result = ~result;
+        }
+        return result;
+    }
+
+    private readPackedIntFromString(bytes: Buffer): [number, number] {
+        let bits = 6;
+
+        let byteValue = bytes ? bytes[6] : 0;
+        let negative: boolean = (byteValue & 0x40) !== 0
+        let result = (byteValue & 0x3F)
+
+        while ((byteValue & 0x80) !== 0) {
+            byteValue = bytes ? bytes[7] : 0;
+            result |= (byteValue & 0x7F) << bits;
+            bits += 7;
+        }
+
+        if (negative) {
+            result = ~result;
+        }
+        return [result, bits];
+    }
+
+    // Read a string from the response
+    private readString(data: Buffer): string {
+        console.log("### DEBUG DATA -> " + data)
+        let [len, pos] = this.readPackedIntFromString(data)
+        return data.slice(7 + (pos / 7), 7 + (pos / 7) + len).toString();
+    }
+
+    // Close the socket connection
+    async close(): Promise<void> {
+        if (this.writer) {
+            this.writer.end();
+        }
+    }
+
+    // Discover cluster information
+    async discoverClusterInfo(): Promise<DiscoveredCluster> {
+        const cluster = new DiscoveredCluster();
+        cluster.ns_port = this.port;
+        cluster.host = this.host;
+
+        //cluster.cluster_name = await this.lookup(CLUSTER_NAME_LOOKUP);
+        //cluster.management_urls = parseResults(await this.lookup(NS_PREFIX + MANAGEMENT_LOOKUP));
+        //cluster.jmx_urls = parseResults(await this.lookup(NS_PREFIX + JMX_LOOKUP));
+        //cluster.metrics_urls = parseResults(await this.lookup(NS_PREFIX + METRICS_LOOKUP));
+        cluster.grpc_proxy_endpoints = parseResults(await this.lookup(NS_PREFIX + GRPC_PROXY_LOOKUP));
+
+        return cluster;
+    }
+
+    // Discover name service ports
+    async discoverNameServicePorts(): Promise<ClusterNSPort[]> {
+        const localCluster = await this.lookup(CLUSTER_NAME_LOOKUP);
+        const otherClusters = await this.lookup(NS_PREFIX + CLUSTER_FOREIGN_LOOKUP);
+        const otherClustersList = parseResults(otherClusters);
+
+        const clusterNames = [localCluster, ...otherClustersList];
+        const listClusters = clusterNames.map((name) => {
+            return new ClusterNSPort(name, name === localCluster ? this.port : 0, this.host, name === localCluster);
+        });
+
+        // Determine port for other clusters
+        for (const clusterNSPort of listClusters.slice(1)) {
+            clusterNSPort.port = parseInt(await this.lookup(`${NS_PREFIX}${CLUSTER_FOREIGN_LOOKUP}/${clusterNSPort.cluster_name}${NS_LOCAL_PORT}`));
+        }
+
+        return listClusters;
+    }
+
+    // Resolve NS lookup address
+    static async resolveNSLookupAddress(name: string): Promise<string> {
+        let nslookup = null;
+        let clusterInfo = null;
+
+        try {
+            nslookup = await AsyncNSLookup.open(name);
+            clusterInfo = await nslookup.discoverClusterInfo();
+        } catch (error) {
+            console.error('Error:', error);
+        } finally {
+            if (nslookup !== null) {
+                await nslookup.close();
+            }
+        }
+
+        if (clusterInfo) {
+            const pairs = clusterInfo.grpc_proxy_endpoints
+                .map((e, i, arr) => (i % 2 === 0 ? `${e}:${arr[i + 1]}` : null))
+                .filter(Boolean);
+
+            return randomBytes(pairs.length).toString();
+        } else {
+            return 'localhost:1408';
+        }
+    }
+}
+
+// Main function to test the connection
+async function main() {
+    try {
+        const nslookup = await AsyncNSLookup.open('localhost:7574');
+        console.log(`Connected to ${nslookup.host}:${nslookup.port}`);
+
+        // Example: Perform a lookup
+        const clusterInfo = await nslookup.discoverClusterInfo();
+        console.log(`Cluster Name: ${clusterInfo.cluster_name}`);
+        console.log(`Management URLs: ${clusterInfo.management_urls}`);
+        console.log(`JMX URLs: ${clusterInfo.jmx_urls}`);
+        console.log(`Metrics URLs: ${clusterInfo.metrics_urls}`);
+        console.log(`GRPC Endpoints: ${clusterInfo.grpc_proxy_endpoints}`);
+
+        const nsPorts = await nslookup.discoverNameServicePorts();
+        nsPorts.forEach((portInfo) => {
+            console.log(`Cluster Name: ${portInfo.cluster_name}, Port: ${portInfo.port}, Host: ${portInfo.host_name}, Is Local: ${portInfo.is_local}`);
+        });
+
+    } catch (error) {
+        console.error('Error:', error);
+    }
+    process.exit(0)
+}
+
+// Run the main function
+main();

src/session.ts

@@ -13,6 +13,8 @@ import {event} from './events'
 import {NamedCache, NamedCacheClient, NamedMap} from './named-cache-client'
 import {util} from './util'
 import {ConnectivityState} from "@grpc/grpc-js/build/src/connectivity-state";
+import {registerResolver} from "@grpc/grpc-js/build/src/resolver";
+import CoherenceResolver = util.CoherenceResolver;
 
 /**
  * Supported {@link Session} options.
@@ -100,9 +102,9 @@ export class Options {
       return;
     }
     // ensure address is sane
-    if (!Options.ADDRESS_REGEXP.test(address)) {
-      throw new Error('Expected address format is \'<hostname>:<port>\'.  Configured: ' + address)
-    }
+    // if (!Options.ADDRESS_REGEXP.test(address)) {
+    //   throw new Error('Expected address format is \'<hostname>:<port>\'.  Configured: ' + address)
+    // }
 
     this._address = address
   }
@@ -507,6 +509,9 @@ export class Session
    */
   constructor (sessionOptions?: Options | object) {
     super()
+
+    registerResolver("coherence", CoherenceResolver)
+
     if (sessionOptions) {
       this._sessionOptions = Object.assign(new Options(), sessionOptions)
       // @ts-ignore  -- added for 'tls' index access