CB-6305 websocket fix (#3293)

server/bundles/io.cloudbeaver.server/src/io/cloudbeaver/server/websockets/CBWebSocketServerConfigurator.java

@@ -1,6 +1,6 @@
 /*
  * DBeaver - Universal Database Manager
- * Copyright (C) 2010-2024 DBeaver Corp and others
+ * Copyright (C) 2010-2025 DBeaver Corp and others
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -74,17 +74,20 @@ public class CBWebSocketServerConfigurator extends ServerEndpointConfig.Configur
         // possible desktop client session
         try {
             var headlessSession = createHeadlessSession(requestInfo, request);
-            if (headlessSession == null) {
-                log.debug("Couldn't create headless session");
-                return;
+            if (headlessSession != null) {
+                sec.getUserProperties().put(PROP_WEB_SESSION, headlessSession);
+            } else  {
+                log.trace("Couldn't create headless session");
             }
-            sec.getUserProperties().put(PROP_WEB_SESSION, headlessSession);
         } catch (SMAccessTokenExpiredException e) {
             sec.getUserProperties().put(PROP_TOKEN_EXPIRED, true);
         } catch (DBException e) {
             log.error("Error resolve websocket session", e);
             throw new RuntimeException(e.getMessage(), e);
         }
+        if (sec.getUserProperties().get(PROP_WEB_SESSION) == null) {
+            throw new RuntimeException("No web session found for websocket request");
+        }
     }
 
 

webapp/packages/core-root/src/SessionEventSource.ts

@@ -7,9 +7,12 @@
  */
 import {
   catchError,
+  concatMap,
   debounceTime,
+  defer,
   delayWhen,
   filter,
+  from,
   interval,
   map,
   merge,
@@ -19,13 +22,15 @@ import {
   repeat,
   retry,
   share,
+  shareReplay,
   Subject,
+  switchMap,
   throwError,
 } from 'rxjs';
 import { webSocket, WebSocketSubject } from 'rxjs/webSocket';
 
 import { injectable } from '@cloudbeaver/core-di';
-import { type ISyncExecutor, SyncExecutor } from '@cloudbeaver/core-executor';
+import { Executor, type IExecutor, type ISyncExecutor, SyncExecutor } from '@cloudbeaver/core-executor';
 import {
   CbClientEventId as ClientEventId,
   EnvironmentService,
@@ -58,6 +63,7 @@ const RETRY_INTERVAL = 30 * 1000;
 @injectable()
 export class SessionEventSource implements IServerEventEmitter<ISessionEvent, ISessionEvent, SessionEventId, SessionEventTopic> {
   readonly eventsSubject: Observable<ISessionEvent>;
+  readonly onActivate: IExecutor;
   readonly onInit: ISyncExecutor;
 
   private readonly closeSubject: Subject<CloseEvent>;
@@ -75,6 +81,7 @@ export class SessionEventSource implements IServerEventEmitter<ISessionEvent, IS
     private readonly sessionExpireService: SessionExpireService,
     environmentService: EnvironmentService,
   ) {
+    this.onActivate = new Executor();
     this.onInit = new SyncExecutor();
     this.oldEventsSubject = new Subject();
     this.disconnectSubject = new Subject();
@@ -91,21 +98,28 @@ export class SessionEventSource implements IServerEventEmitter<ISessionEvent, IS
       openObserver: this.openSubject,
     });
 
+    const ready$ = defer(() => from(this.onActivate.execute())).pipe(shareReplay(1));
+
     this.emitSubject = new Subject();
-    this.emitSubject.pipe(this.handleDisconnected()).subscribe(this.subject);
+    this.emitSubject
+      .pipe(
+        this.handleDisconnected(),
+        concatMap(value => ready$.pipe(concatMap(() => from([value])))),
+      )
+      .subscribe(this.subject);
 
     this.openSubject.subscribe(() => {
       this.onInit.execute();
     });
 
     this.closeSubject.subscribe(event => {
-      console.info(`Websocket closed: ${event.reason}`);
+      console.warn(`Websocket closed (${event.code}): ${event.reason}`);
     });
 
-    this.eventsSubject = merge(this.oldEventsSubject, this.subject).pipe(this.handleErrors());
+    this.eventsSubject = merge(this.oldEventsSubject, ready$.pipe(switchMap(() => this.subject))).pipe(this.handleErrors());
 
     this.errorSubject.pipe(debounceTime(1000)).subscribe(error => {
-      console.error(error);
+      console.error('Websocket:', error);
     });
 
     this.errorHandler = this.errorHandler.bind(this);

webapp/packages/core-root/src/SessionResource.ts

@@ -11,7 +11,7 @@ import { CachedDataResource } from '@cloudbeaver/core-resource';
 import { GraphQLService, type SessionStateFragment } from '@cloudbeaver/core-sdk';
 
 import { ServerConfigResource } from './ServerConfigResource.js';
-import { ServerEventId } from './SessionEventSource.js';
+import { ServerEventId, SessionEventSource } from './SessionEventSource.js';
 import { type ISessionStateEvent, SessionInfoEventHandler } from './SessionInfoEventHandler.js';
 
 export type SessionState = SessionStateFragment;
@@ -26,6 +26,7 @@ export class SessionResource extends CachedDataResource<SessionState | null> {
 
   constructor(
     private readonly graphQLService: GraphQLService,
+    sessionEventSource: SessionEventSource,
     private readonly sessionInfoEventHandler: SessionInfoEventHandler,
     serverConfigResource: ServerConfigResource,
     private readonly localizationService: LocalizationService,
@@ -34,6 +35,7 @@ export class SessionResource extends CachedDataResource<SessionState | null> {
 
     this.handleSessionStateEvent = this.handleSessionStateEvent.bind(this);
 
+    sessionEventSource.onActivate.addHandler(() => this.load());
     sessionInfoEventHandler.onEvent(ServerEventId.CbSessionState, this.handleSessionStateEvent, undefined, this);
 
     this.action = null;