RxJs 在订阅中获取原始事件而不传递

Question

一点伪代码解释一下标题：

   Subject source = new Subject();

   function sendInNewResource(newNumber, anotherVariable) {
       source.next(newNumber);   
   }

   const newSource = source
     .map((myNumber) => myNumber++);

   newSource.subscribe((data) => {
       //How can I access myObject here?
   });

   sendInNewResource(1, {myObject});

我认为伪代码解释了我想做的一些事情。 到目前为止，我已经尝试过扩展 observable，但如果我这样做了，看起来好像我需要覆盖 observable 中的运算符。

我几乎想为活动提供元数据。

我不想将 myObject 向下传递，因为这是一个供外部使用的库，我将在我的代码库中订阅可观察对象，但链接可观察对象将在外部完成。

有人对如何实现这一点有任何想法吗？

对于上下文， 我正在尝试创建一个纯粹的 RxJs http 库，它允许我执行以下操作：

const server$ = RxHttpServer(server)

const api$ = server.filter((request) => request.url.startsWith('/api'));

const apiEndpoint$ = api.map(() => ({
   status: 400,
   body: 'this is an api endpoint',
   headers: {
   }
}));

// Inside my library, note the use of the response object that I don't have access to:
apiEndpoint.subscribe((dataToSend) => response.write(dataToSend))

这些对象的内部逻辑和处理已经写好了，只需要返回那个响应对象。

Answer 1

您可以使用 responseFor 帮助器，它会获取响应元数据并返回一个可观察的响应主题，每个 HTTP 请求都有一个。图书馆用户可以订阅它并将数据写入每个响应主题。虽然您在库中可以订阅相同的响应主题，并在完成后将其发送回客户端。比如：

// even better extending the Observable prototype
// const apiEndpoint$ = api$.mapResponseFor(...
const apiEndpoint$ = api$.map(responseFor({
   status: 400,
   body: 'this is an api endpoint',
   headers: {}
}));

apiEndpoint$.subscribe(response => {
  response.next(data)
  response.complete()
})