Why is fetch event source closing shortly after creation?

I am trying to fetch instant and scheduled notifications from spring-boot server to angular client using the microsoft fetch event source package, but the event source only closes shortly after firing. I will appreciate any solution to this. Angular service class

constructor(private http:HttpClient,
private authService:AuthService, private zone:NgZone) {


const currentUser = toSignal(this.authService.loggedInUserObs$);

effect((onCleanup) => {

const user = currentUser?.();

if(user && this.isAdminUser(user)){

this.connectToNotifications(user)
} else {

this.disconnectFromSSE();
}


onCleanup(() => {
this.disconnectFromSSE();
}
)
}, {allowSignalWrites:true});



}

// disconnect from SSE notifications
disconnectFromSSE() {

this.abortController?.abort();
this.abortController = null;
this.connectionState.set('disconnected');
}
private async connectToNotifications(user: User) {


//this.disconnectFromSSE();
this.connectionState.set('connecting');

this.abortController = new AbortController();

const token = user.accessToken;



try{

await fetchEventSource(`${this.connectionUrl}?_xxid=${user.id}`, {
method: 'GET',
// credentials: 'include',
headers: {
'Authorization': `Bearer ${token}`,
'Accept': 'text/event-stream',
'Cache-Control': 'no-cache',
'Connection': 'keep-alive',


},


signal: this.abortController.signal,
onopen: async (response) => {

this.zone.run(() => {

if(response.ok && response.status === 200){

console.log('200 ok response')

this.connectionState.set('connected');

this.retryCount = 0;

return;


}else if(response.status >= 400){
console.log('4xx error response')
this.connectionState.set('error');
}


throw new Error(`Connection failed: ${response.status} ${response.statusText}`);
});

},

onmessageevent:EventSourceMessage) => {

this.zone.run(() => {

try{

if(event.event==='responseUpdate'){


const data:AssessmentResponseRecord = JSON.parse(event.data);

console.log(JSON.stringify(data, null,1))

const index = this.notifications().findIndex((n) => n.topic === data.topic && n.instructorId === data.instructorId && n.postedOn === data.postedOn);
if(index === -1){

this.notifications.update(prev => [...prev, data]);


}
}

} catch (error) {
console.error('Error processing event message:', error);
}
});
},

onerrorerr) => {

this.zone.run(() => {

console.error('SSE error', err);
this.connectionState.set('error');

if(err.name === 'AbortError') return;

this.attemptReconnection(user);

});
},

onclose: () => {
this.zone.run(() => {
console.log('SSE connection closed');
this.connectionState.set('disconnected');



});
}
})
}catch(err){

this.zone.run(() => {
this.connectionState.set('error');
console.error('SSE connection failed:', err);

// this.attemptReconnection(user);
})
}

}

private attemptReconnection(user: User) {
if (this.retryCount < this.maxRetries) {

const delay = Math.min(this.baseDelay * 2 ** this.retryCount, 30000);

this.retryCount++;

setTimeout(() => this.connectToNotifications(user), delay);
}
}



Spring-boot connection service

@Service
public class ConnectorService{
private Map<Integer, SseEmitter> connectors = new ConcurrentHashMap<>();


@Override
public SseEmitter establishConnection(Integer instructorId) {

// get previous connector if there was
SseEmitter emitter = connectors.get(instructorId);

// create new connector if there was non previously
if(emitter == null) {

emitter = new SseEmitter(0L);

connectors.put(instructorId, emitter);

System.out.println("new connection made");
}else {

System.out.println("previously connected");
}
heartbeat(emitter);

emitter.onCompletion(() -> connectors.remove(instructorId));

emitter.onTimeout(() -> connectors.remove(instructorId));





return emitter;
}
}


Scheduled notification service:

@Service
public class ScheduledNotificationService{
@Scheduled(fixedRate = 100000) //schedule notifications at 20sec interval
public void scheduleFixedTimeNotification() {



RLock rlock = redissonClient.getLock("notification:lock");
try {

if(rlock.tryLock(5, 15, TimeUnit.SECONDS)) {



try {
Set<String> keys = cachingKeyUtils.getAllCacheKeys(RedisValues.ASSESSMENT_RESPONSE_NOTIFICATION);

Cache cache = cacheManager.getCache(RedisValues.ASSESSMENT_RESPONSE_NOTIFICATION);



for(String key: keys) {


try {


Cache.ValueWrapper wrapper = cache.get(key);

if(wrapper != null) {

//
List<AssessmentResponseRecord> notifications = mapper.convertValue(wrapper.get(),

new TypeReference<List<AssessmentResponseRecord>>() {}
);

broadcaster.broadcastPreviousNotifications(notifications);




}



} catch (Exception e) {
System.err.println(e);
}
}
} finally {
rlock.unlock();
}
}
} catch (Exception e) {
Thread.currentThread().interrupt();
System.err.println(e);
}


}




After setting up server event sse emitter and client subscriber, I expected the connection to be long-lived, but the client fetch-event-source closes immediately.

Continue reading...