RxJS 服务器轮询

Question

我需要轮询服务器并在客户端更新数据。为此，我有一个调度程序，它调度一个名为FRONT_PAGE 的操作。一旦应用程序启动，该操作就会被调度，并且客户端应该每秒发送两次请求。正在发送请求，但出现以下错误。

×
TypeError: Object(...)(...) is not a function
Observable.pipe
A:src/internal/Observable.ts:439
▶ 2 stack frames were collapsed.
doInnerSub
A:internal/operators/mergeInternals.ts:71
outerNext
A:internal/operators/mergeInternals.ts:53
  50 |         }
  51 |     }));
  52 | };
> 53 | source.subscribe(new OperatorSubscriber(subscriber, outerNext, function () {
     | ^  54 |     isComplete = true;
  55 |     checkComplete();
  56 | }));

代码。

import { from, of, timer } from 'rxjs'
import { ajax } from 'rxjs/ajax'
import { map } from 'rxjs/operators'
import { FRONT_PAGE } from './constants'

const poll_server = url => {
    timer(0, 500)
        .pipe(from(fetch(url))
            .pipe((x) => { console.log("Polling server.."); return x })
            .pipe(map(response => response.json())))
}

export const server_dispatch = action => {
    switch (action.type) {
        case FRONT_PAGE: {
            poll_server('http://localhost:8080/warnings')
        }
        default:
            return of(action)
    }
}

终点

app.get('/warnings', (req, res) => {
    console.log("[GET] /warnings")
    const baseline = req.query.baseline ?? -1
    if (version > baseline) {
        res.send(warnings(alerts.filter(a => a.prediction)))
    } else if (game) {
        res.status(204).send({})
    } else {
        res.status(404).send();
    }
})

Answer 1

1. poll_server 是一个几乎没有任何效果的函数，然后什么也不返回/void。

例如：

// return arg + 1
const lambdaFunction = arg => arg + 1;

// returns nothing
const lambdaFunction = arg => {arg + 1;};

// return arg + 1
const lambdaFunction = arg => {return arg + 1;};

---

2. .pipe(from(fetch(url)).pipe(...你在这里尝试什么？

from not 不是可管道操作符。您可以使用它从 Promise、observableLike 或迭代器创建可观察对象。不是可组合的运算符。

---

3. 下面的代码确实有效，因为运算符实际上只是函数（母鸡为什么它们可以组合得这么好！）。所以这里的身份函数很好，但我很确定它没有做你认为的那样。

.pipe((x) => { console.log("Polling server.."); return x })

console.log("Polling server..") 将在您的 observable 被订阅之前运行。在这种情况下，您的 observable 永远不会订阅（因为它永远不会返回），但我想您可能会在控制台中看到“Polling server..”。

---

4.

.pipe(x => { console.log("Polling server.."); return x })
.pipe(map(response => response.json())))

等同于：

.pipe(
  x => { console.log("Polling server.."); return x },
  map(response => response.json())
)

这是因为链接应用程序与组合是一回事。

---

5. server_dispatch 只为 switch 中的默认情况返回一个 observable。否则返回undefined。这真的是你想要的行为吗？