feat: streaming csv

CONTRIBUTING.md

@@ -21,7 +21,7 @@ We try to make it as easy as possible for you but there are a few things to cons
 - Make use of the `.editorconfig`-file if provided with the repository.
 - Make commits of logical units and describe them properly, documenting anything new that you add.
 - If possible, submit tests to your patch / new feature so it can be tested easily.
-- Assure nothing is broken by running all the tests via `npm test`. Start a test-database by running `docker run -p 8086:8086 influxdb:1.8`
+- Assure nothing is broken by running all the tests via `npm test`. Also see [examples/testing.md](examples/testing.md)
 
 ## Submit Changes
 

src/index.ts

@@ -209,6 +209,13 @@ export interface IQueryOptions {
  placeholders?: Record<string, string | number>;
 }
 
+export interface IStreamOptions extends IQueryOptions {
+ /**
+ * If true, will instruct the influxdb to answer with csv data.
+ */
+ csv?: boolean;
+}
+
 export interface IParseOptions {
  /**
  * Precision at which the points are written, defaults to nanoseconds 'n'.
@@ -1525,29 +1532,32 @@ export class InfluxDB {
  *
  * @param query
  * @param callback A function receiving an optional error as the first argument,
- * and an IncomingMessage as second parameter. The IncomingMessage can be used
- * to listen for example for "data" and "end" events.
+ * and an IncomingMessage as second parameter. The IncomingMessage is a readable
+ * stream and can be used to listen for example for "data" and "end" events.
  * @param [options]
  * @example
  * influx.stream('select * from perf', (err, res) => {
  * res.on('data', (data: string): void => {
  * console.log(data)
  * })
- * })
+ * }, { csv: true })
  */
  public stream(
  query: string | string[],
  callback: (err: Error | undefined, res: http.IncomingMessage) => void,
- options: IQueryOptions = {}
+ options: IStreamOptions = {}
  ): void {
  this._pool.stream(
- this._getQueryOpts({
- db: options.database ?? this._defaultDB(),
- epoch: options.precision,
- q: query,
- rp: options.retentionPolicy,
- params: JSON.stringify(options.placeholders ?? {}),
- }),
+ {
+ ...this._getQueryOpts({
+ db: options.database ?? this._defaultDB(),
+ epoch: options.precision,
+ q: query,
+ rp: options.retentionPolicy,
+ params: JSON.stringify(options.placeholders ?? {}),
+ }),
+ csv: options.csv || false,
+ },
  callback
  );
  }

src/pool.ts

@@ -68,6 +68,13 @@ export interface IPoolRequestOptions {
  retries?: number;
 }
 
+export interface IPoolStreamOptions extends IPoolRequestOptions {
+ /**
+ * If true, will instruct the influxdb to answer with csv data.
+ */
+ csv?: boolean;
+}
+
 /**
  * An ServiceNotAvailableError is returned as an error from requests that
  * result in a > 500 error code.
@@ -358,7 +365,7 @@ export class Pool {
  * if possible. An error is returned on a non-2xx status code.
  */
  public stream(
- options: IPoolRequestOptions,
+ options: IPoolStreamOptions,
  callback: (err: Error, res: http.IncomingMessage) => void
  ): void {
  if (!this.hostIsAvailable()) {
@@ -374,11 +381,17 @@ export class Pool {
  path += "?" + querystring.stringify(options.query);
  }
 
+ const headers: Record<string, string | number> = {
+ "content-length": options.body ? Buffer.from(options.body).length : 0,
+ };
+
+ if (options.csv) {
+ headers.accept = "application/csv";
+ }
+
  const req = request(
  {
- headers: {
- "content-length": options.body ? Buffer.from(options.body).length : 0,
- },
+ headers,
  hostname: host.url.hostname,
  method: options.method,
  path,