RXJS：根据传入的可观察值发出额外的值

Question

我是 RxJs 6.0（或任何 RxJs 版本）的新手，虽然我看到它有多么强大，但一些简单的概念让我无法理解。

我有一种情况，我想根据源流向输出流发出一个额外的值，但对于我的一生，我不知道该怎么做。我真的需要一个可以采用方法而不是静态值的 startsWith 运算符，然后我可以实现这一点。这是一些设置场景的愚蠢代码。

import { startWith, scan, tap, mergeMap, map, concat } from 'rxjs/operators';

interface IData {
  data: number;
  emitExtraVal: boolean;
}

class obsData implements IData {
  constructor(data: number) {
    this.data = data;
    this.emitExtraVal = false;
  }
  public data: number;
  public emitExtraVal: boolean;

}

class extraData implements IData {
  constructor(data: number) {
    this.data = data;
    this.emitExtraVal = true;
  }

  public data: number;
  public emitExtraVal: boolean;
}
const sourceOne = of(new obsData(1),new obsData(2),new obsData(3));
/*const finalSource = sourceOne.pipe(
  map((sData) => <IData>new extraData(sData.data)),
  map((sData) => sData)
);*/
const finalSource = sourceOne.pipe(
  mergeMap((sData) => concat(of(<IData>new extraData(sData.data), of(sData))))
);
const subscribe = finalSource.subscribe(val => console.log('Data:' + val.emitExtraVal));

我想要做的是输出一个 extraData 的实例，其中包含 obsData 中的数字，然后是我刚刚从源中获取的 obsData。这不是我正在尝试的确切场景，但它展示了我正在尝试做的事情的核心，即创建一个额外的输出，然后创建另一个输出，这两者都依赖于单个源输入。

此问题的更新示例基于 cmets，但此示例无法运行，因为语法不正确

这会产生以下错误：

您提供了 'function (source) { return source.lift.call(concat_1.concat.apply(void 0, [source].concat(observables))); }' 需要流的地方。您可以提供 Observable、Promise、Array 或 Iterable。

---更新--- 由于响应，这是有效的最终答案。我遇到的主要问题是您可以从 rxjs/operators 或 rxjs 导入 concat。如果您在管道命令中使用它，则必须从 rxjs 导入它。

// RxJS v6+
import { of, fromEvent, combineLatest, concat } from 'rxjs';
import { startWith, scan, tap, mergeMap, map } from 'rxjs/operators';

interface IData {
  data: number;
  emitExtraVal: boolean;
}

class obsData implements IData {
  constructor(data: number) {
    this.data = data;
    this.emitExtraVal = false;
  }
  public data: number;
  public emitExtraVal: boolean;

}

class extraData implements IData {
  constructor(data: number) {
    this.data = data;
    this.emitExtraVal = true;
  }

  public data: number;
  public emitExtraVal: boolean;
}
const sourceOne = of(new obsData(1),new obsData(2),new obsData(3));

const finalSource = sourceOne.pipe(
  mergeMap((sData) => concat(of(<IData>new extraData(sData.data), <IData>sData)))
);
const subscribe = finalSource.subscribe(val => console.log('Data:' + val.emitExtraVal));

Answer 1

创建一个额外的输出，然后创建另一个输出，两者都依赖于单个源输入。

在我的示例中，dataItem 是“单一源输入”，它使用 from 运算符转换为可观察的 it 元素。稍后您可以使用flattening operators（如mergeMap 或concatMap，具体取决于要求）将所有这些Observables“扁平化”为一个。看例子：

const { Observable, of, from, concat } = rxjs; // = require("rxjs")
const { mergeMap } = rxjs.operators; // = require("rxjs/operators")

const complexAjaxCall = id => of(`${id}-from-ajax`);

const ids = [1, 2, 3];

from(ids).pipe(
  mergeMap(id => concat(
    complexAjaxCall(id),
    of(id)
  )),
).subscribe(e => console.log(e));

<script src="https://unpkg.com/rxjs@6.2.2/bundles/rxjs.umd.min.js"></script>