resource-stream.js 3.0 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182
  1. "use strict";
  2. /*!
  3. * Copyright 2019 Google Inc. All Rights Reserved.
  4. *
  5. * Licensed under the Apache License, Version 2.0 (the "License");
  6. * you may not use this file except in compliance with the License.
  7. * You may obtain a copy of the License at
  8. *
  9. * http://www.apache.org/licenses/LICENSE-2.0
  10. *
  11. * Unless required by applicable law or agreed to in writing, software
  12. * distributed under the License is distributed on an "AS IS" BASIS,
  13. * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14. * See the License for the specific language governing permissions and
  15. * limitations under the License.
  16. */
  17. Object.defineProperty(exports, "__esModule", { value: true });
  18. exports.ResourceStream = void 0;
  19. const stream_1 = require("stream");
  20. class ResourceStream extends stream_1.Transform {
  21. constructor(args, requestFn) {
  22. const options = Object.assign({ objectMode: true }, args.streamOptions);
  23. super(options);
  24. this._ended = false;
  25. this._maxApiCalls = args.maxApiCalls === -1 ? Infinity : args.maxApiCalls;
  26. this._nextQuery = args.query;
  27. this._reading = false;
  28. this._requestFn = requestFn;
  29. this._requestsMade = 0;
  30. this._resultsToSend = args.maxResults === -1 ? Infinity : args.maxResults;
  31. this._otherArgs = [];
  32. }
  33. /* eslint-disable @typescript-eslint/no-explicit-any */
  34. end(...args) {
  35. this._ended = true;
  36. return super.end(...args);
  37. }
  38. _read() {
  39. if (this._reading) {
  40. return;
  41. }
  42. this._reading = true;
  43. // Wrap in a try/catch to catch input linting errors, e.g.
  44. // an invalid BigQuery query. These errors are thrown in an
  45. // async fashion, which makes them un-catchable by the user.
  46. try {
  47. this._requestFn(this._nextQuery, (err, results, nextQuery, ...otherArgs) => {
  48. if (err) {
  49. this.destroy(err);
  50. return;
  51. }
  52. this._otherArgs = otherArgs;
  53. this._nextQuery = nextQuery;
  54. if (this._resultsToSend !== Infinity) {
  55. results = results.splice(0, this._resultsToSend);
  56. this._resultsToSend -= results.length;
  57. }
  58. let more = true;
  59. for (const result of results) {
  60. if (this._ended) {
  61. break;
  62. }
  63. more = this.push(result);
  64. }
  65. const isFinished = !this._nextQuery || this._resultsToSend < 1;
  66. const madeMaxCalls = ++this._requestsMade >= this._maxApiCalls;
  67. if (isFinished || madeMaxCalls) {
  68. this.end();
  69. }
  70. if (more && !this._ended) {
  71. setImmediate(() => this._read());
  72. }
  73. this._reading = false;
  74. });
  75. }
  76. catch (e) {
  77. this.destroy(e);
  78. }
  79. }
  80. }
  81. exports.ResourceStream = ResourceStream;
  82. //# sourceMappingURL=resource-stream.js.map