如何做一个轻量级的只能订阅一次的`Replay`运算符？
要实现一个轻量级的只能订阅一次的Replay运算符，可以按照以下步骤进行：
- 首先，需要定义一个自定义的
Replay类，该类实现了Observable接口，用于创建可观察对象。 - 在
Replay类中，需要定义一个内部缓存队列，用于存储订阅者订阅时产生的数据。 - 实现
Replay类的subscribe方法，该方法用于订阅可观察对象。在订阅时，首先判断缓存队列是否为空，如果不为空，则将缓存队列中的数据发送给订阅者，并清空缓存队列。如果缓存队列为空，则将订阅者添加到订阅列表中。 - 实现
Replay类的next方法，该方法用于向缓存队列中添加数据。当有新的数据产生时，将数据添加到缓存队列中，并遍历订阅列表，将数据发送给所有订阅者。 - 实现
Replay类的unsubscribe方法，该方法用于取消订阅。当订阅者取消订阅时，将其从订阅列表中移除。
以下是一个示例代码：
class Replay:
def __init__(self):
self.subscribers = []
self.cache = []
def subscribe(self, subscriber):
if self.cache:
for data in self.cache:
subscriber(data)
self.cache = []
else:
self.subscribers.append(subscriber)
def next(self, data):
self.cache.append(data)
for subscriber in self.subscribers:
subscriber(data)
def unsubscribe(self, subscriber):
self.subscribers.remove(subscriber)这样，我们就实现了一个轻量级的只能订阅一次的Replay运算符。可以通过创建Replay对象，然后调用subscribe方法进行订阅，调用next方法添加数据，调用unsubscribe方法取消订阅。
请注意，以上代码仅为示例，实际使用时可能需要根据具体需求进行适当的修改和扩展。
相关·内容
我们是否需要在一个可观察的对象中返回我们自己的拆卸逻辑或取消订阅方法？
const observable = new Observable(function subscribe(subscriber) {
// Keep track of the interval resource
const intervalId = setInterval(() => {
subscriber.next('hi');
}, 1000);
// Provide a way of canceling and disposing the interval resou
浏览 5提问于2021-03-24得票数 0
如果我有一个可观察的对象o：
let o: Observable<Object> = ...
我可以订阅这个对象，但是为什么不允许订阅Subscriber对象。举一个真实世界的例子：
myServiceCall() {
let o: Observable<Object> = ...
return o.subscribe(...)
}
ngOnInit() {
this.myServiceCall().subscribe(...) // Not allowed on a Subscriber object
}
有了prom
浏览 1提问于2016-05-27得票数 0
我对RxScala可观察到的东西并不熟悉，当我使用RxScala(N)和doOnCompleted()的组合时，我正经历着奇怪的行为。
下面我有一个测试示例，其中我认为第一个订阅是正确的(在开始时是采取(2))，并输出订阅、下一步、完成和取消订阅方法的所有日志，但是第二个测试(在doOnComplete之后采取(2))永远不会到达doOnCompleted方法。
import rx.lang.scala.Observable
object Tester extends App {
val obs = Observable.from(List(1,2,3,4))
val o
浏览 0提问于2018-04-26得票数 3
回答已采纳
3回答
我创建了一个可观察的，生成潜在无限数量的数据(例如，计时器)。这些数据是通过一个主题访问的，因此多个观察者将接收相同的值。
如何阻止可观察到的产生新的价值？(不修改可观察到的实现)
// custom Observable, to visualize internal behavior
const timer$ = new rxjs.Observable(subscriber => {
console.log("observable init");
var counter = 0;
const intervalId = setInterval(() =>
浏览 0提问于2018-12-10得票数 3
回答已采纳
我需要一个类似于exahustMap的运算符，但它会记住最后跳过的可观察对象，并在当前可观察对象完成后执行它。
例如，考虑exhaustMap的大理石图
在我的例子中，在发出蓝色的值之后，它将被三个值50所跟随。当然，在本例中，它看起来就像concatMap，但是如果在3和5之间也有一个4，那么它就不会反映在输出中。
我已经设法编写了自己的运算符，类似于exhaustMap的实现方式：
function exhaustLatestMap<T, R>(project: (value: T) => Subscribable<R>): OperatorFunc
浏览 2提问于2018-08-13得票数 4
我正在尝试将我们在代码库中使用的模式提取到一个更通用、更可重用的构造中。但是，我似乎无法让泛型类型注释与mypy一起工作。
我得到的是：
from abc import (
ABC,
abstractmethod
)
import asyncio
import contextlib
from typing import (
Any,
Iterator,
Generic,
TypeVar
)
_TMsg = TypeVar('_TMsg')
class MsgQueueExposer(ABC, Generic[_TMsg]):
浏览 0提问于2018-06-26得票数 12
回答已采纳
1回答
这是我用socket io连接客户端到服务器的方式。
export class OverviewService {
socket:any;
readonly uri:string='http://localhost:4000';
constructor(private http: HttpClient) {
this.socket = io(this.uri);
}
当我打电话的时候
getOnPeak(){
this.socket.emit('on_peak/today');
return new Obs
浏览 5提问于2022-02-16得票数 1
在我的Angular 8应用程序中，我有一个订阅服务并等待服务已加载的通知的组件，如下所示：
constructor(public contentService: ContractService) {
let self = this;
let sub = contentService.loaded.subscribe(function(loaded) {
if (loaded) {
self.loading = false;
if (sub) {
sub.unsubscribe();
浏览 0提问于2019-10-21得票数 10
回答已采纳
我尝试在多播中使用Subject来完成，但它没有显示在complete()方法中设置的完成消息。原因何在？以及如何使用error和complete方法？
let source = new Observable(subscriber => {
subscriber.next(1);
subscriber.next(2);
return () => {
subscriber.complete();
console.log('Unsubscribed');
}
});
let subject = new Subjec
浏览 3提问于2021-12-03得票数 0
我有一个服务，它使用可观察的方法公开一些全局状态。必须通过发出HTTP调用来初始化状态。我希望状态的所有使用者都接收相同的值，在订阅可观测值时接收当前值，并在状态更改时接收更新。
这个场景通常使用Subject来处理，它可以通过订阅HTTP请求返回的可观察到的内容来初始化。
但是，我不希望在第一次订阅全局状态之前执行HTTP请求。如果使用subscribe方法的Subject，则调用Subject请求。
我在服务中有以下代码。这是可行的，但似乎应该有一个更简单的方法来做到这一点。
private initialized = false;
private readonly subject
浏览 2提问于2022-05-04得票数 1
回答已采纳
当将publish()与observeOn和subscribeOn结合使用时，我会观察到一些奇怪的行为。请看一看民间的例子。
代码：
ConnectableObservable<String> observable = Observable.create(
new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
int i=0;
浏览 2提问于2015-05-15得票数 2
回答已采纳
1回答
如何定义RX主题上的运算符？我希望能够在不更改原始源流的情况下对数据流进行节流/获取/去抖动。我尝试了以下(失败的)实现。感谢您的帮助。
var source = Rx.Observable.interval(1000).take(10);
var subject = new Rx.Subject();
source.subscribe(subject);
var subscriber1 = subject.subscribe(
function (x) { $("#result1").append('next: ' + x + '<b
浏览 1提问于2016-01-27得票数 2
我正在为一个订阅者而苦苦挣扎，它会在出错时自动取消订阅：
observable
.subscribe(
(data) => {
// some logic encountering an error during execution
// comparable to throw new Error()
}
)
我可以使用try /catch来防止它：
observable
.subscribe(
(data) => try {
// some logic encountering an error dur
浏览 9提问于2017-07-08得票数 6
我在找一个主题(或类似的东西)可以：
如果没有订阅服务器，则可以接收项并将其保存在队列或缓冲区中。
一旦我们有了订阅者，所有的项目都会被消耗，并且再也不会被发出。
我可以订阅/取消订阅/退出主题
BehaviorSubject几乎可以完成这项工作，但它保留了最后观察到的项。
更新
在接受答案的基础上，对单个观测项目给出了类似的解。还添加了取消订阅部分，以避免内存泄漏。
class LastEventObservable private constructor(
private val onSubscribe: OnSubscribe<Any>,
浏览 1提问于2016-06-09得票数 20
回答已采纳
3回答
当我每秒钟调用一个请求时，我会得到以下错误：
core.js:5967 ERROR TypeError:无法读取null的属性“删除”
at LoadingBackdropService.hide (loading backdrop.service.ts:18)
在TapSubscriber._tapNext (payment.component.ts:57)
在TapSubscriber._next (tap.js:40)
在TapSubscriber.next (Subscriber.js:49)
在SwitchMapSubscriber.notifyNext (switchMap.js
浏览 0提问于2021-01-09得票数 0
1回答
我有一个API调用，我想用可观察到的方式包装它：
private Observable<RealmResults<Account>> getAccounts() {
final Observable<RealmResults<Account>> realmAccounts =
Observable.defer(new Func0<Observable<RealmResults<Account>>>() {
@Override
浏览 2提问于2015-03-26得票数 2
回答已采纳
从这个
RxJS引入了可观测系统，这是一种新的JavaScript推送系统。一个可观察的人是多重价值的生产者，把他们“推”给观察者(消费者)。订阅一个可观察到的函数类似于调用一个函数。
要调用可观察到的数据，我们应该从可观测对象本身调用subscribe()函数，并将观察者作为可观测对象传递的数据的使用者传递，例如：
observable.subscribe( { /*this is an observer*/ } );
另外，这个说：
什么是观察者？观察者是一个可观察到的价值的消费者。观察者只是一组回调，每个类型的通知都由可观察的: next、error和complete提供。下面
浏览 1提问于2021-04-04得票数 6
回答已采纳
2回答
我使用节点-oracledb连接到Oracle数据库。API提供了自己的承诺，这些承诺可以转换为Promise<T>，从而“转换”为Observable<T>。
利用可观测数据，我想：
打开数据库连接
选择N条记录
关闭数据库连接，即使#2抛出异常。
使用传统的、阻塞的、程序化的方式，它应该是这样的：
try
{
connection = Oracle.getConnection(...);
resultSet = connection.execute("SELECT ... FROM ...");
}
catch (
浏览 1提问于2018-08-01得票数 4
回答已采纳
在下面的代码中，new Observable(function(subscriber)中匿名函数调用中的observer对象是observer对象的包装器吗？
const observer = {
next: (data: any) => console.log('Observer got a next value: ', data),
error: (err: any) => console.error('Observer got an error: ' + err),
complete: () => console.
浏览 4提问于2022-03-12得票数 0
回答已采纳
在rxjs5中，我有一个AsyncSubject并希望多次订阅它，但是只有一个订阅者可以接收next()事件。所有其他(如果尚未取消订阅)都应该立即获得没有complete()的next()事件。
示例：
let fired = false;
let as = new AsyncSubject();
const setFired = () => {
if (fired == true) throw new Error("Multiple subscriptions executed");
fired = true;
}
let subscription
浏览 5提问于2016-10-27得票数 9
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储活动推荐
腾讯云开发者
