在设置 RxJs 订阅后，是否有一种干净的方法可以立即执行函数？

Question

我想做的事情：

向 websocket 服务器发送 'request' 事件。
接收与一些附加数据捆绑在一起的即时响应事件。
随着时间的推移不断收到回复。

我的问题是：

是否有更简洁的方法来完成我在下面所做的事情？不使用 setTimeout。

首先，看一下这个简化的工作示例：

const { Subject, defer, interval, of} = rxjs;
const EventEmitter = EventEmitter3;


const emitter = new EventEmitter();
const subject = new Subject();

// The next lines mock a websocket server message listener, imagine this being present on server side
emitter.on("request", () => {
  subject.next(`Immediate response`);
  interval(1000).subscribe((index) =>
    subject.next(`Delayed response...${index}`)
  );
});



// Imagine the following code being present in the browser
function getData() {
  return defer(() => {
    // The next line mocks a websocket client sent event
    setTimeout(() => emitter.emit("request"), 0);
    return subject;
  });
}

getData().subscribe((data) => console.log(data));

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.3/rxjs.umd.min.js"></script>
<script src="https://unpkg.com/eventemitter3@latest/umd/eventemitter3.min.js"></script>

现在看一个无法工作的例子。在这里，客户端不会立即得到响应，因为 'request' 事件是在订阅设置之前发送的。

const { Subject, defer, interval, of} = rxjs;
const EventEmitter = EventEmitter3;


const emitter = new EventEmitter();
const subject = new Subject();

// The next lines mock a websocket server message listener, imagine this being present on server side
emitter.on("request", () => {
  subject.next(`Immediate response`);
  interval(1000).subscribe((index) =>
    subject.next(`Delayed response...${index}`)
  );
});


// Imagine the following code being present in the browser
function getData() {
  return defer(() => {
    emitter.emit('request');
    return subject;
  });
}

getData().subscribe((data) => console.log(data));

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.3/rxjs.umd.min.js"></script>
<script src="https://unpkg.com/eventemitter3@latest/umd/eventemitter3.min.js"></script>

Answer 1

您可以使用 BehaviorSubject 而不是 Subject:

const { BehaviorSubject, defer, interval, of } = rxjs;
const EventEmitter = EventEmitter3;


const emitter = new EventEmitter();
const subject = new BehaviorSubject();

// The next lines mock a websocket server message listener, imagine this being present on server side
emitter.on("request", () => {
  // the timing of this synchronous response would be impossible in a real socket
  subject.next(`Immediate response`);
  interval(1000).subscribe((index) =>
    subject.next(`Delayed response...${index}`)
  );
});


// Imagine the following code being present in the browser
function getData() {
  return defer(() => {
    emitter.emit('request');
    return subject;
  });
}

getData().subscribe((data) => console.log(data));

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.3/rxjs.umd.min.js"></script>
<script src="https://unpkg.com/eventemitter3@latest/umd/eventemitter3.min.js"></script>

请记住，在发送套接字请求后服务器的响应不可能同步返回，这与您为此演示实现模拟的方式相同，因此虽然 Subject 不会产生此处的预期结果，它可以在真正的套接字上正常工作。

在设置 RxJs 订阅后，是否有一种干净的方法可以立即执行函数？

Is there a clean way of executing a function immediately after an RxJs subscription is set up?

javascript

events

websocket

rxjs