Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[mercedesme] Websocket decoupling #17753

Open
wants to merge 9 commits into
base: main
Choose a base branch
from
+234 −89
Open

[mercedesme] Websocket decoupling #17753

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
cb236c4
decouple websocket thread from handler update
weymann Nov 12, 2024
f5d4f9d
decouple websocket thread
weymann Nov 13, 2024
a11db88
socket thread decoupling from updating things
weymann Nov 15, 2024
30e0379
close stream after reading
weymann Nov 15, 2024
f26c510
free websocket as early as possible
weymann Nov 16, 2024
dd13cd7
harmonize function names
weymann Nov 16, 2024
5b8997e
clear eventQueue in case of dispose and thread interrupt
weymann Nov 17, 2024
540925b
just a comment
weymann Nov 17, 2024
8dca914
correct review changes
weymann Nov 23, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
114 changes: 105 additions & 9 deletions ...cedesme/src/main/java/org/openhab/binding/mercedesme/internal/handler/AccountHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
*/
package org.openhab.binding.mercedesme.internal.handler;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
Expand Down Expand Up @@ -57,8 +58,15 @@
import org.slf4j.LoggerFactory;

import com.daimler.mbcarkit.proto.Client.ClientMessage;
import com.daimler.mbcarkit.proto.Protos.AcknowledgeAssignedVehicles;
import com.daimler.mbcarkit.proto.VehicleEvents;
import com.daimler.mbcarkit.proto.VehicleEvents.AcknowledgeVEPUpdatesByVIN;
import com.daimler.mbcarkit.proto.VehicleEvents.PushMessage;
import com.daimler.mbcarkit.proto.VehicleEvents.VEPUpdate;
import com.daimler.mbcarkit.proto.Vehicleapi.AcknowledgeAppTwinCommandStatusUpdatesByVIN;
import com.daimler.mbcarkit.proto.Vehicleapi.AppTwinCommandStatusUpdatesByPID;
import com.daimler.mbcarkit.proto.Vehicleapi.AppTwinCommandStatusUpdatesByVIN;
import com.daimler.mbcarkit.proto.Vehicleapi.AppTwinPendingCommandsRequest;

/**
* The {@link AccountHandler} acts as Bridge between MercedesMe Account and the associated vehicles
Expand All @@ -82,7 +90,9 @@ public class AccountHandler extends BaseBridgeHandler implements AccessTokenRefr

private Optional<AuthServer> server = Optional.empty();
private Optional<AuthService> authService = Optional.empty();
private Optional<ScheduledFuture<?>> scheduledFuture = Optional.empty();
private Optional<ScheduledFuture<?>> refreshScheduler = Optional.empty();
private List<byte[]> eventQueue = new ArrayList<>();
private boolean updateRunning = false;

private String capabilitiesEndpoint = "/v1/vehicle/%s/capabilities";
private String commandCapabilitiesEndpoint = "/v1/vehicle/%s/capabilities/commands";
Expand Down Expand Up @@ -128,13 +138,13 @@ public void initialize() {
updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.NONE,
textKey + " [\"" + thing.getProperties().get("callbackUrl") + "\"]");
} else {
scheduledFuture = Optional.of(scheduler.scheduleWithFixedDelay(this::update, 0,
refreshScheduler = Optional.of(scheduler.scheduleWithFixedDelay(this::refresh, 0,
config.get().refreshInterval, TimeUnit.MINUTES));
}
}
}

public void update() {
public void refresh() {
if (server.isPresent()) {
if (!Constants.NOT_SET.equals(authService.get().getToken())) {
ws.run();
Expand Down Expand Up @@ -203,12 +213,13 @@ public void dispose() {
server = Optional.empty();
Utils.removePort(config.get().callbackPort);
}
ws.interrupt();
scheduledFuture.ifPresent(schedule -> {
refreshScheduler.ifPresent(schedule -> {
if (!schedule.isCancelled()) {
schedule.cancel(true);
}
});
ws.interrupt();
eventQueue.clear();
}

/**
Expand All @@ -217,7 +228,7 @@ public void dispose() {
@Override
public void onAccessTokenResponse(AccessTokenResponse tokenResponse) {
if (!Constants.NOT_SET.equals(tokenResponse.getAccessToken())) {
scheduler.schedule(this::update, 2, TimeUnit.SECONDS);
scheduler.schedule(this::refresh, 2, TimeUnit.SECONDS);
} else if (server.isEmpty()) {
// server not running - fix first
String textKey = Constants.STATUS_TEXT_PREFIX + thing.getThingTypeUID().getId()
Expand Down Expand Up @@ -262,7 +273,7 @@ public void registerVin(String vin, VehicleHandler handler) {
activeVehicleHandlerMap.put(vin, handler);
VEPUpdate updateForVin = vepUpdateMap.get(vin);
if (updateForVin != null) {
handler.distributeContent(updateForVin);
handler.enqueueUpdate(updateForVin);
}
}

Expand All @@ -284,12 +295,97 @@ public void getVehicleCapabilities(String vin) {
}
}

/**
* functions for websocket handling
*/

public void enqueueMessage(byte[] data) {
synchronized (eventQueue) {
eventQueue.add(data);
scheduler.execute(this::scheduleMessage);
}
}

private void scheduleMessage() {
byte[] data;
synchronized (eventQueue) {
while (updateRunning) {
try {
eventQueue.wait();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
eventQueue.clear();
return;
}
}
if (!eventQueue.isEmpty()) {
data = eventQueue.remove(0);
} else {
return;
}
updateRunning = true;
}
try {
handleMessage(data);
} finally {
synchronized (eventQueue) {
updateRunning = false;
eventQueue.notifyAll();
}
}
}

private void handleMessage(byte[] array) {
try {
PushMessage pm = VehicleEvents.PushMessage.parseFrom(array);
if (pm.hasVepUpdates()) {
boolean distributed = distributeVepUpdates(pm.getVepUpdates().getUpdatesMap());
logger.trace("Distributed VEPUpdate {}", distributed);
if (distributed) {
AcknowledgeVEPUpdatesByVIN ack = AcknowledgeVEPUpdatesByVIN.newBuilder()
.setSequenceNumber(pm.getVepUpdates().getSequenceNumber()).build();
ClientMessage cm = ClientMessage.newBuilder().setAcknowledgeVepUpdatesByVin(ack).build();
ws.sendAcknowledgeMessage(cm);
}
} else if (pm.hasAssignedVehicles()) {
for (int i = 0; i < pm.getAssignedVehicles().getVinsCount(); i++) {
String vin = pm.getAssignedVehicles().getVins(0);
discovery(vin);
}
AcknowledgeAssignedVehicles ack = AcknowledgeAssignedVehicles.newBuilder().build();
ClientMessage cm = ClientMessage.newBuilder().setAcknowledgeAssignedVehicles(ack).build();
ws.sendAcknowledgeMessage(cm);
} else if (pm.hasApptwinCommandStatusUpdatesByVin()) {
AppTwinCommandStatusUpdatesByVIN csubv = pm.getApptwinCommandStatusUpdatesByVin();
commandStatusUpdate(csubv.getUpdatesByVinMap());
AcknowledgeAppTwinCommandStatusUpdatesByVIN ack = AcknowledgeAppTwinCommandStatusUpdatesByVIN
.newBuilder().setSequenceNumber(csubv.getSequenceNumber()).build();
ClientMessage cm = ClientMessage.newBuilder().setAcknowledgeApptwinCommandStatusUpdateByVin(ack)
.build();
ws.sendAcknowledgeMessage(cm);
} else if (pm.hasApptwinPendingCommandRequest()) {
AppTwinPendingCommandsRequest pending = pm.getApptwinPendingCommandRequest();
if (!pending.getAllFields().isEmpty()) {
logger.trace("Pending Command {}", pending.getAllFields());
}
} else if (pm.hasDebugMessage()) {
logger.trace("MB Debug Message: {}", pm.getDebugMessage().getMessage());
} else {
logger.trace("MB Message: {} not handled", pm.getAllFields());
}
} catch (IOException e) {
logger.trace("IOException decoding message {}", e.getMessage());
} catch (Error err) {
logger.debug("Error caught {}", err.getMessage());
}
}

public boolean distributeVepUpdates(Map<String, VEPUpdate> map) {
List<String> notFoundList = new ArrayList<>();
map.forEach((key, value) -> {
VehicleHandler h = activeVehicleHandlerMap.get(key);
if (h != null) {
h.distributeContent(value);
h.enqueueUpdate(value);
} else {
if (value.getFullUpdate()) {
vepUpdateMap.put(key, value);
Expand Down Expand Up @@ -430,7 +526,7 @@ public void sendCommand(@Nullable ClientMessage cm) {
if (cm != null) {
ws.setCommand(cm);
}
scheduler.schedule(this::update, 2, TimeUnit.SECONDS);
scheduler.schedule(this::refresh, 2, TimeUnit.SECONDS);
}

public void keepAlive(boolean b) {
Expand Down
47 changes: 43 additions & 4 deletions ...cedesme/src/main/java/org/openhab/binding/mercedesme/internal/handler/VehicleHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -134,6 +134,8 @@ public class VehicleHandler extends BaseThingHandler {
private JSONObject chargeGroupValueStorage = new JSONObject();
private Map<String, State> hvacGroupValueStorage = new HashMap<>();
private String vehicleType = NOT_SET;
private List<VEPUpdate> eventQueue = new ArrayList<>();
private boolean updateRunning = false;

Map<String, ChannelStateMap> eventStorage = new HashMap<>();
Optional<AccountHandler> accountHandler = Optional.empty();
Expand Down Expand Up @@ -182,6 +184,7 @@ public void dispose() {
accountHandler.ifPresent(ah -> {
ah.unregisterVin(config.get().vin);
});
eventQueue.clear();
super.dispose();
}

Expand Down Expand Up @@ -587,13 +590,49 @@ public void distributeCommandStatus(AppTwinCommandStatusUpdatesByPID cmdUpdates)
});
}

public void distributeContent(VEPUpdate data) {
public void enqueueUpdate(VEPUpdate update) {
synchronized (eventQueue) {
eventQueue.add(update);
scheduler.execute(this::scheduleUpdate);
}
}

private void scheduleUpdate() {
VEPUpdate data;
synchronized (eventQueue) {
while (updateRunning) {
try {
eventQueue.wait();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
eventQueue.clear();
return;
}
}
if (!eventQueue.isEmpty()) {
data = eventQueue.remove(0);
} else {
return;
}
updateRunning = true;
}
try {
handleUpdate(data);
} finally {
synchronized (eventQueue) {
updateRunning = false;
eventQueue.notifyAll();
}
}
}

public void handleUpdate(VEPUpdate update) {
updateStatus(ThingStatus.ONLINE);
boolean fullUpdate = data.getFullUpdate();
boolean fullUpdate = update.getFullUpdate();
/**
* Deliver proto update
*/
String newProto = Utils.proto2Json(data, thing.getThingTypeUID());
String newProto = Utils.proto2Json(update, thing.getThingTypeUID());
String combinedProto = newProto;
ChannelUID protoUpdateChannelUID = new ChannelUID(thing.getUID(), GROUP_VEHICLE, OH_CHANNEL_PROTO_UPDATE);
ChannelStateMap oldProtoMap = eventStorage.get(protoUpdateChannelUID.getId());
Expand All @@ -609,7 +648,7 @@ public void distributeContent(VEPUpdate data) {
StringType.valueOf(combinedProto));
updateChannel(dataUpdateMap);

Map<String, VehicleAttributeStatus> atts = data.getAttributesMap();
Map<String, VehicleAttributeStatus> atts = update.getAttributesMap();
/**
* handle "simple" values
*/
Expand Down
73 changes: 18 additions & 55 deletions ....mercedesme/src/main/java/org/openhab/binding/mercedesme/internal/server/MBWebsocket.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,13 +39,6 @@
import org.slf4j.LoggerFactory;

import com.daimler.mbcarkit.proto.Client.ClientMessage;
import com.daimler.mbcarkit.proto.Protos.AcknowledgeAssignedVehicles;
import com.daimler.mbcarkit.proto.VehicleEvents;
import com.daimler.mbcarkit.proto.VehicleEvents.AcknowledgeVEPUpdatesByVIN;
import com.daimler.mbcarkit.proto.VehicleEvents.PushMessage;
import com.daimler.mbcarkit.proto.Vehicleapi.AcknowledgeAppTwinCommandStatusUpdatesByVIN;
import com.daimler.mbcarkit.proto.Vehicleapi.AppTwinCommandStatusUpdatesByVIN;
import com.daimler.mbcarkit.proto.Vehicleapi.AppTwinPendingCommandsRequest;

/**
* {@link MBWebsocket} as socket endpoint to communicate with Mercedes
Expand Down Expand Up @@ -128,9 +121,10 @@ public void run() {
accountHandler.updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.COMMUNICATION_ERROR,
"@text/mercedesme.account.status.websocket-failure");
logger.warn("Websocket handling exception: {}", t.getMessage());
}
synchronized (this) {
running = false;
} finally {
synchronized (this) {
running = false;
}
}
}

Expand All @@ -157,7 +151,7 @@ private boolean sendMessage() {
return false;
}

private void sendAcknowledgeMessage(ClientMessage message) {
public void sendAcknowledgeMessage(ClientMessage message) {
try {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
message.writeTo(baos);
Expand All @@ -169,10 +163,6 @@ private void sendAcknowledgeMessage(ClientMessage message) {
}
}

public boolean isRunning() {
return running;
}

public void interrupt() {
synchronized (this) {
runTill = Instant.MIN;
Expand Down Expand Up @@ -202,47 +192,20 @@ public void keepAlive(boolean b) {
@OnWebSocketMessage
public void onBytes(InputStream is) {
try {
PushMessage pm = VehicleEvents.PushMessage.parseFrom(is);
if (pm.hasVepUpdates()) {
boolean distributed = accountHandler.distributeVepUpdates(pm.getVepUpdates().getUpdatesMap());
if (distributed) {
AcknowledgeVEPUpdatesByVIN ack = AcknowledgeVEPUpdatesByVIN.newBuilder()
.setSequenceNumber(pm.getVepUpdates().getSequenceNumber()).build();
ClientMessage cm = ClientMessage.newBuilder().setAcknowledgeVepUpdatesByVin(ack).build();
sendAcknowledgeMessage(cm);
}
} else if (pm.hasAssignedVehicles()) {
for (int i = 0; i < pm.getAssignedVehicles().getVinsCount(); i++) {
String vin = pm.getAssignedVehicles().getVins(0);
accountHandler.discovery(vin);
}
AcknowledgeAssignedVehicles ack = AcknowledgeAssignedVehicles.newBuilder().build();
ClientMessage cm = ClientMessage.newBuilder().setAcknowledgeAssignedVehicles(ack).build();
sendAcknowledgeMessage(cm);
} else if (pm.hasApptwinCommandStatusUpdatesByVin()) {
AppTwinCommandStatusUpdatesByVIN csubv = pm.getApptwinCommandStatusUpdatesByVin();
accountHandler.commandStatusUpdate(csubv.getUpdatesByVinMap());
AcknowledgeAppTwinCommandStatusUpdatesByVIN ack = AcknowledgeAppTwinCommandStatusUpdatesByVIN
.newBuilder().setSequenceNumber(csubv.getSequenceNumber()).build();
ClientMessage cm = ClientMessage.newBuilder().setAcknowledgeApptwinCommandStatusUpdateByVin(ack)
.build();
sendAcknowledgeMessage(cm);
} else if (pm.hasApptwinPendingCommandRequest()) {
AppTwinPendingCommandsRequest pending = pm.getApptwinPendingCommandRequest();
if (!pending.getAllFields().isEmpty()) {
logger.trace("Pending Command {}", pending.getAllFields());
}
} else if (pm.hasDebugMessage()) {
logger.trace("MB Debug Message: {}", pm.getDebugMessage().getMessage());
} else {
logger.trace("MB Message: {} not handled", pm.getAllFields());
}
byte[] array = is.readAllBytes();
is.close();
accountHandler.enqueueMessage(array);
/**
* https://community.openhab.org/t/mercedes-me/136866/12
* Release Websocket thread as early as possible to avoid execeptions
*
* 1. Websocket thread responsible for reading stream in bytes and enqueue for AccountHandler.
* 2. AccountHamdler thread responsible for encoding proto message. In case of update enqueue proto message
* at VehicleHandöer
* 3. VehicleHandler responsible to update channels
*/
} catch (IOException e) {
// don't report thing status errors here.
// Sometimes messages cannot be decoded which doesn't effect the overall functionality
logger.trace("IOException {}", e.getMessage());
} catch (Error err) {
logger.trace("Error caught {}", err.getMessage());
logger.debug("IOException reading input stream {}", e.getMessage());
}
}

Expand Down
Loading