demus/openhab-addons
1
0
Fork 0
mirror of https://github.com/danieldemus/openhab-addons synced 2026-07-29 12:34:21 +02:00
Code Issues Packages Projects Releases Wiki Activity

[tuya] Improve connection handling (#18829)

Browse Source 
* [tuya] Fix format of DP_REFRESH payload

Signed-off-by: Mike Jagdis <mjagdis@eris-associates.co.uk>
This commit is contained in:
mjagdis
2025-09-15 07:23:07 +02:00
committed by GitHub
parent c4dceefccd
commit 07f232af39
11 changed files with 303 additions and 291 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bundles/org.openhab.binding.tuya/README.md
+5 -5
View File
@@ -71,12 +71,12 @@ If you want to manually configure the device, you can also read those values fro
For line powered device on the same subnet `ip` address and `protocol` version are automatically detected.
Tuya devices announce their presence via UDP broadcast packets, which is usually not available in other subnets.
Battery powered devices do not announce their presence at all.
There is no clear rule how to determine if a device has protocol 3.3 or 3.1.
It is recommended to start with 3.3 and watch the log file if it that works and use 3.1 otherwise.
If automatic protocol version detection does not work there is no clear rule how to determine if a device has protocol 3.3 or 3.1.
In this case it is recommended to start with 3.3 and watch the log file. If 3.3 does not work try use 3.1.
Some devices do not automatically refresh channels (e.g. some power meters).
The `pollingInterval` can be increased from the default value `0` (off) to a minimum of 10s or higher.
The device is then requested to refresh its data channels and reports the status.
Some devices do not automatically refresh channels or only refresh at fairly long intervals even though the data is sampled at a much higher rate. (e.g. some power meters only send updates every 10 minutes but sample continuously.)
The `pollingInterval` can be used to adjust how often channels are updated and can be set to `0` (off) or to any integer value of 10 seconds or higher. The default is 10 seconds.
Note that this has no practical effect on battery powered devices. These only wake up when they have something to say and then go straight back to sleep.
In case something is not working, please open an issue on [GitHub](https://github.com/openhab/openhab-addons/issues/new?title=[tuya]) and add TRACE level logs.
bundles/org.openhab.binding.tuya/src/main/java/org/openhab/binding/tuya/internal/TuyaBindingConstants.java
+30 -4
View File
@@ -50,6 +50,8 @@ public class TuyaBindingConstants {
public static final String CONFIG_LOCAL_KEY = "localKey";
public static final String CONFIG_DEVICE_ID = "deviceId";
public static final String CONFIG_PRODUCT_ID = "productId";
public static final String CONFIG_IP = "ip";
public static final String CONFIG_PROTOCOL = "protocol";
public static final ChannelTypeUID CHANNEL_TYPE_UID_COLOR = new ChannelTypeUID(BINDING_ID, "color");
public static final ChannelTypeUID CHANNEL_TYPE_UID_DIMMER = new ChannelTypeUID(BINDING_ID, "dimmer");
@@ -59,10 +61,6 @@ public class TuyaBindingConstants {
public static final ChannelTypeUID CHANNEL_TYPE_UID_SWITCH = new ChannelTypeUID(BINDING_ID, "switch");
public static final ChannelTypeUID CHANNEL_TYPE_UID_IR_CODE = new ChannelTypeUID(BINDING_ID, "ir-code");
public static final int TCP_CONNECTION_HEARTBEAT_INTERVAL = 10; // in s
public static final int TCP_CONNECTION_TIMEOUT = 60; // in s;
public static final int TCP_CONNECTION_MAXIMUM_MISSED_HEARTBEATS = 3;
public static final Map<String, Map<String, SchemaDp>> SCHEMAS = getSchemas();
private static Map<String, Map<String, SchemaDp>> getSchemas() {
@@ -83,4 +81,32 @@ public class TuyaBindingConstants {
return Map.of();
}
}
// The heartbeat interval specifies the maximum amount of time that can pass without us
// sending anything to a device. Once the heartbeat interval is reached we send a heartbeat
// message to let the device know we are still present. Devices that do not receive traffic
// on a connection for some interval (unspecified but typically ~30 seconds) close or discard
// the connection.
public static final int TCP_CONNECTION_HEARTBEAT_INTERVAL = 15; // Seconds
// The amount of time a device has to respond to a message. If we don't see anything from
// the device for this long after sending a message we consider the connection dead, close
// it, and start trying to reconnect.
public static final int TCP_CONNECTION_MESSAGE_RESPONSE = 1; // Seconds
// How long to wait for a TCP session to connect before closing it and starting again. We do
// not rely on TCP's own retry strategy because that varies between implementations so we
// cannot know when the retry interval has become so great that it exceeds the amount of
// time a battery device may be awake.
public static final int TCP_CONNECT_TIMEOUT = 1000; // Milliseconds
// How long to wait before attempting another connection after the previous closed or failed.
// Note that if the previous attempt failed because the device was not reachable the interval
// between attempts will be TCP_CONNECT_RETRY_INTERVAL however if the previous attempt failed
// because the device was not responding the interval between attempts will be extended by
// TCP_CONNECT_TIMEOUT. In order to catch battery powered devices while they are awake the sum
// of TCP_CONNECT_TIMEOUT and TCP_CONNECT_RETRY_INTERVAL must therefore be small enough that at
// least one, preferably two or three, connection attempts will be made during the time the
// device is awake.
public static final int TCP_CONNECT_RETRY_INTERVAL = 1000; // Milliseconds
}
bundles/org.openhab.binding.tuya/src/main/java/org/openhab/binding/tuya/internal/handler/TuyaDeviceHandler.java
+70 -58
View File
@@ -19,6 +19,8 @@ import static org.openhab.binding.tuya.internal.TuyaBindingConstants.CHANNEL_TYP
import static org.openhab.binding.tuya.internal.TuyaBindingConstants.CHANNEL_TYPE_UID_QUANTITY;
import static org.openhab.binding.tuya.internal.TuyaBindingConstants.CHANNEL_TYPE_UID_STRING;
import static org.openhab.binding.tuya.internal.TuyaBindingConstants.CHANNEL_TYPE_UID_SWITCH;
import static org.openhab.binding.tuya.internal.TuyaBindingConstants.CONFIG_IP;
import static org.openhab.binding.tuya.internal.TuyaBindingConstants.CONFIG_PROTOCOL;
import static org.openhab.core.library.CoreItemFactory.NUMBER;
import java.math.BigDecimal;
@@ -34,7 +36,6 @@ import java.util.Objects;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import javax.measure.Unit;
@@ -109,10 +110,8 @@ public class TuyaDeviceHandler extends BaseThingHandler implements DeviceInfoSub
private final Map<String, SchemaDp> schemaDps;
private boolean oldColorMode = false;
private @Nullable ScheduledFuture<?> reconnectFuture;
private @Nullable ScheduledFuture<?> pollingJob;
private @Nullable ScheduledFuture<?> irLearnJob;
private boolean disposing = false;
private final Map<Integer, String> dpToChannelId = new HashMap<>();
private final Map<Integer, List<String>> dp2ToChannelId = new HashMap<>();
@@ -136,17 +135,13 @@ public class TuyaDeviceHandler extends BaseThingHandler implements DeviceInfoSub
@Override
public void processDeviceStatus(Map<Integer, Object> deviceStatus) {
logger.trace("'{}' received status message '{}'", thing.getUID(), deviceStatus);
// Older devices may need to use the control method to request device status.
if (deviceStatus.isEmpty()) {
// if status is empty -> need to use control method to request device status
Map<Integer, @Nullable Object> commandRequest = new HashMap<>();
dpToChannelId.keySet().forEach(dp -> commandRequest.put(dp, null));
dp2ToChannelId.keySet().forEach(dp -> commandRequest.put(dp, null));
TuyaDevice tuyaDevice = this.tuyaDevice;
if (tuyaDevice != null) {
tuyaDevice.set(commandRequest);
logger.debug("'{}' switching to control instead of query", thing.getUID());
tuyaDevice.setQueryUsesControl();
tuyaDevice.requestStatus(List.of());
}
return;
}
@@ -253,14 +248,29 @@ public class TuyaDeviceHandler extends BaseThingHandler implements DeviceInfoSub
@Override
public void connectionStatus(boolean status) {
if (status) {
logger.debug("{}: connected", thing.getUID().getId());
// Tuya devices are never offline (if they are battery devices they are expected
// to be unreachable practically all the time) so really we're just clearing the
// status message here rather than actually setting the Thing online.
updateStatus(ThingStatus.ONLINE);
int pollingInterval = configuration.pollingInterval;
TuyaDevice tuyaDevice = this.tuyaDevice;
if (tuyaDevice != null && pollingInterval > 0) {
pollingJob = scheduler.scheduleWithFixedDelay(() -> {
tuyaDevice.refreshStatus(
Stream.concat(dpToChannelId.keySet().stream(), dp2ToChannelId.keySet().stream()).toList());
}, pollingInterval, pollingInterval, TimeUnit.SECONDS);
if (tuyaDevice != null) {
// When we first connect the device state is unknown so we want to know everything
// it is willing to tell us.
tuyaDevice.requestStatus(List.of());
if (pollingJob == null) {
int pollingInterval = configuration.pollingInterval;
if (pollingInterval > 0) {
pollingJob = scheduler.scheduleWithFixedDelay(() -> {
tuyaDevice.refreshStatus(List.of());
}, pollingInterval, pollingInterval, TimeUnit.SECONDS);
}
} else {
logger.debug("{}: polling job already exists?!?", thing.getUID().getId());
}
}
// start learning code if thing is online and presents 'ir-code' channel
@@ -268,19 +278,16 @@ public class TuyaDeviceHandler extends BaseThingHandler implements DeviceInfoSub
.map(Map.Entry::getKey).findAny().map(channelIdToConfiguration::get)
.ifPresent(irCodeChannelConfig -> irStartLearning(irCodeChannelConfig.activeListen));
} else {
updateStatus(ThingStatus.OFFLINE);
logger.debug("{}: disconnected", thing.getUID().getId());
updateStatus(ThingStatus.ONLINE, ThingStatusDetail.NONE, "Waiting for device wake up");
ScheduledFuture<?> pollingJob = this.pollingJob;
if (pollingJob != null) {
pollingJob.cancel(true);
this.pollingJob = null;
}
TuyaDevice tuyaDevice = this.tuyaDevice;
ScheduledFuture<?> reconnectFuture = this.reconnectFuture;
// only re-connect if a device is present, we are not disposing the thing and either the reconnectFuture is
// empty or already done
if (tuyaDevice != null && !disposing && (reconnectFuture == null || reconnectFuture.isDone())) {
this.reconnectFuture = scheduler.schedule(this::connectDevice, 5000, TimeUnit.MILLISECONDS);
}
if (channelIdToChannelTypeUID.containsValue(CHANNEL_TYPE_UID_IR_CODE)) {
irStopLearning();
}
@@ -447,24 +454,22 @@ public class TuyaDeviceHandler extends BaseThingHandler implements DeviceInfoSub
@Override
public void dispose() {
disposing = true;
ScheduledFuture<?> future = reconnectFuture;
logger.debug("{}: dispose", thing.getUID().getId());
ScheduledFuture<?> future = this.pollingJob;
if (future != null) {
this.pollingJob = null;
future.cancel(true);
}
future = this.pollingJob;
if (future != null) {
future.cancel(true);
}
if (configuration.ip.isEmpty()) {
// unregister listener only if IP is not fixed
udpDiscoveryListener.unregisterListener(this);
}
udpDiscoveryListener.unregisterListener(this);
TuyaDevice tuyaDevice = this.tuyaDevice;
if (tuyaDevice != null) {
tuyaDevice.dispose();
this.tuyaDevice = null;
tuyaDevice.dispose();
}
irStopLearning();
dpToChannelId.clear();
@@ -485,27 +490,45 @@ public class TuyaDeviceHandler extends BaseThingHandler implements DeviceInfoSub
thing.getChannels().forEach(this::configureChannel);
if (!configuration.ip.isBlank()) {
deviceInfoChanged(new DeviceInfo(configuration.ip, configuration.protocol));
updateStatus(ThingStatus.ONLINE, ThingStatusDetail.NONE, "Waiting for device wake up");
this.tuyaDevice = new TuyaDevice(gson, this, eventLoopGroup, configuration.deviceId,
configuration.localKey.getBytes(StandardCharsets.UTF_8), configuration.ip, configuration.protocol);
} else {
updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.CONFIGURATION_PENDING, "Waiting for IP address");
udpDiscoveryListener.registerListener(configuration.deviceId, this);
}
disposing = false;
udpDiscoveryListener.registerListener(configuration.deviceId, this);
}
@Override
public void deviceInfoChanged(DeviceInfo deviceInfo) {
logger.info("Configuring IP address '{}' for thing '{}'.", deviceInfo, thing.getUID());
if (!configuration.ip.equals(deviceInfo.ip) || !configuration.protocol.equals(deviceInfo.protocolVersion)) {
logger.info("Configuring IP address '{}' for thing '{}'.", deviceInfo, thing.getUID());
TuyaDevice tuyaDevice = this.tuyaDevice;
if (tuyaDevice != null) {
tuyaDevice.dispose();
TuyaDevice tuyaDevice = this.tuyaDevice;
if (tuyaDevice != null) {
this.tuyaDevice = null;
tuyaDevice.dispose();
}
try {
Configuration newConfig = editConfiguration();
newConfig.put(CONFIG_IP, deviceInfo.ip);
newConfig.put(CONFIG_PROTOCOL, deviceInfo.protocolVersion);
updateConfiguration(newConfig);
configuration.ip = deviceInfo.ip;
configuration.protocol = deviceInfo.protocolVersion;
this.tuyaDevice = new TuyaDevice(gson, this, eventLoopGroup, configuration.deviceId,
configuration.localKey.getBytes(StandardCharsets.UTF_8), configuration.ip,
configuration.protocol);
} catch (IllegalArgumentException e) {
logger.warn("{}", e.getMessage());
updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.COMMUNICATION_ERROR, e.getMessage());
}
}
updateStatus(ThingStatus.UNKNOWN);
this.tuyaDevice = new TuyaDevice(gson, this, eventLoopGroup, configuration.deviceId,
configuration.localKey.getBytes(StandardCharsets.UTF_8), deviceInfo.ip, deviceInfo.protocolVersion);
}
private void addChannels() {
@@ -655,17 +678,6 @@ public class TuyaDeviceHandler extends BaseThingHandler implements DeviceInfoSub
}
}
private void connectDevice() {
TuyaDevice tuyaDevice = this.tuyaDevice;
if (tuyaDevice == null) {
logger.warn("Cannot connect {} because the device is not set.", thing.getUID());
return;
}
// clear the future here because timing issues can prevent the next attempt if we fail again
reconnectFuture = null;
tuyaDevice.connect();
}
private List<CommandOption> toCommandOptionList(List<String> options) {
return options.stream()
.map(c -> new CommandOption(c, StringUtils.capitalizeByWhitespace(c.replaceAll("_", " ")))).toList();
bundles/org.openhab.binding.tuya/src/main/java/org/openhab/binding/tuya/internal/local/TuyaDevice.java
+153 -43
View File
@@ -13,22 +13,27 @@
package org.openhab.binding.tuya.internal.local;
import static org.openhab.binding.tuya.internal.TuyaBindingConstants.TCP_CONNECTION_HEARTBEAT_INTERVAL;
import static org.openhab.binding.tuya.internal.TuyaBindingConstants.TCP_CONNECTION_TIMEOUT;
import static org.openhab.binding.tuya.internal.TuyaBindingConstants.TCP_CONNECTION_MESSAGE_RESPONSE;
import static org.openhab.binding.tuya.internal.TuyaBindingConstants.TCP_CONNECT_RETRY_INTERVAL;
import static org.openhab.binding.tuya.internal.TuyaBindingConstants.TCP_CONNECT_TIMEOUT;
import static org.openhab.binding.tuya.internal.local.CommandType.CONTROL;
import static org.openhab.binding.tuya.internal.local.CommandType.CONTROL_NEW;
import static org.openhab.binding.tuya.internal.local.CommandType.DP_QUERY;
import static org.openhab.binding.tuya.internal.local.CommandType.DP_REFRESH;
import static org.openhab.binding.tuya.internal.local.CommandType.HEART_BEAT;
import static org.openhab.binding.tuya.internal.local.CommandType.SESS_KEY_NEG_START;
import static org.openhab.binding.tuya.internal.local.ProtocolVersion.V3_4;
import static org.openhab.binding.tuya.internal.local.ProtocolVersion.V3_5;
import java.util.List;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import org.eclipse.jdt.annotation.NonNullByDefault;
import org.eclipse.jdt.annotation.Nullable;
import org.openhab.binding.tuya.internal.local.handlers.HeartbeatHandler;
import org.openhab.binding.tuya.internal.local.handlers.TuyaDecoder;
import org.openhab.binding.tuya.internal.local.handlers.TuyaEncoder;
import org.openhab.binding.tuya.internal.local.handlers.TuyaMessageHandler;
@@ -43,12 +48,15 @@ import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.ChannelPromise;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.AttributeKey;
@@ -67,7 +75,9 @@ public class TuyaDevice implements ChannelFutureListener {
private final Logger logger = LoggerFactory.getLogger(TuyaDevice.class);
private final Bootstrap bootstrap = new Bootstrap();
private @Nullable ScheduledFuture<?> reconnectFuture;
private final DeviceStatusListener deviceStatusListener;
private final EventLoopGroup eventLoopGroup;
private final String deviceId;
private final byte[] deviceKey;
@@ -75,41 +85,89 @@ public class TuyaDevice implements ChannelFutureListener {
private final ProtocolVersion protocolVersion;
private @Nullable Channel channel;
private boolean queryUsesControl = false;
private class HeartbeatSender extends IdleStateHandler {
public HeartbeatSender() {
super(0, TCP_CONNECTION_HEARTBEAT_INTERVAL, 0);
}
@Override
public void write(@Nullable ChannelHandlerContext ctx, @Nullable Object msg, @Nullable ChannelPromise promise)
throws Exception {
if (ctx != null) {
// All messages sent to the device should trigger a timely response.
ctx.pipeline().replace(this, "idleHandler", new IdleHandler());
ctx.write(msg, promise);
}
}
@Override
protected void channelIdle(@Nullable ChannelHandlerContext ctx, @Nullable IdleStateEvent evt) throws Exception {
if (ctx != null) {
logger.trace("{}{}: Sending heart beat", deviceId, address);
ctx.channel().writeAndFlush(new MessageWrapper<>(HEART_BEAT, Map.of("dps", "")));
}
}
}
private class IdleHandler extends IdleStateHandler {
public IdleHandler() {
super(TCP_CONNECTION_MESSAGE_RESPONSE, 0, 0);
}
@Override
public void channelRead(@Nullable ChannelHandlerContext ctx, @Nullable Object msg) throws Exception {
if (ctx != null) {
ctx.pipeline().replace(this, "heartbeatSender", new HeartbeatSender());
ctx.fireChannelRead(msg);
}
}
@Override
protected void channelIdle(@Nullable ChannelHandlerContext ctx, @Nullable IdleStateEvent evt) throws Exception {
if (ctx != null) {
logger.debug("{}{}: Connection seems to be dead.", deviceId, address);
ctx.close();
}
}
}
public TuyaDevice(Gson gson, DeviceStatusListener deviceStatusListener, EventLoopGroup eventLoopGroup,
String deviceId, byte[] deviceKey, String address, String protocolVersion) {
this.address = address;
this.deviceStatusListener = deviceStatusListener;
this.eventLoopGroup = eventLoopGroup;
this.deviceId = deviceId;
this.deviceKey = deviceKey;
this.deviceStatusListener = deviceStatusListener;
this.address = address;
this.protocolVersion = ProtocolVersion.fromString(protocolVersion);
bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class);
bootstrap.option(ChannelOption.TCP_NODELAY, true).option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 2000);
bootstrap.option(ChannelOption.TCP_NODELAY, true).option(ChannelOption.CONNECT_TIMEOUT_MILLIS,
TCP_CONNECT_TIMEOUT);
bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("idleStateHandler",
new IdleStateHandler(TCP_CONNECTION_TIMEOUT, TCP_CONNECTION_HEARTBEAT_INTERVAL, 0));
pipeline.addLast("messageEncoder", new TuyaEncoder(gson));
pipeline.addLast("messageDecoder", new TuyaDecoder(gson));
pipeline.addLast("heartbeatHandler", new HeartbeatHandler());
pipeline.addLast("heartbeatSender", new HeartbeatSender());
pipeline.addLast("deviceHandler", new TuyaMessageHandler(deviceStatusListener));
pipeline.addLast("userEventHandler", new UserEventHandler());
}
});
connect();
}
public void connect() {
bootstrap.connect(address, 6668).addListener(this);
public void setQueryUsesControl() {
queryUsesControl = true;
}
private void disconnect() {
Channel channel = this.channel;
if (channel != null) { // if channel == null we are not connected anyway
channel.pipeline().fireUserEventTriggered(new UserEventHandler.DisposeEvent());
this.channel = null;
}
private void connect() {
logger.trace("{}: connecting", deviceId);
bootstrap.connect(address, 6668).addListener(this);
}
public void set(Map<Integer, @Nullable Object> command) {
@@ -123,60 +181,112 @@ public class TuyaDevice implements ChannelFutureListener {
}
}
public void requestStatus() {
MessageWrapper<?> m = new MessageWrapper<>(DP_QUERY, Map.of("dps", Map.of()));
Channel channel = this.channel;
if (channel != null) {
channel.writeAndFlush(m);
public void requestStatus(Collection<Integer> dps) {
if (!queryUsesControl) {
// CommandType commandType = (protocolVersion == V3_4 || protocolVersion == V3_5) ? DP_QUERY_NEW : DP_QUERY;
CommandType commandType = DP_QUERY;
MessageWrapper<?> m = new MessageWrapper<>(commandType, Map.of("dps", dps));
Channel channel = this.channel;
if (channel != null) {
channel.writeAndFlush(m);
} else {
logger.warn("{}: Querying status failed. Device is not connected.", deviceId);
}
} else {
logger.warn("{}: Querying status failed. Device is not connected.", deviceId);
Map<Integer, @Nullable Object> dpMap = new HashMap<>();
dps.forEach(dp -> dpMap.put(dp, null));
set(dpMap);
}
}
public void refreshStatus(List<Integer> dps) {
public void refreshStatus(Collection<Integer> dps) {
MessageWrapper<?> m = new MessageWrapper<>(DP_REFRESH, Map.of("dpId", dps));
Channel channel = this.channel;
if (channel != null) {
channel.writeAndFlush(m);
requestStatus();
// We could try a requestStatus(dps) here however it shouldn't be necessary as
// once new values for the DPs have been sampled the device should send an update
// (but not necessarily if the new values are the same as the old).
} else {
logger.warn("{}: Refreshing status failed. Device is not connected.", deviceId);
}
}
public void dispose() {
disconnect();
logger.debug("{}: disposed", deviceId);
Channel channel = this.channel;
this.channel = null;
if (channel != null) {
channel.closeFuture().cancel(true);
}
ScheduledFuture<?> future = reconnectFuture;
if (future != null) {
future.cancel(true);
reconnectFuture = null;
}
if (channel != null) {
channel.close();
}
}
@Override
public void operationComplete(@NonNullByDefault({}) ChannelFuture channelFuture) throws Exception {
if (channelFuture.isSuccess()) {
logger.debug("{}{}: channel connected", deviceId,
Objects.requireNonNullElse(channelFuture.channel().remoteAddress(), ""));
Channel channel = channelFuture.channel();
channel.attr(DEVICE_ID_ATTR).set(deviceId);
channel.attr(PROTOCOL_ATTR).set(protocolVersion);
// session key is device key before negotiation
channel.attr(SESSION_KEY_ATTR).set(deviceKey);
if (protocolVersion == V3_4 || protocolVersion == V3_5) {
byte[] sessionRandom = CryptoUtil.generateRandom(16);
channel.attr(SESSION_RANDOM_ATTR).set(sessionRandom);
if (channel != null) {
this.channel = channel;
// handshake for session key required
MessageWrapper<?> m = new MessageWrapper<>(SESS_KEY_NEG_START, sessionRandom);
channel.writeAndFlush(m);
} else {
this.channel = channel;
channel.closeFuture().addListener(new ChannelFutureListener() {
@Override
public void operationComplete(@NonNullByDefault({}) ChannelFuture channelFuture) {
logger.debug("{}{}: channel closed", deviceId,
Objects.requireNonNullElse(channelFuture.channel().remoteAddress(), ""));
// no handshake for 3.1/3.3
requestStatus();
deviceStatusListener.connectionStatus(false);
if (!channelFuture.isCancelled()) {
reconnectFuture = eventLoopGroup.schedule(() -> {
logger.debug("{}{}: reconnect", deviceId,
Objects.requireNonNullElse(channelFuture.channel().remoteAddress(), ""));
connect();
}, TCP_CONNECT_RETRY_INTERVAL, TimeUnit.MILLISECONDS);
}
}
});
channel.attr(DEVICE_ID_ATTR).set(deviceId);
channel.attr(PROTOCOL_ATTR).set(protocolVersion);
// session key is device key before negotiation
channel.attr(SESSION_KEY_ATTR).set(deviceKey);
if (protocolVersion == V3_4 || protocolVersion == V3_5) {
byte[] sessionRandom = CryptoUtil.generateRandom(16);
channel.attr(SESSION_RANDOM_ATTR).set(sessionRandom);
// handshake for session key required
MessageWrapper<?> m = new MessageWrapper<>(SESS_KEY_NEG_START, sessionRandom);
channel.writeAndFlush(m);
} else {
// no handshake for 3.1/3.3
deviceStatusListener.connectionStatus(true);
}
}
} else {
logger.debug("{}{}: Failed to connect: {}", deviceId,
logger.trace("{}{}: Failed to connect: {}", deviceId,
Objects.requireNonNullElse(channelFuture.channel().remoteAddress(), ""),
channelFuture.cause().getMessage());
this.channel = null;
deviceStatusListener.connectionStatus(false);
channelFuture.channel().close();
reconnectFuture = eventLoopGroup.schedule(this::connect, TCP_CONNECT_RETRY_INTERVAL, TimeUnit.MILLISECONDS);
}
}
}
bundles/org.openhab.binding.tuya/src/main/java/org/openhab/binding/tuya/internal/local/UdpDiscoveryListener.java
+7 -16
View File
@@ -12,8 +12,8 @@
*/
package org.openhab.binding.tuya.internal.local;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.eclipse.jdt.annotation.NonNullByDefault;
import org.openhab.binding.tuya.internal.local.dto.DeviceInfo;
@@ -52,8 +52,8 @@ public class UdpDiscoveryListener implements ChannelFutureListener {
private final Gson gson = new Gson();
private final Map<String, DeviceInfo> deviceInfos = new HashMap<>();
private final Map<String, DeviceInfoSubscriber> deviceListeners = new HashMap<>();
private final Map<String, DeviceInfo> deviceInfos = new ConcurrentHashMap<>();
private final Map<String, DeviceInfoSubscriber> deviceListeners = new ConcurrentHashMap<>();
private @NonNullByDefault({}) Channel encryptedChannel;
private @NonNullByDefault({}) Channel encryptedChannel35;
@@ -101,16 +101,9 @@ public class UdpDiscoveryListener implements ChannelFutureListener {
public void deactivate() {
deactivate = true;
encryptedChannel.pipeline().fireUserEventTriggered(new UserEventHandler.DisposeEvent());
encryptedChannel35.pipeline().fireUserEventTriggered(new UserEventHandler.DisposeEvent());
rawChannel.pipeline().fireUserEventTriggered(new UserEventHandler.DisposeEvent());
try {
encryptedChannel.closeFuture().sync();
encryptedChannel35.closeFuture().sync();
rawChannel.closeFuture().sync();
} catch (InterruptedException e) {
// do nothing
}
encryptedChannel.close();
encryptedChannel35.close();
rawChannel.close();
}
public void registerListener(String deviceId, DeviceInfoSubscriber subscriber) {
@@ -124,9 +117,7 @@ public class UdpDiscoveryListener implements ChannelFutureListener {
}
public void unregisterListener(DeviceInfoSubscriber deviceInfoSubscriber) {
if (!deviceListeners.entrySet().removeIf(e -> deviceInfoSubscriber.equals(e.getValue()))) {
logger.warn("Tried to unregister a listener for '{}' but no registration found.", deviceInfoSubscriber);
}
deviceListeners.entrySet().removeIf(e -> deviceInfoSubscriber.equals(e.getValue()));
}
@Override
bundles/org.openhab.binding.tuya/src/main/java/org/openhab/binding/tuya/internal/local/handlers/HeartbeatHandler.java
-100
View File
@@ -1,100 +0,0 @@
/*
* Copyright (c) 2010-2025 Contributors to the openHAB project
*
* See the NOTICE file(s) distributed with this work for additional
* information.
*
* This program and the accompanying materials are made available under the
* terms of the Eclipse Public License 2.0 which is available at
* http://www.eclipse.org/legal/epl-2.0
*
* SPDX-License-Identifier: EPL-2.0
*/
package org.openhab.binding.tuya.internal.local.handlers;
import static org.openhab.binding.tuya.internal.TuyaBindingConstants.TCP_CONNECTION_MAXIMUM_MISSED_HEARTBEATS;
import static org.openhab.binding.tuya.internal.TuyaBindingConstants.TCP_CONNECTION_TIMEOUT;
import java.util.Map;
import java.util.Objects;
import org.eclipse.jdt.annotation.NonNullByDefault;
import org.openhab.binding.tuya.internal.local.CommandType;
import org.openhab.binding.tuya.internal.local.MessageWrapper;
import org.openhab.binding.tuya.internal.local.TuyaDevice;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
/**
* The {@link HeartbeatHandler} is responsible for sending and receiving heartbeat messages
*
* @author Jan N. Klug - Initial contribution
*/
@NonNullByDefault
public class HeartbeatHandler extends ChannelDuplexHandler {
private final Logger logger = LoggerFactory.getLogger(HeartbeatHandler.class);
private int heartBeatMissed = 0;
@Override
public void userEventTriggered(@NonNullByDefault({}) ChannelHandlerContext ctx, @NonNullByDefault({}) Object evt)
throws Exception {
if (!ctx.channel().hasAttr(TuyaDevice.DEVICE_ID_ATTR)) {
logger.warn("{}: Failed to retrieve deviceId from ChannelHandlerContext. This is a bug.",
Objects.requireNonNullElse(ctx.channel().remoteAddress(), ""));
return;
}
String deviceId = ctx.channel().attr(TuyaDevice.DEVICE_ID_ATTR).get();
if (evt instanceof IdleStateEvent e) {
if (IdleState.READER_IDLE.equals(e.state())) {
logger.warn("{}{}: Did not receive a message from for {} seconds. Connection seems to be dead.",
deviceId, Objects.requireNonNullElse(ctx.channel().remoteAddress(), ""),
TCP_CONNECTION_TIMEOUT);
ctx.close();
} else if (IdleState.WRITER_IDLE.equals(e.state())) {
heartBeatMissed++;
if (heartBeatMissed > TCP_CONNECTION_MAXIMUM_MISSED_HEARTBEATS) {
logger.warn("{}{}: Missed more than {} heartbeat responses. Connection seems to be dead.", deviceId,
Objects.requireNonNullElse(ctx.channel().remoteAddress(), ""),
TCP_CONNECTION_MAXIMUM_MISSED_HEARTBEATS);
ctx.close();
} else {
logger.trace("{}{}: Sending ping", deviceId,
Objects.requireNonNullElse(ctx.channel().remoteAddress(), ""));
ctx.channel().writeAndFlush(new MessageWrapper<>(CommandType.HEART_BEAT, Map.of("dps", "")));
}
}
} else {
super.userEventTriggered(ctx, evt);
}
}
@Override
public void channelRead(@NonNullByDefault({}) ChannelHandlerContext ctx, @NonNullByDefault({}) Object msg)
throws Exception {
if (!ctx.channel().hasAttr(TuyaDevice.DEVICE_ID_ATTR)) {
logger.warn("{}: Failed to retrieve deviceId from ChannelHandlerContext. This is a bug.",
Objects.requireNonNullElse(ctx.channel().remoteAddress(), ""));
return;
}
String deviceId = ctx.channel().attr(TuyaDevice.DEVICE_ID_ATTR).get();
if (msg instanceof MessageWrapper<?> m) {
if (CommandType.HEART_BEAT.equals(m.commandType)) {
logger.trace("{}{}: Received pong", deviceId,
Objects.requireNonNullElse(ctx.channel().remoteAddress(), ""));
heartBeatMissed = 0;
// do not forward HEART_BEAT messages
ctx.fireChannelReadComplete();
return;
}
}
// forward to next handler
ctx.fireChannelRead(msg);
}
}
bundles/org.openhab.binding.tuya/src/main/java/org/openhab/binding/tuya/internal/local/handlers/TuyaDecoder.java
+21 -5
View File
@@ -15,6 +15,8 @@ package org.openhab.binding.tuya.internal.local.handlers;
import static org.openhab.binding.tuya.internal.local.CommandType.BROADCAST_LPV34;
import static org.openhab.binding.tuya.internal.local.CommandType.DP_QUERY;
import static org.openhab.binding.tuya.internal.local.CommandType.DP_QUERY_NOT_SUPPORTED;
import static org.openhab.binding.tuya.internal.local.CommandType.HEART_BEAT;
import static org.openhab.binding.tuya.internal.local.CommandType.SESS_KEY_NEG_RESPONSE;
import static org.openhab.binding.tuya.internal.local.CommandType.STATUS;
import static org.openhab.binding.tuya.internal.local.CommandType.UDP;
import static org.openhab.binding.tuya.internal.local.CommandType.UDP_NEW;
@@ -102,7 +104,7 @@ public class TuyaDecoder extends ByteToMessageDecoder {
if (prefix == 0x006699 && protocol != V3_5) {
protocol = V3_5;
logger.debug("Set protocol version to {}", protocol.getString());
logger.trace("Set protocol version to {}", protocol.getString());
}
int headerLength = protocol == V3_5 ? 22 : 16;
@@ -192,10 +194,13 @@ public class TuyaDecoder extends ByteToMessageDecoder {
}
MessageWrapper<?> m;
String decodedString;
if (commandType == UDP) {
// UDP is unencrypted
decodedString = new String(payload);
m = new MessageWrapper<>(commandType,
Objects.requireNonNull(gson.fromJson(new String(payload), DiscoveryMessage.class)));
Objects.requireNonNull(gson.fromJson(decodedString, DiscoveryMessage.class)));
} else {
byte[] decodedMessage = switch (protocol) {
case V3_5 -> CryptoUtil.decryptAesGcm(payload, sessionKey, header, null);
@@ -224,8 +229,9 @@ public class TuyaDecoder extends ByteToMessageDecoder {
HexUtils.bytesToHex(decodedMessage));
}
decodedString = new String(decodedMessage).trim();
try {
String decodedString = new String(decodedMessage).trim();
if (commandType == DP_QUERY && "json obj data unvalid".equals(decodedString)) {
// "json obj data unvalid" would also result in a JSONSyntaxException but is a known error when
// DP_QUERY is not supported by the device. Using a CONTROL message with null values is a known
@@ -239,8 +245,10 @@ public class TuyaDecoder extends ByteToMessageDecoder {
} else if (commandType == UDP_NEW || commandType == BROADCAST_LPV34) {
m = new MessageWrapper<>(commandType,
Objects.requireNonNull(gson.fromJson(decodedString, DiscoveryMessage.class)));
} else {
} else if (commandType == SESS_KEY_NEG_RESPONSE) {
m = new MessageWrapper<>(commandType, decodedMessage);
} else {
m = new MessageWrapper<>(commandType, decodedString);
}
} catch (JsonSyntaxException e) {
logger.warn("{}{} failed to parse JSON: {}", deviceId,
@@ -249,7 +257,15 @@ public class TuyaDecoder extends ByteToMessageDecoder {
}
}
logger.debug("{}{}: Received {}", deviceId, Objects.requireNonNullElse(ctx.channel().remoteAddress(), ""), m);
if (m.commandType != HEART_BEAT && m.commandType != UDP_NEW && m.commandType != UDP
&& m.commandType != BROADCAST_LPV34) {
logger.debug("{}{}: Received {}", deviceId, Objects.requireNonNullElse(ctx.channel().remoteAddress(), ""),
m);
} else {
logger.trace("{}{}: Received {}", deviceId, Objects.requireNonNullElse(ctx.channel().remoteAddress(), ""),
m);
}
out.add(m);
}
}
bundles/org.openhab.binding.tuya/src/main/java/org/openhab/binding/tuya/internal/local/handlers/TuyaEncoder.java
+10 -5
View File
@@ -88,7 +88,7 @@ public class TuyaEncoder extends MessageToByteEncoder<MessageWrapper<?>> {
if (msg.content == null || msg.content instanceof Map<?, ?>) {
Map<String, Object> content = (Map<String, Object>) msg.content;
Map<String, Object> payload = new HashMap<>();
if (msg.commandType == REQ_DEVINFO) {
if (msg.commandType == REQ_DEVINFO || msg.commandType == DP_REFRESH) {
if (content != null) {
payload.putAll(content);
}
@@ -112,15 +112,20 @@ public class TuyaEncoder extends MessageToByteEncoder<MessageWrapper<?>> {
}
}
logger.debug("{}{}: Sending {}, payload {}", deviceId,
Objects.requireNonNullElse(ctx.channel().remoteAddress(), ""), msg.commandType, payload);
if (msg.commandType != REQ_DEVINFO && msg.commandType != HEART_BEAT) {
logger.debug("{}{}: Sending {}, payload {}", deviceId,
Objects.requireNonNullElse(ctx.channel().remoteAddress(), ""), msg.commandType, payload);
} else {
logger.trace("{}{}: Sending {}, payload {}", deviceId,
Objects.requireNonNullElse(ctx.channel().remoteAddress(), ""), msg.commandType, payload);
}
String json = gson.toJson(payload);
payloadBytes = json.getBytes(StandardCharsets.UTF_8);
} else if (msg.content instanceof byte[] contentBytes) {
if (logger.isDebugEnabled()) {
logger.debug("{}{}: Sending payload {}", deviceId,
Objects.requireNonNullElse(ctx.channel().remoteAddress(), ""),
logger.debug("{}{}: Sending {}, payload {}", deviceId,
Objects.requireNonNullElse(ctx.channel().remoteAddress(), ""), msg.commandType,
HexUtils.bytesToHex(contentBytes));
}
payloadBytes = contentBytes.clone();
bundles/org.openhab.binding.tuya/src/main/java/org/openhab/binding/tuya/internal/local/handlers/TuyaMessageHandler.java
+2 -28
View File
@@ -49,34 +49,6 @@ public class TuyaMessageHandler extends ChannelDuplexHandler {
this.deviceStatusListener = deviceStatusListener;
}
@Override
public void channelActive(@NonNullByDefault({}) ChannelHandlerContext ctx) throws Exception {
if (!ctx.channel().hasAttr(TuyaDevice.DEVICE_ID_ATTR) || !ctx.channel().hasAttr(SESSION_KEY_ATTR)) {
logger.warn("{}: Failed to retrieve deviceId or sessionKey from ChannelHandlerContext. This is a bug.",
Objects.requireNonNullElse(ctx.channel().remoteAddress(), ""));
return;
}
String deviceId = ctx.channel().attr(TuyaDevice.DEVICE_ID_ATTR).get();
logger.debug("{}{}: Connection established.", deviceId,
Objects.requireNonNullElse(ctx.channel().remoteAddress(), ""));
deviceStatusListener.connectionStatus(true);
}
@Override
public void channelInactive(@NonNullByDefault({}) ChannelHandlerContext ctx) throws Exception {
if (!ctx.channel().hasAttr(TuyaDevice.DEVICE_ID_ATTR) || !ctx.channel().hasAttr(SESSION_KEY_ATTR)) {
logger.warn("{}: Failed to retrieve deviceId or sessionKey from ChannelHandlerContext. This is a bug.",
Objects.requireNonNullElse(ctx.channel().remoteAddress(), ""));
return;
}
String deviceId = ctx.channel().attr(TuyaDevice.DEVICE_ID_ATTR).get();
logger.debug("{}{}: Connection terminated.", deviceId,
Objects.requireNonNullElse(ctx.channel().remoteAddress(), ""));
deviceStatusListener.connectionStatus(false);
}
@Override
@SuppressWarnings("unchecked")
public void channelRead(@NonNullByDefault({}) ChannelHandlerContext ctx, @NonNullByDefault({}) Object msg)
@@ -137,6 +109,8 @@ public class TuyaMessageHandler extends ChannelDuplexHandler {
return;
}
ctx.channel().attr(TuyaDevice.SESSION_KEY_ATTR).set(newSessionKey);
deviceStatusListener.connectionStatus(true);
}
}
}
bundles/org.openhab.binding.tuya/src/main/java/org/openhab/binding/tuya/internal/local/handlers/UserEventHandler.java
+3 -21
View File
@@ -32,21 +32,6 @@ import io.netty.channel.ChannelHandlerContext;
public class UserEventHandler extends ChannelDuplexHandler {
private final Logger logger = LoggerFactory.getLogger(UserEventHandler.class);
@Override
public void userEventTriggered(@NonNullByDefault({}) ChannelHandlerContext ctx, @NonNullByDefault({}) Object evt) {
if (!ctx.channel().hasAttr(TuyaDevice.DEVICE_ID_ATTR)) {
logger.warn("Failed to retrieve deviceId from ChannelHandlerContext. This is a bug.");
return;
}
String deviceId = ctx.channel().attr(TuyaDevice.DEVICE_ID_ATTR).get();
if (evt instanceof DisposeEvent) {
logger.debug("{}{}: Received DisposeEvent, closing channel", deviceId,
Objects.requireNonNullElse(ctx.channel().remoteAddress(), ""));
ctx.close();
}
}
@Override
public void exceptionCaught(@NonNullByDefault({}) ChannelHandlerContext ctx, @NonNullByDefault({}) Throwable cause)
throws Exception {
@@ -59,16 +44,13 @@ public class UserEventHandler extends ChannelDuplexHandler {
String deviceId = ctx.channel().attr(TuyaDevice.DEVICE_ID_ATTR).get();
if (cause instanceof IOException) {
logger.debug("{}{}: IOException caught, closing channel.", deviceId,
Objects.requireNonNullElse(ctx.channel().remoteAddress(), ""), cause);
logger.debug("IOException caught: ", cause);
logger.debug("{}{}: {}, closing channel.", deviceId,
Objects.requireNonNullElse(ctx.channel().remoteAddress(), ""), cause.getMessage());
} else {
logger.warn("{}{}: {} caught, closing the channel", deviceId,
Objects.requireNonNullElse(ctx.channel().remoteAddress(), ""), cause.getClass(), cause);
}
ctx.close();
}
public static class DisposeEvent {
}
}
bundles/org.openhab.binding.tuya/src/main/resources/OH-INF/thing/thing-types.xml
+2 -6
View File
@@ -90,13 +90,9 @@
<limitToOptions>true</limitToOptions>
<advanced>true</advanced>
</parameter>
<parameter name="pollingInterval" type="integer" min="10" unit="s">
<parameter name="pollingInterval" type="integer" min="0" unit="s">
<label>Polling Interval</label>
<options>
<option value="0">disabled</option>
</options>
<default>0</default>
<limitToOptions>false</limitToOptions>
<default>10</default>
<advanced>true</advanced>
</parameter>
</config-description>