ZK: detect and destroy disconnected desktops

alternative-per-page-usage.zul

<zk>
    <!-- example of a per Page initialization -->
    <zscript><![CDATA[
        import zk.example.util.DesktopWatchDog;
        import java.util.concurrent.TimeUnit;
        // in a java Composer call this in doAfterCompose()
        // consider using MINUTES instead of SECONDS in a real live scenario
        DesktopWatchDog watchDog = new DesktopWatchDog(desktop, 2, 10, TimeUnit.SECONDS);
    ]]></zscript>
    <div>
        <button label="simulate long execution 5s " onClick="Thread.sleep(5000);"/>
        <button label="simulate long execution 20s" onClick="Thread.sleep(20000);"/>
        click to test how the alive check is skipped during long executions
        <separator/>
        <button label="cancel watchdog"
                onClick="watchDog.cancel(); watchDog.cancelClientSide();"/>
    </div>
</zk>

DesktopWatchDog.java

package zk.example.util;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zkoss.zk.au.AuRequest;
import org.zkoss.zk.au.AuService;
import org.zkoss.zk.ui.Desktop;
import org.zkoss.zk.ui.Execution;
import org.zkoss.zk.ui.Session;
import org.zkoss.zk.ui.event.Events;
import org.zkoss.zk.ui.sys.WebAppCtrl;
import org.zkoss.zk.ui.util.*;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

public class DesktopWatchDog {
	static Logger logger = LoggerFactory.getLogger(DesktopWatchDog.class);
	private static final String HEART_BEAT_FLAG = "heartBeat";
	//shared single thread scheduler ... (choose more if needed, which is unlikely)
	private static ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
	private static AtomicInteger clientIdSequence = new AtomicInteger(0);

	private Desktop desktop;
	private final AuService auService;
	private final DesktopCleanup desktopCleanup;
	private final ExecutionCleanup executionCleanup;
	private final ScheduledFuture<?> aliveCheck;
	private final String clientId;
	private final AtomicBoolean alive = new AtomicBoolean(true);

	public DesktopWatchDog(Desktop desktop, int heartBeatInterval, int aliveCheckInterval, TimeUnit unit) {
		this.desktop = desktop;
		this.clientId = "heartbeatIntervalId_" + clientIdSequence.getAndIncrement();

		aliveCheck = executorService.scheduleWithFixedDelay(() -> {
			//skip alive check if busy
			if (desktop.getExecution() != null) {
				logger.debug("skipping alive check: {}", desktop.getId());
				return;
			}
			//cancel aliveCheck, and destroy the dead desktop
			if (!alive.getAndSet(false)) {
				//no heart beat received within aliveCheckInterval ...
				logger.info("desktop is dead ... cleaning up: {}", desktop.getId());
				cancel();
				destroyDesktop(desktop);
			} else {
				logger.debug("desktop was alive: {}", desktop.getId());
			}
		}, aliveCheckInterval, aliveCheckInterval, unit);

		desktop.addListener(new ExecutionInit() {
			@Override
			public void init(Execution exec1, Execution parent1) throws Exception {
				//initialize JS in an execution init listener
				//(calling Clients.evalJavaScript directly is too early)
				Clients.evalJavaScript(String.format(
						"if( !zk.desktopWatchDogIds ) { zk.desktopWatchDogIds = {}; }" +
								"zk.desktopWatchDogIds['%s'] = setInterval(function() { " +
								"zAu.send(new zk.Event(zkdt(), '%s', {%s: true}, {ignorable: true, duplicateIgnore: {onTimer: true}})) " +
								"}, %d);",
						clientId, Events.ON_TIMER, HEART_BEAT_FLAG, unit.toMillis(heartBeatInterval)));
				//only once per desktop
				desktop.removeListener(this);
			}
		});

		auService = (AuRequest request, boolean everError) -> {
			if (Events.ON_TIMER.equals(request.getCommand())
					&& (Boolean) request.getData().getOrDefault(HEART_BEAT_FLAG, false)) {
				logger.debug("received heartbeat on desktop: {}", desktop.getId());
				return true;
			}
			return false;
		};

		executionCleanup = (exec, parent, err) -> {
			logger.debug("set alive after execution: {}", desktop.getId());
			if (parent == null) {
				alive.set(true);
			}
		};

		//cancel aliveCheck if desktop is gone regularly
		desktopCleanup = dt -> {
			logger.debug("desktop destroyed: {}", desktop.getId());
			cancel();
		};

		desktop.addListener(auService);
		desktop.addListener(executionCleanup);
		desktop.addListener(desktopCleanup);
	}

	public void cancel() {
		desktop.removeListener(auService);
		desktop.removeListener(executionCleanup);
		desktop.removeListener(desktopCleanup);
		aliveCheck.cancel(true);
	}

	public void cancelClientSide() {
		Clients.evalJavaScript(String.format("clearInterval(zk.desktopWatchDogIds['%s'])", this.clientId));
	}

	private void destroyDesktop(Desktop desktop) {
		//destroy async - potentially too expensive for the shared scheduler
		CompletableFuture.runAsync(() -> {
			Session session = desktop.getSession();
			if (session != null) {
				final WebAppCtrl webAppCtrl = (WebAppCtrl) session.getWebApp();
				webAppCtrl.getDesktopCache(session).removeDesktop(desktop);
			}
		});
	}
}

DesktopWatchDogInit.java

package zk.example.util;

import org.zkoss.zk.ui.Desktop;
import org.zkoss.zk.ui.util.DesktopInit;

import static java.util.concurrent.TimeUnit.MINUTES;

/**
 * {@link DesktopInit} starting the {@link DesktopWatchDog} globally.
 * Configured in zk.xml via <a href="https://www.zkoss.org/wiki/ZK%20Configuration%20Reference/zk.xml/The%20listener%20Element">&lt;listener>-element</a>
 */
public class DesktopWatchDogInit implements DesktopInit {
	@Override
	public void init(Desktop desktop, Object request) throws Exception {
		new DesktopWatchDog(desktop, 2, 10, MINUTES);
	}
}