是否有可以在组件内使用的等效异步管道？

Question

是否存在与async 管道等效的东西，我可以在这样的组件中使用它

   @Component({
      selector: 'my-component',
    })

    export class myComponent {

      myObservable$: Observable<string>;

      method() {
        doSomething(this.myObservable$);
        // here I would like to access directly the current string value of
        // myObservable$ without having to subscribe 
      }
    }

Answer 1

你必须问自己：你想通过避免subscribe() 调用来实现什么？我的猜测是，您想防止保留订阅并不得不手动取消订阅。

如果是这种情况，你只需要确保你得到一个完整的 Observable。那么以后就不需要退订了。您可以将任何 Observable（有限或无限）转换为最终完成的。

这是您的案例的示例：

  method() {
    this.myObservable$.take(1).subscribe(
      val => doSomething(val)
    );
  }

正确的方法实际上取决于你的 observable 做了什么以及你想用这个值做什么。例如，Angular2 http 调用自行完成，no need to unsubscribe!

或者，如果你想为你的 observable 中的每个新值调用 doSomething，上述解决方案将不适合，因为你只获得第一个值，然后 observable 完成。正如我所说，这归结为问题的背景。

Answer 2

不，没有。您需要手动订阅和手动取消订阅以避免内存泄漏。

对于简单的订阅，您可能会很想这样做：

@Component({
  selector: 'my-component',
})
export class myComponent implements OnInit, OnDestroy {
  public myObservable$: Observable<string>;
  private myObservableSub: Subscription;

  ngOnInit() {
    this.myObservableSub = this
      .myObservable$
      .subscribe(_ => {
        // do something
      });
  }

  ngOnDestroy() {
    this.myObservableSub();
  }
}

但是如果你有很多订阅呢？ 您是否应该这样做：

@Component({
  selector: 'my-component',
})
export class myComponent implements OnInit, OnDestroy {
  public myObservable1$: Observable<string>;
  private myObservableSub1: Subscription;

  public myObservable2$: Observable<string>;
  private myObservableSub2: Subscription;

  public myObservable3$: Observable<string>;
  private myObservableSub3: Subscription;

  ngOnInit() {
    this.myObservableSub1 = this
      .myObservable1$
      .subscribe(_ => {
        // do something
      });

    this.myObservableSub2 = this
      .myObservable2$
      .subscribe(_ => {
        // do something
      });

    this.myObservableSub3 = this
      .myObservable3$
      .subscribe(_ => {
        // do something
      });
  }

  ngOnDestroy() {
    this.myObservableSub1();
    this.myObservableSub2();
    this.myObservableSub3();
  }
}

---

**答案是不**

---

更新回复 (16/11/20)

在原始答案中（请参阅此编辑后），我建议使用 take until 但这会强制您创建一个主题并在组件被销毁时触发一个事件。虽然这背后的想法很好，但在我们订阅流的所有组件中放入相当多的样板。

除此之外，我们可以创建一个自定义运算符，我们可以调用 takeUntilDestroyed 并执行以下操作：

@Component({
  selector: 'my-component',
})
export class myComponent implements OnInit, OnDestroy {
  public myObservable1$: Observable<string>; // define your observable

  public myObservable2$: Observable<string>; // define your observable

  public myObservable3$: Observable<string>; // define your observable

  ngOnInit() {
    this
      .myObservable1$
      .pipe(takeUntilDestroyed(this))
      .subscribe(_ => {
        // do something
      });

    this.myObservableSub2 = this
      .myObservable2$
      .pipe(takeUntilDestroyed(this))
      .subscribe(_ => {
        // do something
      });

    this.myObservableSub3 = this
      .myObservable3$
      .pipe(takeUntilDestroyed(this))
      .subscribe(_ => {
        // do something
      });
  }

  ngOnDestroy() {}
}

这个自定义操作符的实现可以在这里找到：https://github.com/cloudnc/ngx-sub-form/blob/1115b21a007f72c54b521b3bed7c40051302145a/projects/ngx-sub-form/src/lib/shared/ngx-sub-form-utils.ts#L145-L148

原始回复

您应该执行以下操作：

@Component({
  selector: 'my-component',
})
export class myComponent implements OnInit, OnDestroy {
  private componentDestroyed$ = new Subject<void>();

  public myObservable1$: Observable<string>;
  public myObservable2$: Observable<string>;
  public myObservable3$: Observable<string>;

  ngOnInit() {
    this
      .myObservable1$
      .takeUntil(componentDestroyed$)
      .subscribe(_ => {
        // do something
      });

    this
      .myObservable2$
      .takeUntil(componentDestroyed$)
      .subscribe(_ => {
        // do something
      });

    this
      .myObservable3$
      .takeUntil(componentDestroyed$)
      .subscribe(_ => {
        // do something
      });
  }

  ngOnDestroy() {
    this.componentDestroyed$.next();
    this.componentDestroyed$.complete();
  }
}

如果您想了解更多相关信息，请查看 Ben Lesh 的一篇精彩文章：https://medium.com/@benlesh/rxjs-dont-unsubscribe-6753ed4fda87

编辑：
感谢@olsn，我编辑了我的答案并添加了next 行，因为确实完整并不会阻止其他流。

我创建了一个小 Plunkr 来演示该行为：https://plnkr.co/edit/dcueDqUgpqgYimxEAGUn?p=preview

Answer 3

如果您在Component 或可以访问ChangeDetectorRef 的地方，您可以执行以下小技巧：

@Component({
...
  providers: [AsyncPipe]
})
export class ExampleComponent {
  app$ = this.sandbox.allApps$;
  apps = this.pipeAsync.transform(this.app$);

  contructor(
    protected pipeAsync: AsyncPipe,
    protected sandbox: AppAppsSandbox
  ) {}
}

这假定访问名为AppAppsSandbox 的服务，但Observable app$ 可以来自任何地方，这是一种方便的解包方式。

Answer 4

实现目标的唯一方法（不使用订阅）是使用 BehaviorSubject，它有一个方法 getValue()。 Observables 和 BehaviousSubjects 之间存在一些差异（您可以在 documentation 中找到它们）。

var subject = new Rx.BehaviorSubject(current_subject_value);
subject.getValue()

然而有一些 stackoverflow 争论为什么你不应该使用 getValue 方法，而是使用订阅...然而有一个 observable take 运算符将完成可观察序列（因此不必管理订阅）

this.myObservable$.take(1).subscribe()

Answer 5

你可以使用我的图书馆。您可以创建 useReader 挂钩并将其包装在 AsyncPipe 组件中。看看例子。 https://www.npmjs.com/package/reactivex-react