"use strict";

Object.defineProperty(exports, "__esModule", {
  value: true
});
exports.readableStreamReaderIntoObservable = readableStreamReaderIntoObservable;
var _rxjs = require("rxjs");
/*
 * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
 * or more contributor license agreements. Licensed under the Elastic License
 * 2.0; you may not use this file except in compliance with the Elastic License
 * 2.0.
 */

function readableStreamReaderIntoObservable(readableStreamReader) {
  return new _rxjs.Observable(subscriber => {
    let lineBuffer = '';
    async function read() {
      const {
        done,
        value
      } = await readableStreamReader.read();
      if (done) {
        if (lineBuffer) {
          subscriber.next(lineBuffer);
        }
        subscriber.complete();
        return;
      }
      const textChunk = new TextDecoder().decode(value);
      const lines = textChunk.split('\n');
      lines[0] = lineBuffer + lines[0];
      lineBuffer = lines.pop() || '';
      for (const line of lines) {
        subscriber.next(line);
      }
      return read();
    }
    read().catch(err => subscriber.error(err));
    return () => {
      readableStreamReader.cancel().catch(() => {});
    };
  }).pipe((0, _rxjs.share)());
}