PortForwardService { public Closeable forwardPortAsync(final Logger externalProcessLogger, final LabelSelector podSelector, final int remotePort, final int localPort) throws Fabric8ServiceException { final Lock monitor = new ReentrantLock(true); final Condition podChanged = monitor.newCondition(); final Pod[] nextForwardedPod = new Pod[1]; final Thread forwarderThread = new Thread() { @Override public void run() { Pod currentPod = null; Closeable currentPortForward = null; try { monitor.lock(); while (true) { if (podEquals(currentPod, nextForwardedPod[0])) { podChanged.await(); } else { Pod nextPod = nextForwardedPod[0]; try { monitor.unlock(); if (currentPortForward != null) { log.info("Closing port-forward from pod %s", KubernetesHelper.getName(currentPod)); currentPortForward.close(); currentPortForward = null; } if (nextPod != null) { log.info("Starting port-forward to pod %s", KubernetesHelper.getName(nextPod)); currentPortForward = forwardPortAsync(externalProcessLogger, KubernetesHelper.getName(nextPod), remotePort, localPort); } else { log.info("Waiting for a pod to become ready before starting port-forward"); } currentPod = nextPod; } finally { monitor.lock(); } } } } catch (InterruptedException e) { log.debug("Port-forwarding thread interrupted", e); Thread.currentThread().interrupt(); } catch (Exception e) { log.warn("Error while port-forwarding to pod", e); } finally { monitor.unlock(); if (currentPortForward != null) { try { currentPortForward.close(); } catch (Exception e) {} } } } }; Pod newPod = getNewestPod(podSelector); nextForwardedPod[0] = newPod; final Watch watch = KubernetesClientUtil.withSelector(kubernetes.pods(), podSelector, log).watch(new Watcher<Pod>() { @Override public void eventReceived(Action action, Pod pod) { monitor.lock(); try { List<Pod> candidatePods; if (nextForwardedPod[0] != null) { candidatePods = new LinkedList<>(); candidatePods.add(nextForwardedPod[0]); candidatePods.add(pod); } else { candidatePods = Collections.singletonList(pod); } Pod newPod = getNewestPod(candidatePods); if (!podEquals(nextForwardedPod[0], newPod)) { nextForwardedPod[0] = newPod; podChanged.signal(); } } finally { monitor.unlock(); } } @Override public void onClose(KubernetesClientException e) { } }); forwarderThread.start(); final Closeable handle = () -> { try { watch.close(); } catch (Exception e) {} try { forwarderThread.interrupt(); forwarderThread.join(15000); } catch (Exception e) {} }; Runtime.getRuntime().addShutdownHook(new Thread() { @Override public void run() { try { handle.close(); } catch (Exception e) { } } }); return handle; } PortForwardService(KubernetesClient kubernetes, Logger log); Closeable forwardPortAsync(final Logger externalProcessLogger, final LabelSelector podSelector, final int remotePort, final int localPort); void forwardPort(Logger externalProcessLogger, String pod, int remotePort, int localPort); ProcessUtil.ProcessExecutionContext forwardPortAsync(Logger externalProcessLogger, String pod, int remotePort, int localPort);  }
@Test public void testSimpleScenario() throws Exception { OpenShiftMockServer mockServer = new OpenShiftMockServer(false); Pod pod1 = new PodBuilder() .withNewMetadata() .withName("mypod") .addToLabels("mykey", "myvalue") .withResourceVersion("1") .endMetadata() .withNewStatus() .withPhase("run") .endStatus() .build(); PodList pods1 = new PodListBuilder() .withItems(pod1) .withNewMetadata() .withResourceVersion("1") .endMetadata() .build(); mockServer.expect().get().withPath("/api/v1/namespaces/test/pods?labelSelector=mykey%3Dmyvalue").andReturn(200, pods1).always(); mockServer.expect().get().withPath("/api/v1/namespaces/test/pods").andReturn(200, pods1).always(); mockServer.expect().get().withPath("/api/v1/namespaces/test/pods?labelSelector=mykey%3Dmyvalue&watch=true") .andUpgradeToWebSocket().open() .waitFor(1000) .andEmit(new WatchEvent(pod1, "MODIFIED")) .done().always(); mockServer.expect().get().withPath("/api/v1/namespaces/test/pods?resourceVersion=1&watch=true") .andUpgradeToWebSocket().open() .waitFor(1000) .andEmit(new WatchEvent(pod1, "MODIFIED")) .done().always(); OpenShiftClient client = mockServer.createOpenShiftClient(); PortForwardService service = new PortForwardService(client, logger) { @Override public ProcessUtil.ProcessExecutionContext forwardPortAsync(Logger externalProcessLogger, String pod, int remotePort, int localPort) throws Fabric8ServiceException { return new ProcessUtil.ProcessExecutionContext(process, Collections.<Thread>emptyList(), logger); } }; try (Closeable c = service.forwardPortAsync(logger, new LabelSelectorBuilder().withMatchLabels(Collections.singletonMap("mykey", "myvalue")).build(), 8080, 9000)) { Thread.sleep(3000); } }