标签: eventsource

在我们的应用程序中，我们公开了供外部服务使用的回调路由。当我们收到回调时，我们使用客户端/浏览器端的Eventsource和服务器端的抽筋向客户端订阅者发布更新。但是，有时，我们会受到来自此外部服务的回调请求的轰炸，这导致我们向客户端发布了很多更新。在Rails端是否有一种类似于javascript防反跳功能的方法，可以在收到的回调之间等待设置的时间来发布消息？

我们已经在使用sidekiq +线程，因此欢迎使用这些工具的建议。

我想从EventSource(服务器发送事件)创建一个RxJs Observable.

我尝试了以下方法:

import {Component, OnInit} from 'angular2/core';
import {Subject, Observable}  from 'rxjs/Rx';

@Component({
    selector: 'my-app',
    template: `<h1>My second Angular 2 App</h1>
    <ul>
        <li *ngFor="#s of someStrings">
           a string: {{ s }}
        </li>
    </ul>
    `
})
export class AppComponent implements OnInit {

    someStrings:string[] = [];

    ngOnInit() {
        let eventSource = new EventSource('/interval-sse-observable');
        let observable = Observable.create(eventSource);
        observable.subscribe({
            next: aString => this.someStrings.push(aString.data),
            error: err => console.error('something wrong occurred: ' + err)
        });
    }
}

但我得到以下异常:

EXCEPTION: Error: Uncaught (in promise): EXCEPTION: …

首先,我对ng2和打字稿很新.

我想要完成的是在Angular2组件中实现Server-Sent事件.我已经按照earlies帖子中提到的示例进行了操作,但我的问题是"EventSource"对象无法识别(在VS Code中为红色下划线).

我不确定我是否遗漏了一些参考资料......我的参考文献是:

<!-- IE required polyfills, in this exact order -->
  <script src="node_modules/es6-shim/es6-shim.min.js"></script>
  <script src="node_modules/systemjs/dist/system-polyfills.js"></script>
  <script src="node_modules/angular2/es6/dev/src/testing/shims_for_IE.js"></script>
  <script src="node_modules/angular2/bundles/angular2-polyfills.js"></script>
  <script src="node_modules/systemjs/dist/system.src.js"></script>
  <script src="node_modules/rxjs/bundles/Rx.js"></script>
  <script src="node_modules/angular2/bundles/angular2.dev.js"></script>

这就是我实现eventsource客户端的方式:

   ngOnInit() {
      const observable = Observable.create(observer => {
        const eventSource = new EventSource(/API_URL); //Cannot find EventSource
        eventSource.onmessage = x => observer.next(x.data);
        eventSource.onerror = x => observer.error(x);

        return () => {
            eventSource.close();
        };
    });

    observable.subscribe({
        next: guid => {
            this.zone.run(() => this.someStrings.push(guid));
        },
        error: err => console.error('something wrong occurred: ' + err)
    });

Strongloop Loopback是否支持更改流中的过滤？

以下内容适用于所有模型更改，但是我仅对模型部分的更改感兴趣。我可以将过滤添加到EventSource-URL吗？

var src = new EventSource('/api/mymodel/change-stream?_format=event-source');

var changes = createChangeStream(src);
changes.on('data', function(update) {
      console.log("data changed...", JSON.stringify(update));  
});

谢谢

我有一个页面，它在加载时立即通过EventSource开始流式传输事件。为了加速该流的启动，我想使用 HTTP 预加载元标记。但是当我将它添加到我的页面时：

<link rel="preload" as="fetch" crossorigin="crossorigin" href="http://myeventsource" /> 

Chrome 警告我：

已找到“ http://myeventsource ”的预加载，但未使用，因为请求标头不匹配。

这是真的：预加载的Accept标头为*/*，而源自 EventSource 的请求的Accept标头为text/event-stream。我一直在做的阅读表明该标头的内容由as元标记中的参数控制，但据我所知，没有特定于事件流的值。

我有什么办法可以覆盖Accept它发送的标头吗？

我有一个非常简单的节点服务，它公开了一个旨在使用服务器发送事件 (SSE) 连接的端点和一个非常基本的 ReactJs 客户端，通过 EventSource.onmessage 使用它。

首先，当我在 updateAmountState (Chrome Dev) 中设置调试点时，我看不到它被唤起。

其次，我得到 net::ERR_INCOMPLETE_CHUNKED_ENCODING 200（好的）。根据https://github.com/aspnet/KestrelHttpServer/issues/1858 “Chrome 中的 ERR_INCOMPLETE_CHUNKED_ENCODING 通常意味着在写入响应正文的过程中应用程序抛出了未捕获的异常”。然后我检查了服务器端，看看我是否发现了任何错误。好吧，我在 setTimeout(() => {... 的 server.js 中的几个地方设置了断点，我看到它定期运行。我希望每行只运行一次。所以看起来前端是尝试永久调用后端并收到一些错误。

整个应用程序，包括 ReactJs 的前端和 NodeJs 的服务器都可以在https://github.com/jimisdrpc/hello-pocker-coins 中找到。

后端：

const http = require("http");

http
  .createServer((request, response) => {
    console.log("Requested url: " + request.url);

    if (request.url.toLowerCase() === "/coins") {
      response.writeHead(200, {
        Connection: "keep-alive",
        "Content-Type": "text/event-stream",
        "Cache-Control": "no-cache"
      });

      setTimeout(() => {
        response.write('data: {"player": "Player1", "amount": "90"}');
        response.write("\n\n");
      }, 3000);

      setTimeout(() => {
        response.write('data: {"player": "Player2", …

客户端看起来像这样：

var es = new EventSource("http://localhost:8080");
es.addEventListener("message", function(e) {
    alert("e.data") //Alerts "" every couple seconds
})
es.addEventListener("error", function(e) {
    alert("error") //Also fires every couple of seconds
})
var post_request = new XMLHttpRequest();
post_request.open("POST", "http://localhost:8080");
post_request.setRequestHeader("Content-Type", "text/plain");
post_request.addEventListener("readystatechange", function() {
    if (post_request.readyState == 4 && post_request.status == 200) {
        alert(post_request.responseText); //This works
    }
})
post_request.send("This is a test")

服务器端 Node.js 处理 POST 请求如下所示：

function process_request(request, response) {
    var request_body = []
    request.on("data", function(chunk) {
        request_body.push(chunk)
    }) 
    request.on("end", function() {
        request_body = …

我正在尝试使用 Spring SseEmitter实现服务器发送事件，如此Youtube 视频中所述。

我能够启动事件流并从服务器发送的事件接收数据。

但是，我可以看到EventStream从客户端发出并到达服务器的多个类型的请求。我理解它的方式，EventSource应该发送一个HTTP请求，然后应该保持half duplex与服务器的连接，使用哪个服务器将事件发送到客户端。

为什么它会定期发送请求？是不是就像轮询而不是半双工连接？

波纹管是我正在使用的代码。

服务器代码

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter.SseEventBuilder;

@RestController
public class TestService {
    private List<SseEmitter> subscriberList = Collections.synchronizedList(new ArrayList<>());

    @RequestMapping("inbox")
    public SseEmitter inbox() {
        SseEmitter subscriber = new SseEmitter();
        subscriberList.add(subscriber);

        subscriber.onCompletion(() -> {
            subscriberList.remove(subscriber);
            System.out.println("Removed the completed event emitter");
        });

        System.out.println("Subscriber arrived");
        return subscriber;
    }

    @RequestMapping("message")
    public String message(@RequestParam("message") …

我正在使用带有处理服务器发送事件流的快速应用程序的节点服务器。这是通过启用 http2 的 NginX 代理的。SSE 事件通过 React 应用程序中的 EventSource 使用。我每 10 秒发送一次心跳消息以保持连接有效。

这一切都很好，直到出现某种形式的网络中断，例如让我的笔记本电脑进入睡眠状态然后重新唤醒它。

然后从那时起，流将每 40 秒左右出错一次，出现 net::ERR_HTTP2_PROTOCOL_ERROR 200 错误，然后重新连接而不是仅使用稳定流重新连接一次。

Firefox 工作正常。它不会出错并且仅重新连接一次。

如果我将 Node 配置为直接为 http2 提供服务，而不是通过 NGinx 作为测试（通过 spdy 库），那么一切都按预期工作，所以我认为这不是 Node 问题，而且我的 Nginx 配置和 Chrome 肯定缺少某些内容。

Nginx配置如下（位置/stream为SSE代理）

server {
    listen 28443 ssl http2;
    listen [::]:28443 ssl http2;

    server_name example.com;

    ssl on;
    ssl_certificate /etc/letsencrypt/live/example.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/example.com/privkey.pem;

    http2_max_field_size 16k;
    http2_max_header_size 128k;

    root /var/www/example.com;
    index index.html index.htm;

    location / {
        client_max_body_size 100M;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $remote_addr;
        proxy_pass http://localhost:28080;
        proxy_http_version 1.1;
        proxy_set_header …

我有一个使用 EventStream 的 ReactJS 客户端和一个实现 SSE 的 golang 后端。

当我将浏览器连接到在 localhost 上运行的后端时，以及当我的后端在带有端口转发的 k8s 上运行时，一切似乎都正常。

一旦我使用主机名创建入口（这样我就不必一直进行端口转发），SSE 就停止工作。我仍然看到客户端发送请求，并且该请求被后端接收并注册。但是，当发送事件时，它永远不会到达我的 ReactJS 应用程序。

我附上后端 SSE 实现的代码：

package sse

import (
    "encoding/json"
    "fmt"
    "net/http"
    "time"

    "go.uber.org/zap"

    "github.com/talon-one/towers/controller/api/log"
)

// the amount of time to wait when pushing a message to
// a slow client or a client that closed after `range clients` started.
const patience time.Duration = time.Second * 2

type customerStateUpdate struct {
    sseEvent
    CustomerName  string `json:"customer_name"`
    CustomerState string `json:"customer_state"`
}

type contentUpdate struct { …