feat: streaming raw data

src/index.ts

@@ -3,6 +3,7 @@
 
 import { RequestOptions } from "https";
 import * as url from "url";
+import * as http from "http";
 
 import * as b from "./builder";
 import * as grammar from "./grammar";
@@ -1518,6 +1519,39 @@ export class InfluxDB {
  );
  }
 
+ /**
+ * Makes a request and calls back with the IncomingMessage stream,
+ * if possible. An error is returned on a non-2xx status code.
+ *
+ * @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.
+ * @param [options]
+ * @example
+ * influx.stream('select * from perf', (err, res) => {
+ * res.on('data', (data: string): void => {
+ * console.log(data)
+ * })
+ * })
+ */
+ public stream(
+ query: string | string[],
+ callback: (err: Error | undefined, res: http.IncomingMessage) => void,
+ options: IQueryOptions = {}
+ ): void {
+ this._pool.stream(
+ this._getQueryOpts({
+ db: options.database ?? this._defaultDB(),
+ epoch: options.precision,
+ q: query,
+ rp: options.retentionPolicy,
+ params: JSON.stringify(options.placeholders ?? {}),
+ }),
+ callback
+ );
+ }
+
  /**
  * Pings all available hosts, collecting online status and version info.
  * @param timeout Given in milliseconds

test/unit/influx.test.ts

@@ -4,6 +4,7 @@ import * as sinon from "sinon";
 import { FieldType, InfluxDB, toNanoDate } from "../../src";
 import { Pool } from "../../src/pool";
 import { dbFixture } from "./helpers";
+import * as http from "http";
 
 describe("influxdb", () => {
  describe("constructor", () => {
@@ -165,6 +166,7 @@ describe("influxdb", () => {
  sinon.stub(pool, "discard");
  sinon.stub(pool, "json");
  sinon.stub(pool, "text");
+ sinon.stub(pool, "stream");
  });
 
  afterEach(() => {
@@ -1007,6 +1009,33 @@ describe("influxdb", () => {
  });
  });
 
+ describe(".stream", () => {
+ beforeEach(() => setDefaultDB("my_db"));
+
+ it("calls the pools stream method correctly", () => {
+ const callback = (_err: Error, _res: http.IncomingMessage) => {};
+
+ expectQuery(
+ "stream",
+ {
+ q: "select * from series_0",
+ epoch: undefined,
+ rp: undefined,
+ db: "my_db",
+ params: "{}",
+ },
+ "GET",
+ dbFixture("selectFromOne")
+ );
+
+ influx.stream("select * from series_0", callback);
+
+ expect((pool.stream as sinon.SinonSpy).getCall(0).args[1]).to.equal(
+ callback
+ );
+ });
+ });
+
  describe(".createRetentionPolicy", () => {
  beforeEach(() => setDefaultDB("my_db"));