[mercedesme] internal websocket rework (#18984)

* version update, remove private discovery service reg Signed-off-by: Bernd Weymann <bernd.weymann@gmail.com>
2026-07-29 12:34:21 +02:00 · 2025-07-22 00:39:38 +02:00
parent 1d8f6f4e89
commit e8b5839c5d
5 changed files with 282 additions and 140 deletions
@@ -319,14 +319,15 @@ public class Constants {
    public static final String WEBSOCKET_USER_AGENT_CN = "MyStarCN/1.27.0 (com.daimler.ris.mercedesme.cn.ios; build:1758; iOS 16.3.1) Alamofire/5.4.0";
    public static final String WEBSOCKET_USER_AGENT_PA = "mycar-store-ap v1.27.0, android 8.0.0, SDK 2.84.3";

-    public static final String RIS_APPLICATION_VERSION_NA = "3.51.0";
-    public static final String RIS_APPLICATION_VERSION_CN = "1.51.0";
-    public static final String RIS_APPLICATION_VERSION_PA = "1.51.0";
-    public static final String RIS_APPLICATION_VERSION = "1.51.0";
-    public static final String RIS_SDK_VERSION = "2.132.2";
+    // version updates according to https://github.com/weymann/mbapi2020/commit/ede9ab60c1e18293b9436dab568b9195db6381b9
+    public static final String RIS_APPLICATION_VERSION_NA = "3.54.0";
+    public static final String RIS_APPLICATION_VERSION_CN = "1.54.0";
+    public static final String RIS_APPLICATION_VERSION_PA = "1.54.0";
+    public static final String RIS_APPLICATION_VERSION = "1.54.0";
+    public static final String RIS_SDK_VERSION = "3.55.0";
    public static final String RIS_SDK_VERSION_CN = "2.132.2";
-    public static final String RIS_OS_VERSION = "17.4.1";
-    public static final String RIS_OS_NAME = "ios";
+    public static final String RIS_OS_VERSION = "10";
+    public static final String RIS_OS_NAME = "android";
    public static final String X_APPLICATIONNAME = "mycar-store-ece";
    public static final String X_APPLICATIONNAME_ECE = "mycar-store-ece";
    public static final String X_APPLICATIONNAME_CN = "mycar-store-cn";
@@ -23,7 +23,6 @@ import org.openhab.binding.mercedesme.internal.handler.AccountHandler;
 import org.openhab.binding.mercedesme.internal.handler.VehicleHandler;
 import org.openhab.binding.mercedesme.internal.utils.Mapper;
 import org.openhab.binding.mercedesme.internal.utils.Utils;
-import org.openhab.core.config.discovery.DiscoveryService;
 import org.openhab.core.i18n.LocaleProvider;
 import org.openhab.core.i18n.LocationProvider;
 import org.openhab.core.i18n.TimeZoneProvider;
@@ -37,8 +36,6 @@ import org.openhab.core.thing.binding.BaseThingHandlerFactory;
 import org.openhab.core.thing.binding.ThingHandler;
 import org.openhab.core.thing.binding.ThingHandlerFactory;
 import org.osgi.framework.Bundle;
-import org.osgi.framework.ServiceRegistration;
-import org.osgi.service.component.ComponentContext;
 import org.osgi.service.component.annotations.Activate;
 import org.osgi.service.component.annotations.Component;
 import org.osgi.service.component.annotations.Reference;
@@ -62,7 +59,6 @@ public class MercedesMeHandlerFactory extends BaseThingHandlerFactory {
    private final MercedesMeDiscoveryService discoveryService;
    private final MercedesMeCommandOptionProvider mmcop;
    private final MercedesMeStateOptionProvider mmsop;
-    private @Nullable ServiceRegistration<?> discoveryServiceReg;

    public static String ohVersion = "unknown";

@@ -70,17 +66,17 @@ public class MercedesMeHandlerFactory extends BaseThingHandlerFactory {
    public MercedesMeHandlerFactory(@Reference HttpClientFactory hcf, @Reference StorageService storageService,
            final @Reference LocaleProvider lp, final @Reference LocationProvider locationP,
            final @Reference TimeZoneProvider tzp, final @Reference MercedesMeCommandOptionProvider cop,
-            final @Reference MercedesMeStateOptionProvider sop, final @Reference UnitProvider up) {
+            final @Reference MercedesMeStateOptionProvider sop, final @Reference UnitProvider up,
+            final @Reference MercedesMeDiscoveryService discoveryService) {
        this.storageService = storageService;
        localeProvider = lp;
        locationProvider = locationP;
        mmcop = cop;
        mmsop = sop;
-
+        this.discoveryService = discoveryService;
        Utils.initialize(tzp, lp);
        Mapper.initialize(up);
        httpClientFactory = hcf;
-        discoveryService = new MercedesMeDiscoveryService();
    }

    @Override
@@ -99,10 +95,6 @@ public class MercedesMeHandlerFactory extends BaseThingHandlerFactory {

        ThingTypeUID thingTypeUID = thing.getThingTypeUID();
        if (THING_TYPE_ACCOUNT.equals(thingTypeUID)) {
-            if (discoveryServiceReg == null) {
-                discoveryServiceReg = bundleContext.registerService(DiscoveryService.class.getName(), discoveryService,
-                        null);
-            }
            return new AccountHandler((Bridge) thing, discoveryService, httpClientFactory.getCommonHttpClient(),
                    localeProvider, storageService);
        } else if (THING_TYPE_BEV.equals(thingTypeUID) || THING_TYPE_COMB.equals(thingTypeUID)
@@ -112,15 +104,6 @@ public class MercedesMeHandlerFactory extends BaseThingHandlerFactory {
        return null;
    }

-    @Override
-    protected void deactivate(ComponentContext componentContext) {
-        super.deactivate(componentContext);
-        if (discoveryServiceReg != null) {
-            discoveryServiceReg.unregister();
-            discoveryServiceReg = null;
-        }
-    }
-
    public static String getVersion() {
        return ohVersion;
    }
@@ -32,7 +32,8 @@ import org.osgi.service.component.annotations.Component;
 * @author Bernd Weymann - Add vin as representation property
 */
@NonNullByDefault
-@Component(service = DiscoveryService.class, configurationPid = "discovery.mercedesme")
+@Component(service = { DiscoveryService.class,
+        MercedesMeDiscoveryService.class }, configurationPid = "discovery.mercedesme")
 public class MercedesMeDiscoveryService extends AbstractDiscoveryService {

    public MercedesMeDiscoveryService() {
@@ -119,6 +119,7 @@ public class AccountHandler extends BaseBridgeHandler implements AccessTokenRefr
        if (!configValidReason.isEmpty()) {
            updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.CONFIGURATION_ERROR, configValidReason);
        } else {
+            mbWebsocket.dispose(false);
            authService = Optional.of(new AuthService(this, httpClient, config, localeProvider.getLocale(), storage,
                    config.refreshToken));
            refreshScheduler = Optional
@@ -128,7 +129,7 @@ public class AccountHandler extends BaseBridgeHandler implements AccessTokenRefr

    public void refresh() {
        if (!Constants.NOT_SET.equals(authService.get().getToken())) {
-            mbWebsocket.run();
+            mbWebsocket.update();
        } else {
            // all failed - start manual authorization
            String textKey = Constants.STATUS_TEXT_PREFIX + thing.getThingTypeUID().getId()
@@ -160,8 +161,8 @@ public class AccountHandler extends BaseBridgeHandler implements AccessTokenRefr
                schedule.cancel(true);
            }
        });
-        mbWebsocket.dispose();
        eventQueue.clear();
+        mbWebsocket.dispose(true);
    }

    @Override
@@ -459,7 +460,7 @@ public class AccountHandler extends BaseBridgeHandler implements AccessTokenRefr

    public void sendCommand(@Nullable ClientMessage cm) {
        if (cm != null) {
-            mbWebsocket.setCommand(cm);
+            mbWebsocket.addCommand(cm);
        }
        scheduler.schedule(this::refresh, 2, TimeUnit.SECONDS);
    }
@@ -18,7 +18,14 @@ import java.net.URI;
 import java.nio.ByteBuffer;
 import java.time.Instant;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;

 import org.eclipse.jdt.annotation.NonNullByDefault;
 import org.eclipse.jdt.annotation.Nullable;
@@ -27,12 +34,15 @@ import org.eclipse.jetty.websocket.api.Session;
 import org.eclipse.jetty.websocket.api.annotations.OnWebSocketClose;
 import org.eclipse.jetty.websocket.api.annotations.OnWebSocketConnect;
 import org.eclipse.jetty.websocket.api.annotations.OnWebSocketError;
+import org.eclipse.jetty.websocket.api.annotations.OnWebSocketFrame;
 import org.eclipse.jetty.websocket.api.annotations.OnWebSocketMessage;
 import org.eclipse.jetty.websocket.api.annotations.WebSocket;
+import org.eclipse.jetty.websocket.api.extensions.Frame;
 import org.eclipse.jetty.websocket.client.ClientUpgradeRequest;
 import org.eclipse.jetty.websocket.client.WebSocketClient;
 import org.openhab.binding.mercedesme.internal.Constants;
 import org.openhab.binding.mercedesme.internal.handler.AccountHandler;
+import org.openhab.core.common.ThreadPoolManager;
 import org.openhab.core.thing.ThingStatus;
 import org.openhab.core.thing.ThingStatusDetail;
 import org.slf4j.Logger;
@@ -50,92 +60,121 @@ import com.daimler.mbcarkit.proto.VehicleEvents.PushMessage;
@WebSocket
@NonNullByDefault
 public class MBWebsocket {
-    // timeout 14 Minutes - just below scheduling of 15 Minutes by AccountHandler
-    private static final int CONNECT_TIMEOUT_MS = 14 * 60 * 1000;
+    // timeout stays unlimited until binding decides to close
+    private static final int CONNECT_TIMEOUT_MS = 0;
    // standard runtime of Websocket
    private static final int WS_RUNTIME_MS = 60 * 1000;
    // addon time of 1 minute for a new send command
    private static final int ADDON_MESSAGE_TIME_MS = 60 * 1000;
    // check Socket time elapsed each second
-    private static final int CHECK_INTERVAL_MS = 1000;
+    private static final int CHECK_INTERVAL_MS = 60 * 1000;
    // additional 5 minutes after keep alive
    private static final int KEEP_ALIVE_ADDON = 5 * 60 * 1000;

    private final Logger logger = LoggerFactory.getLogger(MBWebsocket.class);
-    private AccountHandler accountHandler;
-    private HttpClient httpClient;
-    private boolean running = false;
-    private Instant runTill = Instant.now();
-    private @Nullable Session session;
+    private final Map<String, Instant> pingPongMap = new HashMap<>();
+    private final AccountHandler accountHandler;
+    private final HttpClient httpClient;
+    private final ScheduledExecutorService scheduler = ThreadPoolManager
+            .getPoolBasedSequentialScheduledExecutorService("mercedesme-websocket", null);
+
+    private Optional<ScheduledFuture<?>> refresher = Optional.empty();
+    private Optional<WebSocketClient> webSocketClient = Optional.empty();
+    private Optional<Session> session = Optional.empty();
    private List<ClientMessage> commandQueue = new ArrayList<>();
-
+    private Instant runTill = Instant.now();
+    private WebsocketState state = WebsocketState.STOPPED;
    private boolean keepAlive = false;
+    private boolean disposed = true;

-    public MBWebsocket(AccountHandler ah, HttpClient hc) {
-        accountHandler = ah;
-        httpClient = hc;
+    public enum WebsocketState {
+        STOPPED,
+        DISCONNECTED,
+        CONNECTED,
+        STARTED
+    }
+
+    public MBWebsocket(AccountHandler accountHandler, HttpClient httpClient) {
+        this.accountHandler = accountHandler;
+        this.httpClient = httpClient;
    }

    /**
-     * Is called by
-     * - scheduler every 15 minutes
-     * - handler sending a command
-     * - handler requesting refresh
+     * Regular update call from AccountHandler to refresh data according to refreshInterval
     */
-    public void run() {
-        synchronized (this) {
-            if (running) {
-                return;
-            } else {
-                running = true;
-                runTill = Instant.now().plusMillis(WS_RUNTIME_MS);
-            }
-        }
-        try {
-            WebSocketClient client = new WebSocketClient(httpClient);
-            client.setMaxIdleTimeout(CONNECT_TIMEOUT_MS);
-            client.setStopTimeout(CONNECT_TIMEOUT_MS);
-            ClientUpgradeRequest request = accountHandler.getClientUpgradeRequest();
-            String websocketURL = accountHandler.getWSUri();
-            if (Constants.JUNIT_TOKEN.equals(request.getHeader("Authorization"))) {
-                // avoid unit test requesting real web socket - simply return
-                return;
-            }
-            logger.trace("Websocket start {} max message size {}", websocketURL, client.getMaxBinaryMessageSize());
-            client.start();
-            client.connect(this, new URI(websocketURL), request);
-            while (keepAlive || Instant.now().isBefore(runTill)) {
-                try {
-                    Thread.sleep(CHECK_INTERVAL_MS);
-                } catch (InterruptedException ie) {
-                    logger.trace("Websocket interrupted during sleeping - stop executing");
-                    runTill = Instant.MIN;
-                }
-                // sends one message per second
-                if (sendMessage()) {
-                    // add additional runtime to execute and finish command
-                    runTill = runTill.plusMillis(ADDON_MESSAGE_TIME_MS);
-                }
-            }
-            logger.trace("Websocket stop");
-            client.stop();
-            client.destroy();
-        } catch (Throwable t) {
-            // catch Exceptions of start stop and declare communication error
-            accountHandler.updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.COMMUNICATION_ERROR,
-                    "@text/mercedesme.account.status.websocket-failure");
-            logger.warn("Websocket handling exception: {}", t.getMessage());
-        } finally {
-            synchronized (this) {
-                running = false;
+    public void update() {
+        scheduler.execute(this::doRefresh);
+    }
+
+    /**
+     * If proto update is received an acknowledge message needs to be sent to the server to avoid repeating this message
+     * again and again.
+     */
+    public void sendAcknowledgeMessage(ClientMessage message) {
+        session.ifPresent(s -> {
+            try {
+                ByteArrayOutputStream baos = new ByteArrayOutputStream();
+                message.writeTo(baos);
+                s.getRemote().sendBytes(ByteBuffer.wrap(baos.toByteArray()));
+            } catch (IOException e) {
+                logger.warn("Error sending acknowledge {} : {}", message.getAllFields(), e.getMessage());
            }
+        });
+    }
+
+    /**
+     * Add command to the command queue to be sent to the server. Immediately request refresh to send command as soon as
+     * possible.
+     *
+     * @param command to be sent
+     */
+    public void addCommand(ClientMessage command) {
+        commandQueue.add(command);
+        scheduler.execute(this::doRefresh);
+    }
+
+    /**
+     * Dispose websocket in case of disposed AccountHandler. Cleanup stored files and stop web socket client.
+     */
+    public void dispose(boolean disposed) {
+        this.disposed = disposed;
+        if (disposed) {
+            runTill = Instant.MIN;
+            keepAlive = false;
+            refresher.ifPresent(job -> {
+                job.cancel(false);
+                refresher = Optional.empty();
+            });
+            scheduler.execute(this::stop);
        }
    }

-    public void setCommand(ClientMessage cm) {
-        commandQueue.add(cm);
+    /**
+     * Set keep alive mode for web socket connection. If keep alive is set to true the web socket will not be closed
+     *
+     * @param alive
+     */
+    public void keepAlive(boolean alive) {
+        if (!keepAlive) {
+            if (alive) {
+                logger.trace("WebSocket - keep alive start");
+            }
+        } else {
+            if (!alive) {
+                // after keep alive is finished add 5 minutes to cover e.g. door events after
+                // trip is finished
+                runTill = Instant.now().plusMillis(KEEP_ALIVE_ADDON);
+                logger.trace("Websocket - keep alive stop - run till {}", runTill.toString());
+            }
+        }
+        keepAlive = alive;
    }

+    /**
+     * Send message to the server if there is a command available in the command queue.
+     *
+     * @return true if command is successfully submitted, false otherwise
+     */
    private boolean sendMessage() {
        if (!commandQueue.isEmpty()) {
            ClientMessage message = commandQueue.remove(0);
@@ -143,10 +182,13 @@ public class MBWebsocket {
            try {
                ByteArrayOutputStream baos = new ByteArrayOutputStream();
                message.writeTo(baos);
-                if (session != null) {
-                    session.getRemote().sendBytes(ByteBuffer.wrap(baos.toByteArray()));
+                if (session.isPresent()) {
+                    session.get().getRemote().sendBytes(ByteBuffer.wrap(baos.toByteArray()));
+                    return true;
+                } else {
+                    logger.warn("Cannot send message {} - no session available", message.getAllFields());
+                    return false;
                }
-                return true;
            } catch (IOException e) {
                logger.warn("Error sending message {} : {}", message.getAllFields(), e.getMessage());
            }
@@ -155,50 +197,124 @@ public class MBWebsocket {
        return false;
    }

-    public void sendAcknowledgeMessage(ClientMessage message) {
-        try {
-            ByteArrayOutputStream baos = new ByteArrayOutputStream();
-            message.writeTo(baos);
-            if (session != null) {
-                session.getRemote().sendBytes(ByteBuffer.wrap(baos.toByteArray()));
+    /**
+     * Start the web socket client and connect to the server. If the web socket client is already running it will not
+     */
+    private void start() {
+        if (!disposed && webSocketClient.isEmpty()) {
+            WebSocketClient client = new WebSocketClient(httpClient);
+            try {
+                client.setMaxIdleTimeout(CONNECT_TIMEOUT_MS);
+                ClientUpgradeRequest request = accountHandler.getClientUpgradeRequest();
+                String websocketURL = accountHandler.getWSUri();
+                if (Constants.JUNIT_TOKEN.equals(request.getHeader("Authorization"))) {
+                    // avoid unit test requesting real web socket - simply return
+                    return;
+                }
+                logger.trace("Websocket start {} max message size {}", websocketURL, client.getMaxBinaryMessageSize());
+                client.start();
+                client.connect(this, new URI(websocketURL), request);
+                webSocketClient = Optional.of(client);
+                runTill = Instant.now().plusMillis(WS_RUNTIME_MS);
+                state = WebsocketState.STARTED;
+            } catch (Exception e) {
+                // catch Exceptions of start stop and declare communication error
+                accountHandler.updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.COMMUNICATION_ERROR,
+                        "@text/mercedesme.account.status.websocket-failure");
+                logger.warn("Websocket handling exception: {}", e.getMessage());
            }
-        } catch (IOException e) {
-            logger.warn("Error sending acknowledge {} : {}", message.getAllFields(), e.getMessage());
-        }
-    }
-
-    public void interrupt() {
-        synchronized (this) {
-            runTill = Instant.MIN;
-            keepAlive = false;
        }
    }

    /**
-     * If disposed temp debug files are deleted
+     * Performs an update of the web socket connection. If websocket is disposed refresh will not be executed. In case
+     * of CONNECTED it will check
+     * - if there are commands to be sent
+     * - send ping to the server
+     * - check if keep alive is set or run time is not over
+     * In case of other state it will start the web socket connection.
     */
-    public void dispose() {
-        interrupt();
-    }
-
-    public void keepAlive(boolean b) {
-        if (!keepAlive) {
-            if (b) {
-                logger.trace("WebSocket - keep alive start");
+    private void doRefresh() {
+        if (!disposed) {
+            if (state == WebsocketState.CONNECTED) {
+                logger.trace("Refresh: Websocket fine - state {}", state);
+                if (sendMessage()) {
+                    // add additional runtime to execute and finish command
+                    runTill = runTill.plusMillis(ADDON_MESSAGE_TIME_MS);
+                }
+                ping();
+                if (keepAlive || Instant.now().isBefore(runTill)) {
+                    // doRefresh is called by AccountHandler, websocket endpoint onConnect and addCommand. To avoid
+                    // multiple future calls cancel the current running or future schedule calls.
+                    refresher.ifPresent(job -> {
+                        job.cancel(false);
+                    });
+                    refresher = Optional
+                            .of(scheduler.schedule(this::doRefresh, CHECK_INTERVAL_MS, TimeUnit.MILLISECONDS));
+                } else {
+                    // run time is over - disconnect
+                    logger.debug("Websocket run time is over - disconnect");
+                    scheduler.execute(this::stop);
+                }
+            } else {
+                logger.trace("Refresh: Websocket needs to be started - state {}", state);
+                scheduler.execute(this::start);
            }
        } else {
-            if (!b) {
-                // after keep alive is finished add 5 minutes to cover e.g. door events after
-                // trip is finished
-                runTill = Instant.now().plusMillis(KEEP_ALIVE_ADDON);
-                logger.trace("Websocket - keep alive stop - run till {}", runTill.toString());
-            }
+            logger.trace("Refresh: Websocket disposed - state {}", state);
        }
-        keepAlive = b;
    }

    /**
-     * endpoints
+     * Request to disconnect the web socket session. This will close the session normally with a status code of 1000
+     */
+    private void disconnect() {
+        session.ifPresent(session -> {
+            // close session normally
+            session.close(1000, "Websocket closed by binding");
+        });
+    }
+
+    /**
+     * Stop the web socket client and disconnect the session if it is still connected.
+     */
+    private void stop() {
+        session.ifPresentOrElse(session -> {
+            logger.trace("Websocket stop - disconnect session first - state {}", state);
+            scheduler.execute(this::disconnect);
+        }, () -> {
+            logger.trace("Websocket stop - state {}", state);
+            webSocketClient.ifPresent(client -> {
+                try {
+                    client.stop();
+                } catch (Exception e) {
+                    logger.warn("Websocket stop exception: {}", e.getMessage());
+                }
+                client.destroy();
+                webSocketClient = Optional.empty();
+            });
+            state = WebsocketState.STOPPED;
+        });
+    }
+
+    /**
+     * Ping the server to keep the connection alive and to check if the connection is still valid.
+     */
+    private void ping() {
+        logger.trace("Websocket ping {}", Instant.now().toString());
+        session.ifPresent(session -> {
+            try {
+                String pingId = UUID.randomUUID().toString();
+                pingPongMap.put(pingId, Instant.now());
+                session.getRemote().sendPing(ByteBuffer.wrap(pingId.getBytes()));
+            } catch (IOException e) {
+                logger.warn("Websocket ping failed {}", e.getMessage());
+            }
+        });
+    }
+
+    /**
+     * web socket endpoints
     */

    @OnWebSocketMessage
@@ -232,24 +348,64 @@ public class MBWebsocket {
        }
    }

-    @OnWebSocketClose
-    public void onDisconnect(Session session, int statusCode, String reason) {
-        logger.debug("Disconnected from server. Status {} Reason {}", statusCode, reason);
-        this.session = null;
-        // ensure execution stop if disconnect was triggered from server side
-        interrupt();
+    @OnWebSocketFrame
+    public void onFrame(Frame frame) {
+        if (Frame.Type.PONG.equals(frame.getType())) {
+            ByteBuffer buffer = frame.getPayload();
+            byte[] bytes = new byte[frame.getPayloadLength()];
+            for (int i = 0; i < frame.getPayloadLength(); i++) {
+                bytes[i] = buffer.get(i);
+            }
+            String payloadString = new String(bytes);
+            Instant sent = pingPongMap.remove(payloadString);
+            if (sent == null) {
+                logger.debug("Websocket received pong without ping {}", payloadString);
+            }
+        } else if (Frame.Type.PING.equals(frame.getType())) {
+            session.ifPresentOrElse((session) -> {
+                ByteBuffer buffer = frame.getPayload();
+                try {
+                    session.getRemote().sendPong(buffer);
+                } catch (IOException e) {
+                    logger.warn("Websocket onPing answer exception {}", e.getMessage());
+                }
+            }, () -> {
+                logger.debug("Websocket onPing answer cannot be initiated");
+            });
+        }
    }

    @OnWebSocketConnect
    public void onConnect(Session session) {
+        this.session = Optional.of(session);
+        state = WebsocketState.CONNECTED;
+        pingPongMap.clear();
        accountHandler.updateStatus(ThingStatus.ONLINE);
-        this.session = session;
+        logger.trace("Websocket connected - state {}", state);
+        // websocket client is started and connected - time to refresh
+        scheduler.execute(this::doRefresh);
+    }
+
+    @OnWebSocketClose
+    public void onDisconnect(Session session, int statusCode, String reason) {
+        onClosedSession(null);
+        logger.trace("Disconnected from server. Status {} Reason {}", statusCode, reason);
    }

    @OnWebSocketError
    public void onError(Throwable t) {
-        logger.debug("Error during web socket connection - {}", t.getMessage());
-        accountHandler.updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.COMMUNICATION_ERROR,
-                "@text/mercedesme.account.status.websocket-failure [\"" + t.getMessage() + "\"]");
+        onClosedSession(t);
+    }
+
+    private void onClosedSession(@Nullable Throwable throwable) {
+        this.session = Optional.empty();
+        state = WebsocketState.DISCONNECTED;
+        pingPongMap.clear();
+        if (throwable != null) {
+            accountHandler.updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.COMMUNICATION_ERROR,
+                    "@text/mercedesme.account.status.websocket-failure [\"" + throwable.getMessage() + "\"]");
+        }
+        // stop web socket client for closed session
+        scheduler.execute(this::stop);
    }
 }