[linkplay] Stability fixes (#21025)

* [linkplay] Add self-healing reconnect loop for offline devices

Signed-off-by: Dan Cunningham <dan@digitaldan.com>

bundles/org.openhab.binding.linkplay/src/main/java/org/openhab/binding/linkplay/internal/LinkPlayHandlerFactory.java

@@ -25,6 +25,8 @@ import java.util.concurrent.ScheduledExecutorService;
 import org.eclipse.jdt.annotation.NonNullByDefault;
 import org.eclipse.jdt.annotation.Nullable;
 import org.eclipse.jetty.client.HttpClient;
+import org.eclipse.jetty.client.api.Request;
+import org.eclipse.jetty.http.HttpHeader;
 import org.eclipse.jetty.util.ssl.SslContextFactory;
 import org.jupnp.UpnpService;
 import org.jupnp.model.meta.LocalDevice;
@@ -94,6 +96,18 @@ public class LinkPlayHandlerFactory extends BaseThingHandlerFactory implements R
         this.networkAddressService = networkAddressService;
         upnpService.getRegistry().addListener(this);
         httpClient = httpClientFactory.createHttpClient("linkplay", new SslContextFactory.Client(true));
+        // LinkPlay devices aggressively close idle keep alive connections after a few seconds, so a pooled connection
+        // is frequently already dead when reused and the next request fails with an EOFException before it reaches the
+        // device. HTTP requests are infrequent (user commands plus periodic polling), so disable connection reuse
+        // entirely.
+        httpClient.getRequestListeners().add(new Request.Listener.Adapter() {
+            @Override
+            public void onBegin(@Nullable Request request) {
+                if (request != null) {
+                    request.header(HttpHeader.CONNECTION, "close");
+                }
+            }
+        });
         try {
             httpClient.start();
         } catch (Exception e) {

bundles/org.openhab.binding.linkplay/src/main/java/org/openhab/binding/linkplay/internal/client/upnp/LinkPlayUpnpClient.java

@@ -23,6 +23,7 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.eclipse.jdt.annotation.NonNullByDefault;
@@ -68,6 +69,7 @@ public class LinkPlayUpnpClient implements UpnpIOParticipant, LinkPlayUpnpComman
     private @Nullable RemoteDevice remoteDevice;
     private String udn = "";
     private volatile boolean disposed;
+    private volatile long lastValueNanos = System.nanoTime();
 
     public LinkPlayUpnpClient(LinkPlayUpnpClientHandler handler, UpnpIOService upnpIOService, UpnpService upnpService,
             ScheduledExecutorService scheduler) {
@@ -101,6 +103,15 @@ public class LinkPlayUpnpClient implements UpnpIOParticipant, LinkPlayUpnpComman
         return remoteDevice;
     }
 
+    /**
+     * Looks up the current {@link RemoteDevice} for this UDN directly from the jUPnP registry.
+     *
+     * @return the registered device, or {@code null} if the device is not currently in the registry
+     */
+    public @Nullable RemoteDevice findRegisteredDevice() {
+        return upnpService.getRegistry().getRemoteDevice(new UDN(udn), false);
+    }
+
     public boolean needsUpnpInitialization() {
         return needsUpnpInitialization.get();
     }
@@ -120,6 +131,7 @@ public class LinkPlayUpnpClient implements UpnpIOParticipant, LinkPlayUpnpComman
 
     @Override
     public void onValueReceived(@Nullable String variable, @Nullable String value, @Nullable String service) {
+        lastValueNanos = System.nanoTime();
         if (!handler.shouldProcessUpnpEvents()) {
             logger.debug("{}: onValueReceived: handler is not ready to process UPnP events", udn);
             return;
@@ -261,6 +273,22 @@ public class LinkPlayUpnpClient implements UpnpIOParticipant, LinkPlayUpnpComman
         subscriptionState.clear();
     }
 
+    /**
+     * Forces a resubscription by removing the current subscriptions and creating fresh ones. This
+     * works around jUPnP's renewal issues which LinkPlay devices frequently fail to honor
+     */
+    public void resubscribe() {
+        removeSubscriptions();
+        addSubscriptions();
+    }
+
+    /**
+     * @return the number of seconds since the last UPnP value (GENA event) was received
+     */
+    public long secondsSinceLastValue() {
+        return TimeUnit.NANOSECONDS.toSeconds(System.nanoTime() - lastValueNanos);
+    }
+
     public boolean isFullySubscribed() {
         for (String s : SERVICE_SUBSCRIPTIONS) {
             if (!subscriptionState.getOrDefault(s, false)) {

bundles/org.openhab.binding.linkplay/src/main/java/org/openhab/binding/linkplay/internal/handler/LinkPlayHandler.java

@@ -102,10 +102,15 @@ public class LinkPlayHandler extends BaseThingHandler
     private final Logger logger = LoggerFactory.getLogger(LinkPlayHandler.class);
 
     private static final int RECONNECT_DELAY = 30;
+    private static final int UPNP_WATCHDOG_INTERVAL_SECONDS = 30;
+    private static final int UPNP_RESUBSCRIBE_INTERVAL_SECONDS = 180;
+    private static final int UPNP_EVENT_GRACE_SECONDS = 60;
     private final HttpClient httpClient;
     private @Nullable ScheduledFuture<?> upnpServiceCheck;
+    private @Nullable ScheduledFuture<?> upnpWatchdogJob;
     private @Nullable ScheduledFuture<?> reconnectJob;
     private @Nullable ScheduledFuture<?> positionJob;
+    private volatile long lastResubscribeNanos = System.nanoTime();
     // Are we currently initializing the device?
     private final AtomicBoolean isInitializing = new AtomicBoolean(false);
     // Have we been disposed, prevent further calls to the device when shutting down
@@ -167,6 +172,11 @@ public class LinkPlayHandler extends BaseThingHandler
          */
         cancelUpnpServiceCheckJob();
         upnpServiceCheck = scheduler.scheduleWithFixedDelay(this::initializationCheck, 0, 15, TimeUnit.SECONDS);
+
+        // Watchdog that detects and recovers dead UPnP subscriptions.
+        cancelUpnpWatchdogJob();
+        upnpWatchdogJob = scheduler.scheduleWithFixedDelay(this::upnpEventWatchdog, UPNP_WATCHDOG_INTERVAL_SECONDS,
+                UPNP_WATCHDOG_INTERVAL_SECONDS, TimeUnit.SECONDS);
     }
 
     @Override
@@ -176,6 +186,7 @@ public class LinkPlayHandler extends BaseThingHandler
         linkPlayGroupService.unregisterParticipant(this);
         cancelReconnectJob();
         cancelUpnpServiceCheckJob();
+        cancelUpnpWatchdogJob();
         cancelPositionJob();
         notificationHandler.dispose();
         upnpClient.dispose();
@@ -581,12 +592,16 @@ public class LinkPlayHandler extends BaseThingHandler
             return;
         }
         if (!allSubscriptionsSuccessful) {
-            updateStatus(ThingStatus.UNKNOWN, ThingStatusDetail.NONE, "Waiting for UPnP subscriptions");
+            if (getThing().getStatus() != ThingStatus.ONLINE) {
+                updateStatus(ThingStatus.UNKNOWN, ThingStatusDetail.NONE, "Waiting for UPnP subscriptions");
+            }
             return;
         }
         if (getThing().getStatus() != ThingStatus.ONLINE) {
             updateStatus(ThingStatus.ONLINE);
         }
+        cancelReconnectJob();
+        lastResubscribeNanos = System.nanoTime();
     }
 
     @Override
@@ -814,15 +829,47 @@ public class LinkPlayHandler extends BaseThingHandler
 
     public void setOffline(String reason) {
         updateStatus(ThingStatus.OFFLINE, ThingStatusDetail.COMMUNICATION_ERROR, reason);
-        RemoteDevice device = upnpClient.getRemoteDevice();
         upnpClient.clearSubscriptionState();
-        if (device != null) {
-            cancelReconnectJob();
-            if (!disposed) {
-                reconnectJob = scheduler.schedule(() -> initFromUpnp(device), RECONNECT_DELAY, TimeUnit.SECONDS);
-            }
-        }
         upnpClient.sendDeviceSearchRequest();
+        scheduleReconnect();
+    }
+
+    private void scheduleReconnect() {
+        if (disposed) {
+            return;
+        }
+        ScheduledFuture<?> job = reconnectJob;
+        if (job != null && !job.isDone()) {
+            return;
+        }
+        reconnectJob = scheduler.scheduleWithFixedDelay(this::reconnect, RECONNECT_DELAY, RECONNECT_DELAY,
+                TimeUnit.SECONDS);
+    }
+
+    private void reconnect() {
+        if (disposed || getThing().getStatus() == ThingStatus.ONLINE) {
+            cancelReconnectJob();
+            return;
+        }
+        if (isInitializing.get()) {
+            return;
+        }
+        logger.debug("{}: Attempting to reconnect", udn);
+        upnpClient.sendDeviceSearchRequest();
+        Slave slave = linkPlayGroupService.findSlaveByUDN(udn);
+        if (slave != null) {
+            initFromGroup(slave);
+            return;
+        }
+        RemoteDevice device = upnpClient.findRegisteredDevice();
+        if (device == null) {
+            device = upnpClient.getRemoteDevice();
+        }
+        if (device != null) {
+            upnpClient.setRemoteDevice(device);
+            upnpClient.setNeedsUpnpInitialization(true);
+            initFromUpnp(device);
+        }
     }
 
     // Derives the host and port from the Upnp device and attempts to connect to the api
@@ -1407,6 +1454,35 @@ public class LinkPlayHandler extends BaseThingHandler
         }
     }
 
+    private void upnpEventWatchdog() {
+        if (disposed || getThing().getStatus() != ThingStatus.ONLINE) {
+            return;
+        }
+        // non-leader group members intentionally have UPnP turned off, so silence is expected
+        if (inGroup && !isLeader) {
+            return;
+        }
+        // don't tear down the subscription mid-notification as notifications are short-lived
+        if (notificationHandler.isInNotification()) {
+            return;
+        }
+        long staleSeconds = upnpClient.secondsSinceLastValue();
+        long sinceResubscribe = TimeUnit.NANOSECONDS.toSeconds(System.nanoTime() - lastResubscribeNanos);
+
+        if (staleSeconds > sinceResubscribe && sinceResubscribe >= UPNP_EVENT_GRACE_SECONDS) {
+            logger.debug("{}: No UPnP events {}s after resubscribe; reconnecting", udn, sinceResubscribe);
+            setOffline("No UPnP events received " + sinceResubscribe + "s after resubscribing");
+            return;
+        }
+
+        if (sinceResubscribe >= UPNP_RESUBSCRIBE_INTERVAL_SECONDS) {
+            logger.debug("{}: Proactively refreshing UPnP subscription ({}s since last event)", udn, staleSeconds);
+            upnpClient.resubscribe();
+            upnpClient.sendDeviceSearchRequest();
+            lastResubscribeNanos = System.nanoTime();
+        }
+    }
+
     private void cancelReconnectJob() {
         cancelJob(reconnectJob);
         reconnectJob = null;
@@ -1417,6 +1493,11 @@ public class LinkPlayHandler extends BaseThingHandler
         upnpServiceCheck = null;
     }
 
+    private void cancelUpnpWatchdogJob() {
+        cancelJob(upnpWatchdogJob);
+        upnpWatchdogJob = null;
+    }
+
     private void cancelJob(@Nullable ScheduledFuture<?> job) {
         if (job != null) {
             job.cancel(true);