How to consume data from a sse server in Nodejs?

Question

I have a server that emits data at regular intervals. I want to use this data in my rest API, how do I fetch it? It needs to be automatically called when the data is pushed from the external source. I have tried the following code but it did not work.

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)
    }

Answer 1

I had the same problem when I wanted to consume the SSE from a different microservice , I followed this approach and it worked for me.

node.js file

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)
    }
  }

if you want to consume event on node.js and send it to client then

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)
    }
  }

Answer 2

For testing purposes set up something like this on the server. Create a stream with and event name so you can listen for it on the client.

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)
  })
})

And on the client you listen for the event like this

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