src/main/java/application/rest/v1/Watcher.java
// Returns null if autoRestart is true.
Returns null if autoRestart is true.
Returns null if autoRestart is true .
public static <T> Object start(final Handler<T> h, boolean autoRestart) {
        final Object LOCK = new Object();
        Thread t = new Thread(new Runnable() {
            private static final int HTTP_STATUS_CODE_GONE = 410;
            @Override
            public void run() {
                while (true) {
                    boolean gone = false;
                    try {
                        ApiClient client = KAppNavEndpoint.getApiClient();
                        ApiClient localclient = client.setHttpClient(client.getHttpClient().newBuilder()
                                                            .readTimeout(0, TimeUnit.SECONDS)
                                                            .build());
                        Configuration.setDefaultApiClient(localclient);                                    
                        final AtomicReference<String> resourceVersion = new AtomicReference<>();
                        long watchStartTime = 0L;
                        Watch<T> watch = null;
                        try {
                            List<T> list = h.listResources(client, resourceVersion);
                            if (Logger.isDebugEnabled()) {
                                Logger.log(getClass().getName(), "run", Logger.LogType.DEBUG, "Retrieved current list of resources for " +
                                        h.getClass().getName() + " at resourceVersion (" + resourceVersion.get() + ").");
                            }
                            list.forEach(v -> {
                                h.processResponse(client, "ADDED", v);
                            });
                            watchStartTime = System.currentTimeMillis();
                            if (Logger.isDebugEnabled()) {
                                Logger.log(getClass().getName(), "run", Logger.LogType.DEBUG, "Watch starting for " + h.getClass().getName() +
                                        " at resourceVersion (" + resourceVersion.get() + ").");
                            }
                            OUTER: while (true) {
                                watch = Watch.createWatch(
                                        client,
                                        h.createWatchCall(client, resourceVersion.get()),
                                        h.getWatchType());
                                for (Watch.Response<T> item : watch) {
                                    if (item.status != null || "ERROR".equals(item.type)) {
                                        if (item.status != null) {
                                            final Integer code = item.status.getCode();
                                            if (code != null) {
                                                final int codeValue = code.intValue();
                                                if (codeValue == HTTP_STATUS_CODE_GONE) {
                                                    gone = true;
                                                    if (Logger.isDebugEnabled()) {
                                                        Logger.log(getClass().getName(), "run", Logger.LogType.DEBUG,
                                                                "The resourceVersion (" + resourceVersion.get() +
                                                                ") is too old (status 410). The watch for " + h.getClass().getName() +
                                                                " will be restarted from the current resource version."); 
                                                    }
                                                }
                                                else if (Logger.isDebugEnabled()) {
                                                    Logger.log(getClass().getName(), "run", Logger.LogType.DEBUG,
                                                            "Status (" + codeValue + ") returned from the watch on resourceVersion (" +
                                                            resourceVersion.get() + "). The watch for " + h.getClass().getName() +
                                                            " will be restarted from the current resource version.");
                                                }
                                            }
                                        }
                                        else if (Logger.isDebugEnabled()) {
                                            Logger.log(getClass().getName(), "run", Logger.LogType.DEBUG,
                                                    "An unknown error was returned from the watch on resourceVersion (" +
                                                    resourceVersion.get() + "). The watch for " + h.getClass().getName() +
                                                    " will be restarted from the current resource version.");
                                        }
                                        break OUTER;
                                    }
                                    h.processResponse(client, item.type, item.object);
                                }
                                watch.close();
                                watch = null;
                                if (Logger.isDebugEnabled()) {
                                    Logger.log(getClass().getName(), "run", Logger.LogType.DEBUG, "Watch restarting for " + h.getClass().getName() +
                                            " at resourceVersion (" + resourceVersion.get() + ").");
                                }
                            }
                        }
                        catch (Exception e) {
                            if (Logger.isDebugEnabled()) {
                                Logger.log(getClass().getName(), "run", Logger.LogType.DEBUG, "Caught Exception from running watch: " + e.toString());
                            }
                        }
                        finally {
                            h.reset(client);
                            if (watch != null) {
                                watch.close();
                            }
                            if (Logger.isDebugEnabled() && watchStartTime > 0L) {
                                Logger.log(getClass().getName(), "run", Logger.LogType.DEBUG, "Watch completed for " + h.getClass().getName() +
                                        " at resourceVersion (" + resourceVersion.get() + ") after " + (System.currentTimeMillis() - watchStartTime) + " ms.");
                            }
                        }
                    }
                    catch (Exception e) {
                        if (Logger.isDebugEnabled()) {
                            Logger.log(getClass().getName(), "run", Logger.LogType.DEBUG, "Caught Exception from watch initialization or shutdown: " + e.toString());
                        }
                    }
                    if (!(gone || autoRestart)) {
                        synchronized (LOCK) {
                            try {
                                LOCK.wait();
                            }
                            catch (InterruptedException e) {
                                if (Logger.isDebugEnabled()) {
                                    Logger.log(getClass().getName(), "run", Logger.LogType.DEBUG, "Thread (" + h.getWatcherThreadName() + ") awakened.");
                                }
                            }
                        }
                    }
                }
            }
        });
        t.setName(h.getWatcherThreadName());
        t.setDaemon(true);
        t.start();
        return !autoRestart ? LOCK : null;
    }
public static < T > Object start ( final Handler < T > h , boolean autoRestart ) { final Object LOCK = new Object ( ) ; Thread t = new Thread ( new Runnable ( ) { private static final int HTTP_STATUS_CODE_GONE = 410 ; @ Override public void run ( ) { while ( true ) { boolean gone = false ; try { ApiClient client = KAppNavEndpoint . getApiClient ( ) ; ApiClient localclient = client . setHttpClient ( client . getHttpClient ( ) . newBuilder ( ) . readTimeout ( 0 , TimeUnit . SECONDS ) . build ( ) ) ; Configuration . setDefaultApiClient ( localclient ) ; final AtomicReference < String > resourceVersion = new AtomicReference < > ( ) ; long watchStartTime = 0L ; Watch < T > watch = null ; try { List < T > list = h . listResources ( client , resourceVersion ) ; if ( Logger . isDebugEnabled ( ) ) { Logger . log ( getClass ( ) . getName ( ) , "run" , Logger . LogType . DEBUG , "Retrieved current list of resources for " + h . getClass ( ) . getName ( ) + " at resourceVersion (" + resourceVersion . get ( ) + ")." ) ; } list . forEach ( v -> { h . processResponse ( client , "ADDED" , v ) ; } ) ; watchStartTime = System . currentTimeMillis ( ) ; if ( Logger . isDebugEnabled ( ) ) { Logger . log ( getClass ( ) . getName ( ) , "run" , Logger . LogType . DEBUG , "Watch starting for " + h . getClass ( ) . getName ( ) + " at resourceVersion (" + resourceVersion . get ( ) + ")." ) ; } OUTER : while ( true ) { watch = Watch . createWatch ( client , h . createWatchCall ( client , resourceVersion . get ( ) ) , h . getWatchType ( ) ) ; for ( Watch . Response < T > item : watch ) { if ( item . status != null || "ERROR" . equals ( item . type ) ) { if ( item . status != null ) { final Integer code = item . status . getCode ( ) ; if ( code != null ) { final int codeValue = code . intValue ( ) ; if ( codeValue == HTTP_STATUS_CODE_GONE ) { gone = true ; if ( Logger . isDebugEnabled ( ) ) { Logger . log ( getClass ( ) . getName ( ) , "run" , Logger . LogType . DEBUG , "The resourceVersion (" + resourceVersion . get ( ) + ") is too old (status 410). The watch for " + h . getClass ( ) . getName ( ) + " will be restarted from the current resource version." ) ; } } else if ( Logger . isDebugEnabled ( ) ) { Logger . log ( getClass ( ) . getName ( ) , "run" , Logger . LogType . DEBUG , "Status (" + codeValue + ") returned from the watch on resourceVersion (" + resourceVersion . get ( ) + "). The watch for " + h . getClass ( ) . getName ( ) + " will be restarted from the current resource version." ) ; } } } else if ( Logger . isDebugEnabled ( ) ) { Logger . log ( getClass ( ) . getName ( ) , "run" , Logger . LogType . DEBUG , "An unknown error was returned from the watch on resourceVersion (" + resourceVersion . get ( ) + "). The watch for " + h . getClass ( ) . getName ( ) + " will be restarted from the current resource version." ) ; } break OUTER ; } h . processResponse ( client , item . type , item . object ) ; } watch . close ( ) ; watch = null ; if ( Logger . isDebugEnabled ( ) ) { Logger . log ( getClass ( ) . getName ( ) , "run" , Logger . LogType . DEBUG , "Watch restarting for " + h . getClass ( ) . getName ( ) + " at resourceVersion (" + resourceVersion . get ( ) + ")." ) ; } } } catch ( Exception e ) { if ( Logger . isDebugEnabled ( ) ) { Logger . log ( getClass ( ) . getName ( ) , "run" , Logger . LogType . DEBUG , "Caught Exception from running watch: " + e . toString ( ) ) ; } } finally { h . reset ( client ) ; if ( watch != null ) { watch . close ( ) ; } if ( Logger . isDebugEnabled ( ) && watchStartTime > 0L ) { Logger . log ( getClass ( ) . getName ( ) , "run" , Logger . LogType . DEBUG , "Watch completed for " + h . getClass ( ) . getName ( ) + " at resourceVersion (" + resourceVersion . get ( ) + ") after " + ( System . currentTimeMillis ( ) - watchStartTime ) + " ms." ) ; } } } catch ( Exception e ) { if ( Logger . isDebugEnabled ( ) ) { Logger . log ( getClass ( ) . getName ( ) , "run" , Logger . LogType . DEBUG , "Caught Exception from watch initialization or shutdown: " + e . toString ( ) ) ; } } if ( ! ( gone || autoRestart ) ) { synchronized ( LOCK ) { try { LOCK . wait ( ) ; } catch ( InterruptedException e ) { if ( Logger . isDebugEnabled ( ) ) { Logger . log ( getClass ( ) . getName ( ) , "run" , Logger . LogType . DEBUG , "Thread (" + h . getWatcherThreadName ( ) + ") awakened." ) ; } } } } } } } ) ; t . setName ( h . getWatcherThreadName ( ) ) ; t . setDaemon ( true ) ; t . start ( ) ; return ! autoRestart ? LOCK : null ; }
Returns null if autoRestart is true.
Returns null if autoRestart is true .
// Synchronization lock used for waking up the watcher thread. // Indicates that the resource requested is no longer available and will not be available again. // Note: While the watch is active this iterator loop will block waiting for notifications of resource changes from the Kube API.  // Breaking out of the outer loop in order to restart the watch from the current resource version. // If the version of the resource being watched is gone or the creator of the watch requested an auto-restart, restart immediately. // Sleep until notified by another thread, then try to re-establish the watch.