Simultaneous Writing and Reading of data using Array

Question

I am developing a Chrome extension that communicates with a remote API via a WebSocket connection using Socket.io. The API sends me streaming data in string format. My goal is to hold the streaming data in a buffer for a few seconds until I receive a specific message from the content script. Once the message "startStreaming" is received, I want to start streaming the data to the content script.

To achieve this, I am thinking to use a buffer array to store the incoming data from the server. E.g

let buffer = [];

socket.on('stream', (wordChunk) => {
    buffer.push(wordChunk);
});

if (msg.msg === 'startStreaming') {
    console.log('Send response back to Tab');
    buffer.forEach(wordChunk => {
        port.postMessage({ msg: 'streamData', wordChunk });
    });
}

But the problem I am facing is ensuring the simultaneous writing and reading of data in a safe manner. Since the data can be continuously streamed from the server while the buffer is being read and processed, I need to handle this situation properly to avoid any data inconsistencies.

I would greatly appreciate any suggestions or best practices on how to handle this scenario effectively, ensuring that the data is read in the FIFO (First-In, First-Out) order and avoiding any race conditions or data corruption issues.

Suggest me if there is any other way rather than array.

Scenario If we don't take Chrome Extension to the context, think it in such a way that, I want to pre-fetch api response before a user's action. E.g I want to start fetching data from the server and store it temporarily when a user just hover over the button. And display the response instantly when he clicks on the button to make him feel that it is super fast. So we don't know when the user clicks on the button, as soon as he hovers or after few times of hovering the button.

Thank you in advance for your assistance!

Answer 1

I think this might be a good candidate for a ReplaySubject from RXJS.

I've put an example below on how it could work in your situation, but in practice I wouldn't recommend the interval thing. It is only to demonstrate based on your code.

import { ReplaySubject } from "rxjs";

const subject = new ReplaySubject<any>()

socket.on('stream', wordChunk => {
  subject.next(wordChunk);
});

const interval = setInterval(() => {
  if(msg.msg === 'startStreaming') {
    subject.subscribe({
      next: (wordChunk) => port.postMessage({ msg: 'streamData', wordChunk })
    })
    clearInterval(interval);
  };
}, 1000);

I've built a sandbox to show a ReplaySubject in action.

Answer 2

Javascript code execution is single-threaded. This means your code is safe from thread interleaving issues, and the socket.on listener cannot run while the forEach loop is executing.

Your real issue is that the forEach loop will complete, and then the socket.on listener may run afterwards. Thus, the buffer may get new entries, but the contents of the buffer will not be iterated over and sent to the port.

Here is how you could code it instead:

let buffer = [];
let streamingEnabled = false;

socket.on('stream', wordChunk => {
  buffer.push(wordChunk);
  if(streamingEnabled) sendBuffer(); 
});

function sendBuffer() {
  buffer.forEach(wordChunk => port.postMessage({ msg: 'streamData', wordChunk }));
  buffer.length = 0; //clear the buffer
}

if (msg.msg === 'startStreaming') {
  streamingEnabled = true;
  sendBuffer();
}