Line data    Source code

       1              : import type * as arrow from 'apache-arrow';
       2              : import * as buf from '@bufbuild/protobuf';
       3              : import * as pb from '../../proto.js';
       4              : 
       5              : import type { EmbeddedClientContext } from '../../platform/database/embedded_database.js';
       6              : import type { HyperDBSourceConnection, HyperDBSourceDatabase } from '../../platform/hyperdb/hyperdb_incremental_extract.js';
       7              : import { consumeQueryResponseStream } from '../query_execution.js';
       8              : import type { HyperDatabaseChannel } from './hyperdb_grpc_client.js';
       9              : 
      10              : /// A query on a channel to a remote Hyper, the channel keeps no session across its queries
      11            1 : class HyperChannelConnection implements HyperDBSourceConnection {
      12            1 :     constructor(
      13              :         /// The channel
      14            1 :         private readonly channel: HyperDatabaseChannel,
      15            1 :     ) {}
      16              : 
      17            0 :     async query(query: string, abort?: AbortSignal): Promise<arrow.Table> {
      18            0 :         const param = buf.create(pb.salesforce_hyperdb_grpc_v1.pb.QueryParamSchema, { query });
      19            0 :         const stream = await this.channel.executeQuery(param, abort);
      20            0 :         return (await consumeQueryResponseStream({ stream, abort, publishResults: true, onProgress: () => {} }))!;
      21            0 :     }
      22              : 
      23            0 :     async execute(statement: string, abort?: AbortSignal): Promise<bigint | null> {
      24            0 :         await this.query(statement, abort);
      25            0 :         return null;
      26            0 :     }
      27              : 
      28            0 :     async close(): Promise<void> {
      29            0 :         await this.channel.close();
      30            0 :     }
      31            1 : }
      32              : 
      33              : /// A remote Hyper as the source of an extract, e.g. of the Hyper log. Every connection is a channel of its own.
      34            1 : export class HyperChannelDatabase implements HyperDBSourceDatabase {
      35            1 :     constructor(
      36              :         /// Open a channel to the Hyper whose queries have a client context
      37            1 :         private readonly openChannel: (clientContext: EmbeddedClientContext | null) => Promise<HyperDatabaseChannel>,
      38            1 :     ) {}
      39              : 
      40            1 :     async connect(options?: { clientContext?: EmbeddedClientContext }): Promise<HyperDBSourceConnection> {
      41            1 :         return new HyperChannelConnection(await this.openChannel(options?.clientContext ?? null));
      42            1 :     }
      43            1 : }