如何在异步管道上使用自定义管道？

我正在尝试在异步管道上创建自定义管道，我尝试了很多解决方案，但仍然不起作用。这是代码片段。

Product.sort.ts - 自定义管道

import { PipeTransform, Pipe } from '@angular/core';
import { Observable } from 'rxjs/Observable';

@Pipe({
    name: 'sortByName'
})
export class ProductPipe implements PipeTransform{
    /*transform(values: Array<any>, term:string){
        return values.filter(obj => obj.pname.startsWith(term))
    }*/

    //CODE NOT WORKING >>>>>
    transform($value: Observable<Array<any>>, term:string){
        if($value){
            $value.subscribe(
                (obj) => {
                    return obj.filter(obj => obj.pname.startsWith(term))
                }
            )
        }
    }
}

products.component.ts - 主要组件

import { Component, OnInit } from '@angular/core';
import { Store } from '@ngrx/store';
import { Observable } from 'rxjs/Observable';
import { AppService } from '../app.service/app.service';
import { ProductPipe } from '../products.sort/products.sort';

@Component({
    selector: 'products-pg',
    template: `
        Products List:
        <ul>
            <li *ngFor = 'let product of $productList | async | sortByName:"A"'>{{product.pname}}</li>
        </ul>
    `
})
export class ProductsComponent implements OnInit{
    private $productList:Observable<Array<any>>;

    constructor(private _service: AppService, private _store: Store<Array<any>>){}

    ngOnInit(){
        this._service.setProductList();
        this.$productList = this._store.select('products');
    }
}

在这里，我使用 store 进行状态管理，我尝试按名称排序，因此传递“A”作为第一个字母。自从$产品列表是可观察的，如何编写处理这样的异步行为的管道，请帮我解决这个问题。

我也遇到了同样的问题，解决方法如下：

<li class="list-group-item" *ngFor='let alert of _alerts$ | ipwrAlertFilter:seeAll | async'>

看看我之前是如何使用自定义过滤器的异步管道，这样我的自定义管道就会得到一个可观察的，然后在我的管道中我有这个：

return value.map(data => data.filter(x => x.originalHasBeenSeen === false));

这样我的自定义管道仍然返回我仍然可以应用的东西async管道打开。因此，对于流中的每个新项目，我的自定义管道仍然会受到影响，并且我的视图也会更新。 希望这可以帮助。