如何在 Nodejs 中使用来自 sse 服务器的数据？答案

【问题标题】：How to consume data from a sse server in Nodejs?如何在 Nodejs 中使用来自 sse 服务器的数据？
【发布时间】：2021-01-17 19:05:35
【问题描述】：

我有一个定期发送数据的服务器。我想在我的 REST API 中使用这些数据，如何获取它？当从外部源推送数据时需要自动调用。我尝试了以下代码，但没有成功。

var EventSource = require("eventsource");

var url = "..." // Source URL
    var es =  new EventSource(url);
    es.onmessage = (event) => {
        console.log(event)
        const parsedData = JSON.parse(event.data);
        console.log(parsedData)
    }

【问题讨论】：

为什么要使用 REST API 中的数据？ REST API 和 SSE 是不同的方法。 REST API由客户端发起，客户端等待响应，与SSE不同。
我有一个定期向我提供数据的外部源，我需要提取数据，进行一些修改并将其存储到数据库中。这就是我想在后端休息服务器中使用它的原因。
当该外部源推送数据时，您必须通过订阅其事件来监听它并将其保存到数据库中（或者您也可以考虑为此编写一个 webhook）。因此，每当您下次调用 REST API 时，您都可以从数据库中获取最新数据。
你能提供一个代码示例吗？我试过了，但它不起作用。检查我上面尝试的代码。

标签： javascript node.js express sockets server-sent-events

【解决方案1】：

为了测试目的，在服务器上设置类似的东西。创建一个带有事件名称的流，以便您可以在客户端上监听它。

const SseStream = require('ssestream')

app.get('/sse', (req, res) => {
  console.log('new connection')

  const sseStream = new SseStream(req)
  sseStream.pipe(res)
  const pusher = setInterval(() => {
    sseStream.write({
      event: 'server-time',
      data: new Date().toTimeString()
    })
  }, 1000)

  res.on('close', () => {
    console.log('lost connection')
    clearInterval(pusher)
    sseStream.unpipe(res)
  })
})

在客户端上，您会像这样监听事件

var EventSource = require('eventsource')
var es = new EventSource(url)
es.addEventListener('message', function (e) {
  console.log(e.data)
})

【讨论】：

我想在我的快递服务器上使用 sse，而不是在客户端。我已经有一个内部资源，可以定期为我提供数据。我只需要使用它，但我无法使用我提供的代码来完成它。
不行，我试过了在我的代码中，检查我上面放的示例。
它带有一个事件侦听器，该侦听器带有来自服务器的事件名称。和你的代码不一样。
如果我不知道活动名称怎么办？不能在 sse 服务器上监听任何事件吗？我可以用其他语言做到这一点，但在 nodejs 中找不到任何东西。
是的，您可以收听“消息”，请参阅编辑后的答案。

【解决方案2】：

当我想从不同的微服务使用 SSE 时，我遇到了同样的问题，我采用了这种方法，它对我有用。

node.js 文件

const  eventSource = require('eventsource');

  async socEventStream(req, res) {

    // list of the event you want to consume

    const list = ['EVENT1_NAME', 'EVENT2_NAME','EVENT3_NAME'];
    try {
      const e = new eventSource('url//of_sse_event', {});
      for (const l of list) {
        e.addEventListener(l, (e) => {
          const data = e.data;

         // Your data
          console.log('event data =====>',data)
          
        });
      };
      res.on('close', () => {
        for (const l of list) {
          e.removeEventListener(l, (e) => {
          });
        }
      })
    } catch (err) {
      console.log(err)
    }
  }

如果你想在 node.js 上消费事件并将其发送给客户端，那么


const  eventSource = require('eventsource');

  async socEventStream(req, res) {
   // setting express timeout for more 24 hrs
   req.setTimeout(24 * 60 * 60 * 1000);

    // setting headers for client to send consumed sse to client

    const headers = {
      'Content-Type': 'text/event-stream',
      'Connection': 'keep-alive',
      'Cache-Control': 'no-cache',
      'Access-Control-Allow-Headers': 'Content-Type, Access-Control-Allow-Headers, Authorization, X-Requested-With,observe,x-access-key',
      'Access-Control-Allow-Methods': 'POST, PUT, GET, OPTIONS, DELETE',
      'Access-Control-Allow-Origin': '*',
    };
    res.setTimeout(24 * 60 * 60 * 1000);
    res.writeHead(200, headers);

    // list of the event you want to consume

    const list = ['EVENT1_NAME', 'EVENT2_NAME','EVENT3_NAME'];
    try {
      const e = new eventSource('url//of_sse_event', {});
      for (const l of list) {
        e.addEventListener(l, (e) => {
          const data = e.data;

         // Your data
          res.write(`event:${l}\ndata:${data}\n\n`);
          
        });
      };
      req.on('close', () => {
        for (const l of list) {
          e.removeEventListener(l, (e) => {
          });
        }
      })
    } catch (err) {
      console.log(err)
    }
  }

【讨论】：