This is an automated email from the ASF dual-hosted git repository. colegreer pushed a commit to branch js-http in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit ad8eaada017f91041013be15ace846ecf735538c Author: Cole Greer <[email protected]> AuthorDate: Thu Feb 5 10:06:20 2026 -0800 Replace JS Websockets with HTTP Initial phase of WS to HTTP replacement. Rips out WS handlers in connection.ts, replaces it with a simple HTTP transport built on top of the fetch() API. Updates to new TP4 RequestMessage format. It is able to roundtrip graphson messages to the TP4 server in this state, and pass basic client-test cases. GraphBinary4 remains unimplemented, and more adjustments are still to come regarding connection and request options, and response handling. Handling of streamed responses is yet to come. --- .../gremlin-javascript/lib/driver/client.ts | 78 +--- .../gremlin-javascript/lib/driver/connection.ts | 458 +++++---------------- .../lib/driver/request-message.ts | 189 +++++++++ .../lib/structure/io/graph-serializer.ts | 51 ++- .../javascript/gremlin-javascript/test/helper.js | 21 +- .../test/integration/client-tests.js | 78 +--- .../test/integration/session-client-tests.js | 76 ---- 7 files changed, 355 insertions(+), 596 deletions(-) diff --git a/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/driver/client.ts b/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/driver/client.ts index 59adff2102..f9b39f5ef4 100644 --- a/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/driver/client.ts +++ b/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/driver/client.ts @@ -17,13 +17,12 @@ * under the License. */ -import * as utils from '../utils.js'; import Connection, { ConnectionOptions } from './connection.js'; import { Readable } from 'stream'; +import {RequestMessage} from "./request-message.js"; export type RequestOptions = { requestId?: string; - session?: string; bindings?: any; language?: string; accept?: string; @@ -67,14 +66,6 @@ export default class Client { url: string, private readonly options: ClientOptions = {}, ) { - if (this.options.processor === 'session') { - // compatibility with old 'session' processor setting - this.options.session = options.session || utils.getUuid(); - } - if (this.options.session) { - // re-assign processor to 'session' when in session mode - this.options.processor = options.processor || 'session'; - } this._connection = new Connection(url, options); } @@ -110,31 +101,21 @@ export default class Client { * @param {Object|null} [bindings] The script bindings, if any. * @param {RequestOptions} [requestOptions] Configuration specific to the current request. * @returns {Promise} - */ + */ //TODO:: tighten return type to Promise<ResultSet> + //TODO:: Remove bytecode as allowable message type submit(message: string, bindings: any | null, requestOptions?: RequestOptions): Promise<any> { - const requestIdOverride = requestOptions && requestOptions.requestId; - if (requestIdOverride) { - delete requestOptions['requestId']; - } - - const args = Object.assign( - { - gremlin: message, - aliases: { g: this.options.traversalSource || 'g' }, - }, - requestOptions, - ); - - args['language'] = 'gremlin-lang'; - args['accept'] = this._connection.mimeType; - - if (bindings) args['bindings'] = bindings; - - if (this.options.session && this.options.processor === 'session') { - args['session'] = this.options.session; - return this._connection.submit('session', 'eval', args, requestIdOverride); - } - return this._connection.submit(this.options.processor || '', 'eval', args, requestIdOverride); + const requestBuilder = RequestMessage.build(message) + .addG(this.options.traversalSource || 'g') + .addLanguage('gremlin-lang'); + + if (requestOptions?.bindings) { + requestBuilder.addBindings(requestOptions.bindings); + } + if (bindings) { + requestBuilder.addBindings(bindings); + } + + return this._connection.submit(requestBuilder.create()); } /** @@ -144,30 +125,9 @@ export default class Client { * @param {RequestOptions} [requestOptions] Configuration specific to the current request. * @returns {ReadableStream} */ + //TODO:: Update stream() to mirror submit() stream(message: string, bindings: any, requestOptions?: RequestOptions): Readable { - const requestIdOverride = requestOptions && requestOptions.requestId; - if (requestIdOverride) { - delete requestOptions['requestId']; - } - - const args = Object.assign( - { - gremlin: message, - aliases: { g: this.options.traversalSource || 'g' }, - }, - requestOptions, - ); - - args['language'] = 'gremlin-lang'; - args['accept'] = this._connection.mimeType; - - if (bindings) args['bindings'] = bindings; - - if (this.options.session && this.options.processor === 'session') { - args['session'] = this.options.session; - return this._connection.stream('session', 'eval', args, requestIdOverride); - } - return this._connection.stream(this.options.processor || '', 'eval', args, requestIdOverride); + throw new Error("Stream not yet implemented"); } /** @@ -176,10 +136,6 @@ export default class Client { * @returns {Promise} */ close(): Promise<void> { - if (this.options.session && this.options.processor === 'session') { - const args = { session: this.options.session }; - return this._connection.submit(this.options.processor, 'close', args, null).then(() => this._connection.close()); - } return this._connection.close(); } diff --git a/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/driver/connection.ts b/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/driver/connection.ts index dc4c9804b9..90fb942b82 100644 --- a/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/driver/connection.ts +++ b/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/driver/connection.ts @@ -24,24 +24,13 @@ import { Buffer } from 'buffer'; import { EventEmitter } from 'eventemitter3'; import type { Agent } from 'node:http'; -import Stream from 'readable-stream'; -import type { - CloseEvent as NodeWebSocketCloseEvent, - ErrorEvent as NodeWebSocketErrorEvent, - MessageEvent as NodeWebSocketMessageEvent, - WebSocket as NodeWebSocket, - Event as NodeWebSocketEvent, -} from 'ws'; -import { - ClientRequest, - IncomingMessage, -} from "http"; import ioc from '../structure/io/binary/GraphBinary.js'; import * as serializer from '../structure/io/graph-serializer.js'; import * as utils from '../utils.js'; -import Authenticator from './auth/authenticator.js'; -import ResponseError from './response-error.js'; import ResultSet from './result-set.js'; +import {RequestMessage} from "./request-message.js"; +import {Readable} from "stream"; +import ResponseError from './response-error.js'; const { DeferredPromise } = utils; const { graphBinaryReader, graphBinaryWriter } = ioc; @@ -53,14 +42,12 @@ const responseStatusCode = { authenticationChallenge: 407, }; -const defaultMimeType = 'application/vnd.graphbinary-v1.0'; +const defaultMimeType = 'application/vnd.gremlin-v4.0+json'; const graphSON2MimeType = 'application/vnd.gremlin-v2.0+json'; const graphBinaryMimeType = 'application/vnd.graphbinary-v1.0'; type MimeType = typeof defaultMimeType | typeof graphSON2MimeType | typeof graphBinaryMimeType; -const uuidPattern = '[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}'; - export type ConnectionOptions = { ca?: string[]; cert?: string | string[] | Buffer; @@ -70,7 +57,6 @@ export type ConnectionOptions = { rejectUnauthorized?: boolean; traversalSource?: string; writer?: any; - authenticator?: Authenticator; headers?: Record<string, string | string[]>; enableUserAgentOnConnect?: boolean; agent?: Agent; @@ -80,25 +66,14 @@ export type ConnectionOptions = { * Represents a single connection to a Gremlin Server. */ export default class Connection extends EventEmitter { - private _ws: WebSocket | NodeWebSocket | undefined; - readonly mimeType: MimeType; - private readonly _responseHandlers: Record<string, { callback: (...args: any[]) => unknown; result: any }> = {}; private readonly _reader: any; private readonly _writer: any; - private _openPromise: ReturnType<typeof DeferredPromise<void>> | null; - private _openCallback: (() => unknown) | null; - private _closePromise: Promise<void> | null; - private _closeCallback: (() => unknown) | null; - - private readonly _header: string; - private readonly _header_buf: Buffer; - isOpen = false; + isOpen = true; traversalSource: string; - private readonly _authenticator: any; private readonly _enableUserAgentOnConnect: boolean; /** @@ -113,7 +88,6 @@ export default class Connection extends EventEmitter { * @param {Boolean} [options.rejectUnauthorized] Determines whether to verify or not the server certificate. * @param {String} [options.traversalSource] The traversal source. Defaults to: 'g'. * @param {GraphSONWriter} [options.writer] The writer to use. - * @param {Authenticator} [options.authenticator] The authentication handler to use. * @param {Object} [options.headers] An associative array containing the additional header key/values for the initial request. * @param {Boolean} [options.enableUserAgentOnConnect] Determines if a user agent will be sent during connection handshake. Defaults to: true * @param {http.Agent} [options.agent] The http.Agent implementation to use. @@ -130,20 +104,9 @@ export default class Connection extends EventEmitter { * @type {String} */ this.mimeType = options.mimeType || defaultMimeType; - - // A map containing the request id and the handler. The id should be in lower case to prevent string comparison issues. - this._responseHandlers = {}; this._reader = options.reader || this.#getDefaultReader(this.mimeType); this._writer = options.writer || this.#getDefaultWriter(this.mimeType); - this._openPromise = null; - this._openCallback = null; - this._closePromise = null; - this._closeCallback = null; - - this._header = String.fromCharCode(this.mimeType.length) + this.mimeType; // TODO: what if mimeType.length > 255 - this._header_buf = Buffer.from(this._header); this.traversalSource = options.traversalSource || 'g'; - this._authenticator = options.authenticator; this._enableUserAgentOnConnect = options.enableUserAgentOnConnect !== false; } @@ -152,155 +115,23 @@ export default class Connection extends EventEmitter { * @returns {Promise} */ async open() { - if (this.isOpen) { - return; - } - if (this._openPromise) { - return this._openPromise; - } - - this._openPromise = DeferredPromise(); - - this.emit('log', 'ws open'); - let headers = this.options.headers; - if (this._enableUserAgentOnConnect) { - if (!headers) { - headers = {}; - } - - const userAgent = await utils.getUserAgent(); - if (userAgent !== undefined) { - headers[utils.getUserAgentHeader()] = userAgent; - } - } - // All these options are available to the `ws` package's constructor, but not the global WebSocket class - const wsSpecificOptions: Set<string> = new Set([ - 'headers', - 'ca', - 'cert', - 'pfx', - 'rejectUnauthorized', - 'agent', - 'perMessageDeflate', - ]); - // Check if any `ws` specific options are provided and are non-null / non-undefined - const hasWsSpecificOptions: boolean = - Object.entries(this.options).some( - ([key, value]) => wsSpecificOptions.has(key) && ![null, undefined].includes(value), - ) || - this._enableUserAgentOnConnect; // global websocket will send "node" as user agent by default which doesn't comply with Gremlin - // Only use the global websocket if we don't have any unsupported options - const useGlobalWebSocket = !hasWsSpecificOptions && globalThis.WebSocket; - const WebSocket = useGlobalWebSocket || (await import('ws')).default; - - this._ws = new WebSocket( - this.url, - !useGlobalWebSocket - ? { - // @ts-expect-error - headers: headers, - ca: this.options.ca, - cert: this.options.cert, - pfx: this.options.pfx, - rejectUnauthorized: this.options.rejectUnauthorized, - agent: this.options.agent, - } - : undefined, - ); - - if ('binaryType' in this._ws!) { - this._ws.binaryType = 'arraybuffer'; - } - - // @ts-expect-error - this._ws!.addEventListener('open', this.#handleOpen); - // @ts-expect-error - this._ws!.addEventListener('error', this.#handleError); - // Only attach unexpected-response listener for Node.js WebSocket (ws package) - // Browser WebSocket does not have this event and .on() method - if (!useGlobalWebSocket) { - // The following listener needs to use `.on` and `.off` because the `ws` package's addEventListener only accepts certain event types - // Ref: https://github.com/websockets/ws/blob/8.16.0/lib/event-target.js#L202-L241 - // @ts-expect-error - this._ws!.on('unexpected-response', this.#handleUnexpectedResponse); - } - - // @ts-expect-error - this._ws!.addEventListener('message', this.#handleMessage); - // @ts-expect-error - this._ws!.addEventListener('close', this.#handleClose); - - return await this._openPromise; + // No-op for HTTP connections + return Promise.resolve(); } /** @override */ - submit(processor: string | undefined, op: string, args: any, requestId?: string | null) { - // TINKERPOP-2847: Use lower case to prevent string comparison issues. - const rid = (requestId || utils.getUuid()).toLowerCase(); - if (!rid.match(uuidPattern)) { - throw new Error('Provided requestId "' + rid + '" is not a valid UUID.'); - } - - return this.open().then( - () => - new Promise((resolve, reject) => { - if (op !== 'authentication') { - this._responseHandlers[rid] = { - callback: (err: Error, result: any) => (err ? reject(err) : resolve(result)), - result: null, - }; - } - - const request = { - requestId: rid, - op: op || 'eval', - // if using op eval need to ensure processor stays unset if caller didn't set it. - processor: !processor && op !== 'eval' ? 'traversal' : processor, - args: args || {}, - }; - - const request_buf = this._writer.writeRequest(request); - const message = utils.toArrayBuffer(Buffer.concat([this._header_buf, request_buf])); - this._ws!.send(message); - }), - ); + submit(request: RequestMessage) { + const request_buf = this._writer.writeRequest(request); + + return this.#makeHttpRequest(request_buf) + .then((response) => { + return this.#handleResponse(response); + }); } /** @override */ - stream(processor: string, op: string, args: any, requestId?: string) { - // TINKERPOP-2847: Use lower case to prevent string comparison issues. - const rid = (requestId || utils.getUuid()).toLowerCase(); - if (!rid.match(uuidPattern)) { - throw new Error('Provided requestId "' + rid + '" is not a valid UUID.'); - } - - const readableStream = new Stream.Readable({ - objectMode: true, - read() {}, - }); - - this._responseHandlers[rid] = { - callback: (err: Error) => (err ? readableStream.destroy(err) : readableStream.push(null)), - result: readableStream, - }; - - this.open() - .then(() => { - const request = { - requestId: rid, - op: op || 'eval', - // if using op eval need to ensure processor stays unset if caller didn't set it. - processor: !processor && op !== 'eval' ? 'traversal' : processor, - args: args || {}, - }; - - const request_buf = this._writer.writeRequest(request); - const message = utils.toArrayBuffer(Buffer.concat([this._header_buf, request_buf])); - this._ws!.send(message); - }) - .catch((err) => readableStream.destroy(err)); - - return readableStream; + stream(request: RequestMessage): Readable { + throw new Error('stream() is not yet implemented'); } #getDefaultReader(mimeType: MimeType) { @@ -319,183 +150,107 @@ export default class Connection extends EventEmitter { return mimeType === graphSON2MimeType ? new serializer.GraphSON2Writer() : new serializer.GraphSONWriter(); } - #handleOpen = (_: Event | NodeWebSocketEvent) => { - this._openPromise?.resolve(); - this.isOpen = true; - }; - - #handleError = (event: Event | NodeWebSocketErrorEvent) => { - const error = 'error' in event ? event.error : event; - this._openPromise?.reject(error); - this.emit('log', `ws error ${error}`); - this.#cleanupWebsocket(error); - this.emit('socketError', error); - }; - - #handleUnexpectedResponse = async (_: ClientRequest, res: IncomingMessage) => { - const body = await new Promise((resolve, reject) => { - const chunks: any[] = []; - res.on('data', data => { - chunks.push(data instanceof Buffer ? data : Buffer.from(data)); - }); - res.on('end', () => { - resolve(chunks.length ? Buffer.concat(chunks) : null); - }); - res.on('error', reject); - }); - // @ts-ignore - const statusCodeErrorMessage = `Unexpected server response code ${res.statusCode}`; - // @ts-ignore - const errorMessage = body ? `${statusCodeErrorMessage} with body:\n${body.toString()}` : statusCodeErrorMessage; - const error = new Error(errorMessage); - this.#handleError({ - error, - message: errorMessage, - type: 'unexpected-response', - target: this._ws - } as NodeWebSocketErrorEvent); - }; - - - - #handleClose = ({ code, reason }: CloseEvent | NodeWebSocketCloseEvent) => { - this.emit('log', `ws close code=${code} message=${reason}`); - this.#cleanupWebsocket(); - if (this._closeCallback) { - this._closeCallback(); + #getReaderForContentType(contentType: string | null) { + if (!contentType) { + return this._reader; } - this.emit('close', code, reason); - }; - - #handleMessage = ({ data: _data }: MessageEvent | NodeWebSocketMessageEvent) => { - const data = _data instanceof ArrayBuffer ? Buffer.from(_data) : _data; - - const response = this._reader.readResponse(data); - if (response.requestId === null || response.requestId === undefined) { - // There was a serialization issue on the server that prevented the parsing of the request id - // We invoke any of the pending handlers with an error - Object.keys(this._responseHandlers).forEach((requestId) => { - const handler = this._responseHandlers[requestId]; - this.#clearHandler(requestId); - if (response.status !== undefined && response.status.message) { - return handler.callback( - // TINKERPOP-2285: keep the old server error message in case folks are parsing that - fix in a future breaking version - new ResponseError( - `Server error (no request information): ${response.status.message} (${response.status.code})`, - response.status, - ), - ); - } - // TINKERPOP-2285: keep the old server error message in case folks are parsing that - fix in a future breaking version - return handler.callback( - new ResponseError(`Server error (no request information): ${JSON.stringify(response)}`, response.status), - ); - }); - return; + + if (contentType === graphBinaryMimeType) { + return graphBinaryReader; } - // TINKERPOP-2847: Use lower case to prevent string comparison issues. - response.requestId = response.requestId.toLowerCase(); - const handler = this._responseHandlers[response.requestId]; + if (contentType === graphSON2MimeType) { + return new serializer.GraphSON2Reader(); + } - if (!handler) { - // The handler for a given request id was not found - // It was probably invoked earlier due to a serialization issue. - return; + if (contentType === defaultMimeType) { + return new serializer.GraphSONReader(); } - if (response.status.code === responseStatusCode.authenticationChallenge && this._authenticator) { - this._authenticator - .evaluateChallenge(response.result.data) - .then((res: any) => this.submit(undefined, 'authentication', res, response.requestId)) - .catch(handler.callback); - - return; - } else if (response.status.code >= 400) { - this.#clearHandler(response.requestId); - // callback in error - return handler.callback( - // TINKERPOP-2285: keep the old server error message in case folks are parsing that - fix in a future breaking version - new ResponseError(`Server error: ${response.status.message} (${response.status.code})`, response.status), - ); + return null; + } + + async #makeHttpRequest(request_buf: Buffer): Promise<Response> { + const headers: Record<string, string> = { + 'Content-Type': this.mimeType, + 'Accept': this.mimeType + }; + + if (this._enableUserAgentOnConnect) { + const userAgent = await utils.getUserAgent(); + if (userAgent !== undefined) { + headers[utils.getUserAgentHeader()] = userAgent; + } } - const isStreamingResponse = handler.result instanceof Stream.Readable; + if (this.options.headers) { + Object.entries(this.options.headers).forEach(([key, value]) => { + headers[key] = Array.isArray(value) ? value.join(', ') : value; + }); + } - switch (response.status.code) { - case responseStatusCode.noContent: - this.#clearHandler(response.requestId); - if (isStreamingResponse) { - handler.result.push(new ResultSet(utils.emptyArray, response.status.attributes)); - return handler.callback(null); - } - return handler.callback(null, new ResultSet(utils.emptyArray, response.status.attributes)); - case responseStatusCode.partialContent: - if (isStreamingResponse) { - handler.result.push(new ResultSet(response.result.data, response.status.attributes)); - break; - } - handler.result = handler.result || []; - handler.result.push.apply(handler.result, response.result.data); - break; - default: - if (isStreamingResponse) { - handler.result.push(new ResultSet(response.result.data, response.status.attributes)); - return handler.callback(null); + return fetch(this.url, { + method: 'POST', + headers, + body: request_buf, + }); + } + + async #handleResponse(response: Response) { + const contentType = response.headers.get("Content-Type"); + const buffer = Buffer.from(await response.arrayBuffer()); + const reader = this.#getReaderForContentType(contentType); + + if (!response.ok) { + const errorMessage = `Server returned HTTP ${response.status}: ${response.statusText}`; + + try { + if (reader) { + const deserialized = reader.readResponse(buffer); + throw new ResponseError(errorMessage, { + code: deserialized.status.code, + message: deserialized.status.message || response.statusText, + attributes: deserialized.status.attributes || new Map(), + }); + } else if (contentType === 'application/json') { + const errorBody = JSON.parse(buffer.toString()); + throw new ResponseError(errorMessage, { + code: response.status, + message: errorBody.message || errorBody.error || response.statusText, + attributes: new Map(), + }); } - if (handler.result) { - handler.result.push.apply(handler.result, response.result.data); - } else { - handler.result = response.result.data; + } catch (err) { + if (err instanceof ResponseError) { + throw err; } - this.#clearHandler(response.requestId); - return handler.callback(null, new ResultSet(handler.result, response.status.attributes)); + } + + throw new ResponseError(errorMessage, { + code: response.status, + message: response.statusText, + attributes: new Map(), + }); } - }; - /** - * clean websocket context - */ - #cleanupWebsocket(err?: Error) { - // Invoke waiting callbacks to complete Promises when closing the websocket - Object.keys(this._responseHandlers).forEach((requestId) => { - const handler = this._responseHandlers[requestId]; - const isStreamingResponse = handler.result instanceof Stream.Readable; - if (isStreamingResponse) { - handler.callback(null); - } else { - const cause = err ? err : new Error('Connection has been closed.'); - handler.callback(cause); - } - }); - // @ts-expect-error - this._ws?.removeEventListener('open', this.#handleOpen); - // @ts-expect-error - this._ws?.removeEventListener('error', this.#handleError); - // Only remove unexpected-response listener for Node.js WebSocket (ws package) - // Browser WebSocket does not have this event and .off() method - if (this._ws && 'off' in this._ws) { - // The following listener needs to use `.on` and `.off` because the `ws` package's addEventListener only accepts certain event types - // Ref: https://github.com/websockets/ws/blob/8.16.0/lib/event-target.js#L202-L241 - this._ws.off('unexpected-response', this.#handleUnexpectedResponse); + if (!reader) { + throw new Error(`Unsupported Content-Type: ${contentType}`); } - // @ts-expect-error - this._ws?.removeEventListener('message', this.#handleMessage); - // @ts-expect-error - this._ws?.removeEventListener('close', this.#handleClose); - this._openPromise = null; - this._closePromise = null; - this.isOpen = false; - } + const deserialized = reader.readResponse(buffer); + + if (deserialized.status.code && deserialized.status.code !== 200 && deserialized.status.code !== 204 && deserialized.status.code !== 206) { + throw new ResponseError( + `Server returned status ${deserialized.status.code}: ${deserialized.status.message || 'Unknown error'}`, + { + code: deserialized.status.code, + message: deserialized.status.message || '', + attributes: deserialized.status.attributes || new Map(), + } + ); + } - /** - * Clears the internal state containing the callback and result buffer of a given request. - * @param requestId - * @private - */ - #clearHandler(requestId: string) { - delete this._responseHandlers[requestId]; + return new ResultSet(deserialized.result.data, deserialized.result.meta || new Map()); } /** @@ -503,15 +258,6 @@ export default class Connection extends EventEmitter { * @return {Promise} */ close() { - if (this.isOpen === false) { - return Promise.resolve(); - } - if (!this._closePromise) { - this._closePromise = new Promise((resolve) => { - this._closeCallback = resolve; - this._ws?.close(); - }); - } - return this._closePromise; + return Promise.resolve(); } } diff --git a/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/driver/request-message.ts b/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/driver/request-message.ts new file mode 100644 index 0000000000..84a85e45c7 --- /dev/null +++ b/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/driver/request-message.ts @@ -0,0 +1,189 @@ +/* + * 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. + */ + +// Token constants +const Tokens = { + ARGS_LANGUAGE: 'language', + ARGS_BINDINGS: 'bindings', + ARGS_G: 'g', + ARGS_MATERIALIZE_PROPERTIES: 'materializeProperties', + TIMEOUT_MS: 'timeoutMs', + BULK_RESULTS: 'bulkResults', + MATERIALIZE_PROPERTIES_TOKENS: 'tokens', + MATERIALIZE_PROPERTIES_ALL: 'all' +}; + +/** + * The model for a request message in the HTTP body that is sent to the server beginning in 4.0.0. + */ +export class RequestMessage { + private gremlin: string; + private language: string; + private timeoutMs?: number; + private bindings?: object; + private g?: string; + private materializeProperties?: string; + private bulkResults?: boolean; + private customFields: Map<string, any>; + + private constructor( + gremlin: string, + language: string, + timeoutMs: number | undefined, + bindings: object | undefined, + g: string | undefined, + materializeProperties: string | undefined, + bulkResults: boolean | undefined, + customFields: Map<string, any> + ) { + if (!gremlin) { + throw new Error('RequestMessage requires gremlin argument'); + } + + this.gremlin = gremlin; + this.language = language; + this.timeoutMs = timeoutMs; + this.bindings = bindings; + this.g = g; + this.materializeProperties = materializeProperties; + this.bulkResults = bulkResults; + this.customFields = customFields; + } + + getGremlin(): string { + return this.gremlin; + } + + getLanguage(): string { + return this.language; + } + + getTimeoutMs(): number | undefined { + return this.timeoutMs; + } + + getBindings(): object | undefined { + return this.bindings; + } + + getG(): string | undefined { + return this.g; + } + + getMaterializeProperties(): string | undefined { + return this.materializeProperties; + } + + getBulkResults(): boolean | undefined { + return this.bulkResults; + } + + getFields(): ReadonlyMap<string, any> { + return this.customFields; + } + + static build(gremlin: string): Builder { + return new Builder(gremlin); + } +} + +/** + * Builder class for RequestMessage. + */ +export class Builder { + private readonly gremlin: string; + private readonly bindings = {}; + public language?: string; + public timeoutMs?: number; + public g?: string; + public materializeProperties?: string; + public bulkResults?: boolean; + public additionalFields = new Map<string, any>(); + + constructor(gremlin: string) { + this.gremlin = gremlin; + } + + addLanguage(language: string): Builder { + if (!language) throw new Error('language argument cannot be null.'); + this.language = language; + return this; + } + + addBinding(key: string, value: any): Builder { + Object.assign(this.bindings, {key: value}) + return this; + } + + addBindings(otherBindings: object): Builder { + if (!otherBindings) throw new Error('bindings argument cannot be null.'); + Object.assign(this.bindings, otherBindings) + return this; + } + + addG(g: string): Builder { + if (!g) throw new Error('g argument cannot be null.'); + this.g = g; + return this; + } + + addMaterializeProperties(materializeProps: string): Builder { + if (!materializeProps) throw new Error('materializeProps argument cannot be null.'); + if (materializeProps !== Tokens.MATERIALIZE_PROPERTIES_TOKENS && + materializeProps !== Tokens.MATERIALIZE_PROPERTIES_ALL) { + throw new Error(`materializeProperties argument must be either "${Tokens.MATERIALIZE_PROPERTIES_TOKENS}" or "${Tokens.MATERIALIZE_PROPERTIES_ALL}".`); + } + + this.materializeProperties = materializeProps; + return this; + } + + addTimeoutMillis(timeout: number): Builder { + if (timeout < 0) throw new Error('timeout argument cannot be negative.'); + this.timeoutMs = timeout; + return this; + } + + addBulkResults(bulking: boolean): Builder { + this.bulkResults = bulking; + return this; + } + + addField(key: string, value: any): Builder { + this.additionalFields.set(key, value); + return this; + } + + /** + * Create the request message given the settings provided to the Builder. + */ + create(): RequestMessage { + // @ts-ignore - accessing private constructor from Builder + return new RequestMessage( + this.gremlin, + this.language || 'gremlin-lang', + this.timeoutMs, + Object.keys(this.bindings).length > 0 ? this.bindings : undefined, + this.g, + this.materializeProperties, + this.bulkResults, + this.additionalFields + ); + } +} diff --git a/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/structure/io/graph-serializer.ts b/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/structure/io/graph-serializer.ts index 2d7bf12a51..091ef6e35f 100644 --- a/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/structure/io/graph-serializer.ts +++ b/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/structure/io/graph-serializer.ts @@ -48,6 +48,7 @@ import { typeKey, valueKey, } from './type-serializers.js'; +import {RequestMessage} from "../../driver/request-message.js"; export type GraphWriterOptions = { serializers?: Record<string, TypeSerializer<any>>; @@ -128,25 +129,41 @@ export class GraphSON2Writer { return JSON.stringify(this.adaptObject(obj)); } - writeRequest({ - requestId, - op, - processor, - args, - }: { - processor: string | undefined; - op: string; - args: any; - requestId?: string | null; - }) { - const req = { - requestId: { '@type': 'g:UUID', '@value': requestId }, - op, - processor, - args: this._adaptArgs(args, true), + writeRequest(request: RequestMessage) { + const fields: Record<string, any> = { + language: request.getLanguage() || "gremlin-lang", + g: request.getG() || "g", + bindings: this._adaptArgs(request.getBindings() || {}, true) }; - return Buffer.from(JSON.stringify(req)); + const timeoutMs = request.getTimeoutMs(); + if (timeoutMs !== undefined) fields.timeoutMs = this.adaptObject(timeoutMs); + + const materializeProperties = request.getMaterializeProperties(); + if (materializeProperties !== undefined) fields.materializeProperties = this.adaptObject(materializeProperties); + + const bulkResults = request.getBulkResults(); + if (bulkResults !== undefined) fields.bulkResults = this.adaptObject(bulkResults); + + const allFields = request.getFields(); + for (const [key, value] of allFields) { + fields[key] = this.adaptObject(value); + } + + const req: { + gremlin: string, + timeoutMs: number | undefined, + bindings: object | undefined + } = { + gremlin: request.getGremlin(), + timeoutMs: undefined, + bindings: undefined + }; + + if (request.getBindings() !== undefined) req.bindings = request.getBindings(); + + const reqString = JSON.stringify(req) + return Buffer.from(reqString); } /** diff --git a/gremlin-javascript/src/main/javascript/gremlin-javascript/test/helper.js b/gremlin-javascript/src/main/javascript/gremlin-javascript/test/helper.js index 304fba7a8c..27f04b8369 100644 --- a/gremlin-javascript/src/main/javascript/gremlin-javascript/test/helper.js +++ b/gremlin-javascript/src/main/javascript/gremlin-javascript/test/helper.js @@ -34,14 +34,14 @@ let serverAuthUrl; let socketServerUrl; let sockerServerConfigPath; if (process.env.DOCKER_ENVIRONMENT === 'true') { - serverUrl = 'ws://gremlin-server-test-js:45940/gremlin'; - serverAuthUrl = 'wss://gremlin-server-test-js:45941/gremlin'; - socketServerUrl = 'ws://gremlin-socket-server-js:'; + serverUrl = 'http://gremlin-server-test-js:45940/gremlin'; + serverAuthUrl = 'https://gremlin-server-test-js:45941/gremlin'; + socketServerUrl = 'http://gremlin-socket-server-js:'; sockerServerConfigPath = '/js_app/gremlin-socket-server/conf/test-ws-gremlin.yaml'; } else { - serverUrl = 'ws://localhost:45940/gremlin'; - serverAuthUrl = 'wss://localhost:45941/gremlin'; - socketServerUrl = 'ws://localhost:'; + serverUrl = 'http://localhost:8182/gremlin'; + serverAuthUrl = 'https://localhost:45941/gremlin'; + socketServerUrl = 'http://localhost:'; sockerServerConfigPath = '../../../../../gremlin-tools/gremlin-socket-server/conf/test-ws-gremlin.yaml'; } @@ -72,15 +72,6 @@ export function getClient(traversalSource) { return new Client(serverUrl, { traversalSource, mimeType: process.env.CLIENT_MIMETYPE }); } -export function getSessionClient(traversalSource) { - const sessionId = utilsJs.getUuid(); - return new Client(serverUrl, { - 'traversalSource': traversalSource, - 'session': sessionId.toString(), - mimeType: process.env.CLIENT_MIMETYPE, - }); -} - function getMimeTypeFromSocketServerSettings(socketServerSettings) { let mimeType; switch(socketServerSettings.SERIALIZER) { diff --git a/gremlin-javascript/src/main/javascript/gremlin-javascript/test/integration/client-tests.js b/gremlin-javascript/src/main/javascript/gremlin-javascript/test/integration/client-tests.js index f235bd53cc..db9208770c 100644 --- a/gremlin-javascript/src/main/javascript/gremlin-javascript/test/integration/client-tests.js +++ b/gremlin-javascript/src/main/javascript/gremlin-javascript/test/integration/client-tests.js @@ -18,7 +18,6 @@ */ import assert from 'assert'; -import Bytecode from '../../lib/process/bytecode.js'; import { Vertex, Edge, VertexProperty } from '../../lib/structure/graph.js'; import { getClient } from '../helper.js'; import { cardinality } from '../../lib/process/traversal.js'; @@ -27,7 +26,7 @@ let client, clientCrew; describe('Client', function () { before(function () { - client = getClient('gmodern'); + client = getClient('g'); clientCrew = getClient('gcrew') return client.open(); }); @@ -36,14 +35,6 @@ describe('Client', function () { return client.close(); }); describe('#submit()', function () { - it('should send bytecode', function () { - return client.submit(new Bytecode().addStep('V', []).addStep('tail', [])) - .then(function (result) { - assert.ok(result); - assert.strictEqual(result.length, 1); - assert.ok(result.first().object instanceof Vertex); - }); - }); it('should send and parse a script', function () { return client.submit('g.V().tail()') .then(function (result) { @@ -53,10 +44,10 @@ describe('Client', function () { }); }); it('should send and parse a script with bindings', function () { - return client.submit('x + x', { x: 3 }) + return client.submit('g.V().has("name", x).values("age")', { x: 'marko' }) .then(function (result) { assert.ok(result); - assert.strictEqual(result.first(), 6); + assert.strictEqual(result.first(), 29); }); }); it('should send and parse a script with non-native javascript bindings', function () { @@ -68,31 +59,15 @@ describe('Client', function () { }); it('should retrieve the attributes', () => { - return client.submit(new Bytecode().addStep('V', []).addStep('tail', [])) + return client.submit("g.V().tail()") .then(rs => { assert.ok(rs.attributes instanceof Map); assert.ok(rs.attributes.get('host')); }); }); - it('should handle Vertex properties for bytecode request', function () { - return client.submit(new Bytecode().addStep('V', [1])) - .then(function (result) { - assert.ok(result); - assert.strictEqual(result.length, 1); - const vertex = result.first().object; - assert.ok(vertex instanceof Vertex); - const age = vertex.properties.find(p => p.key === 'age'); - const name = vertex.properties.find(p => p.key === 'name'); - assert.ok(age); - assert.ok(name); - assert.strictEqual(age.value, 29); - assert.strictEqual(name.value, 'marko'); - }); - }); - - it('should skip Vertex properties for bytecode request with tokens', function () { - return client.submit(new Bytecode().addStep('V', [1]), null, {'materializeProperties': 'tokens'}) + it('should skip Vertex properties for request with tokens', function () { + return client.submit("g.V(1)", null, {'materializeProperties': 'tokens'}) .then(function (result) { assert.ok(result); assert.strictEqual(result.length, 1); @@ -177,21 +152,6 @@ describe('Client', function () { await crewClient.close(); }); - it('should handle VertexProperties properties for bytecode request', async function () { - const crewClient = getClient('gcrew'); - await crewClient.open(); - - const result = await crewClient.submit(new Bytecode().addStep('V', [7])); - - assert.ok(result); - assert.strictEqual(result.length, 1); - const vertex = result.first().object; - - assertVertexProperties(vertex); - - await crewClient.close(); - }); - it('should be able to stream results from the gremlin server', (done) => { const output = []; let calls = 0; @@ -249,30 +209,6 @@ describe('Client', function () { } }); - it("should get error for malformed requestId for bytecode stream", async () => { - try { - const readable = client.stream(new Bytecode().addStep('V', []), {}, {requestId: 'malformed'}); - for await (const result of readable) { - assert.fail("malformed requestId should throw"); - } - } catch (e) { - assert.ok(e); - assert.ok(e.message); - assert.ok(e.message.includes("is not a valid UUID.")); - } - }); - - it("should get error for malformed requestId for bytecode submit", async () => { - try { - await client.submit(new Bytecode().addStep('V', []), {}, {requestId: 'malformed'}); - assert.fail("malformed requestId should throw"); - } catch (e) { - assert.ok(e); - assert.ok(e.message); - assert.ok(e.message.includes("is not a valid UUID.")); - } - }); - it("should reject pending traversal promises if connection closes", async () => { const closingClient = getClient('gmodern'); await closingClient.open(); @@ -283,7 +219,7 @@ describe('Client', function () { const pending = async function submitTraversals() { while (Date.now() < startTime + timeout) { try { - await closingClient.submit(new Bytecode().addStep('V', []).addStep('tail', [])); + await closingClient.submit("g.V().tail()"); } catch (e) { isRejected = true; return; diff --git a/gremlin-javascript/src/main/javascript/gremlin-javascript/test/integration/session-client-tests.js b/gremlin-javascript/src/main/javascript/gremlin-javascript/test/integration/session-client-tests.js deleted file mode 100644 index b2ce2f7b67..0000000000 --- a/gremlin-javascript/src/main/javascript/gremlin-javascript/test/integration/session-client-tests.js +++ /dev/null @@ -1,76 +0,0 @@ -/* - * 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. - */ - -/** - * @author : Liu Jianping - */ - -import assert from 'assert'; -import Bytecode from '../../lib/process/bytecode.js'; -import { getSessionClient } from '../helper.js'; - -let client; - -describe('Client', function () { - before(function () { - client = getSessionClient('g'); - return client.open(); - }); - after(function () { - return client.close(); - }); - - describe('#submit()', function () { - it('should send bytecode and response error', function () { - return client.submit(new Bytecode().addStep('V', []).addStep('tail', [])) - .catch(function (err) { - assert.ok(err); - assert.ok(err.message.indexOf('session') > 0); - }); - }); - - it('should use global cache in session', function () { - return client.submit("x = [0, 1, 2, 3, 4, 5]") - .then(function (result) { - assert.ok(result); - assert.strictEqual(result.length, 6); - //console.log("x : %s", JSON.stringify(result)); - }).then(function () { - client.submit("x[2] + 4") - .then(function(result) { - assert.ok(result); - assert.strictEqual(result.length, 1); - assert.strictEqual(result.first(), 6); - //console.log("x[2] + 4: %s", JSON.stringify(result)); - }); - }); - }); - - it('should use bindings and golbal cache variable in session', function () { - return client.submit('x[3] + y', { y: 3 }) - .then(function (result) { - assert.ok(result); - assert.strictEqual(result.length, 1); - assert.strictEqual(result.first(), 6); - //console.log("x[3] + y: %s", JSON.stringify(result)); - }); - }); - - }); -});
