Support for Parallel SSE Connections

[davpate]

Hi there! Been testing out the module and gotten it working using the build available in #15 with xk6 to connect to a single SSE endpoint, but when I'm trying to expand that to include multiple SSE endpoints the connection appears to "block" the Event Loop until client.close() is called.

Any recommendations on what I might be doing wrong? Might just be that the module isn't setup for that at this time, I tried looking around but I'm not very experienced with Go/k6 extension development so not sure where to start with a PR to add that functionality yet.

Here's a quick test script of what I'm looking for. In this I'd like both of the SSE connections to occur in parallel as my app will be getting data from multiple SSE endpoints at once. In the example below you'll notice that the connection isn't made to url2 until the first SSE connection is closed due to an event.id value of 4 being sent by https://echo.websocket.org/.sse. Given this my console output looks like this:

Current Console Output

INFO[0000] connected                                     source=console
INFO[0000] event id=1, name=server, data=1781505b56ee58  source=console
INFO[0000] event id=2, name=request, data=GET /.sse HTTP/1.1  source=console
INFO[0001] event id=3, name=time, data=2024-08-06T16:01:21Z  source=console
INFO[0002] event id=4, name=time, data=2024-08-06T16:01:22Z  source=console
INFO[0002] closing connection                            source=console
INFO[0002] started sse1                                  source=console
INFO[0003] connected                                     source=console
INFO[0003] event id=, name=, data={"testing":true,"sse_dev":"is great","msg":"It works!","now":1722960083350}  source=console
INFO[0005] event id=, name=, data={"testing":true,"sse_dev":"is great","msg":"It works!","now":1722960085350}  source=console

You can see from the above that the call to sse.open(...) appears to be blocking the event loop as the next line console.log('started sse1') isn't executed until after the client.close() function is called.

Example Script

import sse from "k6/x/sse"

export default function () {
  const url1 = "https://echo.websocket.org/.sse"
  const url2 = "https://sse.dev/test"
  const params = {
    method: 'GET',
    headers: {
      "Authorization": "Bearer XXXX"
    },
    tags: {"my_k6s_tag": "hello sse"}
  }

  const response1 = sse.open(url1, params, handleEvents);
  console.log('started sse1');
  const response2 = sse.open(url2, params, handleEvents);
  console.log('started sse2');
}

function handleEvents(client) {
  client.on('open', function open() {
    console.log('connected')
  })

  client.on('event', function (event) {
    console.log(`event id=${event.id}, name=${event.name}, data=${event.data}`)
    if (parseInt(event.id) === 4) {
      console.log('closing connection');
      client.close()
    }
  })

  client.on('error', function (e) {
    console.log('An unexpected error occurred: ', e.error())
  })
}

[kokokenada]

The blocking behavior of open defeats the purpose of sse support for my use case.

To expand on my use case the VU needs to receive SSE events in response to its actions or the actions of other VUs.

The only work around I can muster is to launch two VUs for the single simulated user.

Looking at https://github.com/phymbert/xk6-sse/blob/main/sse.go , is there a way to call k6's event loop or some sort of next()/yield in the "main control loop"? I couldn't find anything in the K6 docs.

And disturbingly, the WebSocket's docs (https://grafana.com/docs/k6/latest/using-k6/protocols/websockets/) says " the run function will be immediately called, all code inside it will be executed (usually code to set up event handlers), and then blocked until the WebSocket connection is closed". I guess their async stuff has a ways to go....

[phymbert]

I guess their async stuff has a ways to go....

Yes, k6 was designed syncIO, I reused the approach of the websocket extension.

Help is welcomed to introduce new async client.