Receiving all SSE events at once in Angular, not immediately when they are sent by Jersey SSE

I generate SSE events with a Java app developed with Jersey 2.32 and running on Tomcat 9.

Once the get request is received, an event is generated every 500 milliseconds, and I have checked it works fine by outputting the event data on the Java console.

The client is an Angular 12 app. I use the sse.js library as I need to add an authentification header to my request.

The client does receive the SSE messages, but it receives them all at once, instead of receiving them as soon as they are sent by the server.

Please find Angular and Java code below. Any help really appreciated.

Angular client excerpt

        // send GET request to track batch progress using server side events
        this.batchProcessMonitoringService.getMonitoringDataEvents(
            url,
            authorizationString
        ).subscribe(
            (monitoringData) => {
                this.rawMonitoringData = monitoringData;
                console.log(Date.now());
            },
            (error) => {
                console.log(error);
            },
            () => {
                // on complete, close connection
                this.batchProcessMonitoringService.closeConnection();
            },
        );

batchProcessMonitoringService

// Angular imports
import { Observable, Observer } from 'rxjs';
import { Injectable, NgZone } from '@angular/core';

// external imports
import { SSE } from 'sse.js';

// app services imports
import { ServerSideEventsService } from './server-side-events.service';

@Injectable({
    providedIn: 'root',
})
export class BatchProcessMonitoringService {

    constructor(private zone: NgZone, private sseService: ServerSideEventsService) { }

    public getMonitoringDataEvents(url: string, authorizationString: string): Observable<string> {
        return this.getServerSentEvents(url, authorizationString);
    }

    // get event source (SSE)
    private getServerSentEvents(url: string, authorizationString: string): Observable<string> {

        return new Observable((observer: Observer<any>) => {
            const eventSource = this.sseService.getEventSourceWithGet(url, authorizationString);
            eventSource.stream();
            eventSource.onmessage = (event: any) => {
                this.zone.run(() => {
                    observer.next(event.data);
                });
            };
            eventSource.onerror = (error: any) => {
                this.zone.run(() => {
                    observer.error(error.data);
                });
            };
        });
    }

    public closeConnection(): void {
        this.sseService.closeEventSource();
    }

}

ServerSideEventsService

// Angular imports
import { Injectable } from '@angular/core';

// external imports
import { SSE } from 'sse.js';

@Injectable({
    providedIn: 'root',
})
export class ServerSideEventsService {
    eventSource: SSE;

    constructor() { }

    // create an event source of POST request
    public getEventSourceWithPost(url: string, authorizationString: string, requestPayloadJson: string): SSE {
        return this.buildEventSource(url, authorizationString, 'POST', requestPayloadJson);
    }

    // create an event source of GET request
    public getEventSourceWithGet(url: string, authorizationString: string): SSE {
        return this.buildEventSource(url, authorizationString, 'GET', null);
    }

    // build the event source
    private buildEventSource(url: string, authorizationString: string, meth: string, requestPayloadJson: string): SSE {
        const options = this.buildOptions(meth, authorizationString, requestPayloadJson);
        this.eventSource = new SSE(url, options);
        // add listener
        this.eventSource.addEventListener('message', (e: any) => {
            return e.data;
        });
        return this.eventSource;
    }

    // build query options
    private buildOptions(
        meth: string,
        authorizationString: string,
        requestPayloadJson: string
    ): {
        payload: string;
        method: string;
        headers: { Authorization: string };
    } {
        const headerDict = {
            'Content-Type': 'application/json',
            Authorization: authorizationString
        };
        return {
            payload: requestPayloadJson,
            method: meth,
            headers: headerDict
        };
    }

    public closeEventSource(): void {
        if (!!this.eventSource) {
            this.eventSource.close();
        }
    }

}

Java Jersey

    // Server Side Events with User Import process monitoring data
    @GET
    @Path("/{batchId}")
    @Consumes(MediaType.APPLICATION_JSON)
    @Produces(SseFeature.SERVER_SENT_EVENTS)
    public EventOutput getServerSentEvents(@PathParam("batchId") Integer batchId) {
        final EventOutput eventOutput = new EventOutput();
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    UserImportMonitoringRec userImportMonitoring;
                    ObjectMapper objectMapper = new ObjectMapper();
                    BatchDAO batchDAO = new BatchDAO();
                    BatchDAORec batch;
                    batch = batchDAO.searchOneById(batchId);
                    userImportMonitoring = objectMapper.readValue(batch.monitoringData(),
                            UserImportMonitoringRec.class);
                    while (!userImportMonitoring.status().equals(MetaLmsConstants.UserImportCompleted)) {
                        batch = batchDAO.searchOneById(batchId);
                        userImportMonitoring = objectMapper.readValue(batch.monitoringData(),
                                UserImportMonitoringRec.class);
                        OutboundEvent event = buildUserImportOutboundEvent(batch);
                        eventOutput.write(event);
                        Thread.sleep(500);
                    }
                } catch (Exception e) {
                    throw new RuntimeException("Error when writing the event.", e);
                } finally {
                    try {
                        eventOutput.close();
                    } catch (IOException ioClose) {
                        throw new RuntimeException("Error when closing the event output.", ioClose);
                    }
                }
            }
        }).start();
        return eventOutput;

    }

    private OutboundEvent buildUserImportOutboundEvent(final BatchDAORec batch) {
        OutboundEvent.Builder eventBuilder = new OutboundEvent.Builder();
        eventBuilder.name("message");
        eventBuilder.data(String.class, batch.monitoringData());
        eventBuilder.mediaType(MediaType.APPLICATION_JSON_TYPE);
        return eventBuilder.build();

    }

updated version of batch monitoring service

// Angular imports
import { Observable, Observer } from 'rxjs';
import { Injectable, NgZone } from '@angular/core';

// external imports
import { SSE } from 'sse.js';

// app services imports
import { ServerSideEventsService } from './server-side-events.service';

@Injectable({
    providedIn: 'root',
})
export class BatchProcessMonitoringService {

    constructor(private sseService: ServerSideEventsService) { }

    public getMonitoringDataEvents(url: string, authorizationString: string): Observable<string> {
        const batchEventSource = this.sseService.getEventSourceWithGet(url, authorizationString);
        batchEventSource.stream();
        return new Observable<string>(obs => {
            batchEventSource.addEventListener('message', (evt) => {
                obs.next(evt.data);
            });
            return () => batchEventSource.close();
        });
    }

    public closeConnection(): void {
        this.sseService.closeEventSource();
    }

}