1.2.0.RELEASE

build.gradle

@@ -17,7 +17,7 @@
  */
 
 ext {
-    projectVersion = '1.1.0.RELEASE'
+    projectVersion = '1.2.0.RELEASE'
     boonVersion = '0.5.7'
     boonGroup = "io.advantageous.boon"
     springFrameworkVersion = '4.2.5.RELEASE'

qbit/admin/src/main/java/io/advantageous/qbit/admin/ManagedServiceBuilder.java

@@ -23,6 +23,8 @@
 import io.advantageous.qbit.service.discovery.dns.DnsUtil;
 import io.advantageous.qbit.service.health.HealthServiceAsync;
 import io.advantageous.qbit.service.health.HealthServiceBuilder;
+import io.advantageous.qbit.service.health.ServiceHealthManager;
+import io.advantageous.qbit.service.impl.ServiceHealthManagerDefault;
 import io.advantageous.qbit.service.stats.StatsCollector;
 import io.advantageous.qbit.system.QBitSystemManager;
 
@@ -384,6 +386,41 @@ public ManagedServiceBuilder setPublicHost(String publicHost) {
         return this;
     }
 
+
+    /**
+     * Creates a ServiceHealthManager.
+     *
+     * @return ServiceHealthManager
+     */
+    public ServiceHealthManager createServiceHealthManager() {
+        return new ServiceHealthManagerDefault(null, null);
+    }
+
+
+    /**
+     * Creates a ServiceHealthManager.
+     *
+     * @param failCallback called if the service reports a failure.
+     * @return ServiceHealthManager
+     */
+    public ServiceHealthManager createServiceHealthManager(final Runnable failCallback) {
+        return new ServiceHealthManagerDefault(failCallback, null);
+    }
+
+
+    /**
+     * Creates a ServiceHealthManager.
+     *
+     * @param failCallback    called if the service reports a failure.
+     * @param recoverCallback recoverCallback called if the service recovers from a failure.
+     * @return ServiceHealthManager
+     */
+    public ServiceHealthManager createServiceHealthManager(final Runnable failCallback,
+                                                           final Runnable recoverCallback) {
+        return new ServiceHealthManagerDefault(failCallback, recoverCallback);
+    }
+
+
     /**
      * Get the public port for service meta generation (Swagger)
      *
@@ -419,9 +456,9 @@ public ManagedServiceBuilder setPublicPort(int publicPort) {
 
     /**
      * Get the actual port to bind to.
-     *
+     * <p>
      * Defaults to 8080.
-     *
+     * <p>
      * Looks for PORT under PORT_WEB, PORT0, PORT.
      *
      * @return actual http port to bind to.
@@ -534,7 +571,8 @@ public ManagedServiceBuilder setStatsDReplicatorBuilder(
         return this;
     }
 
-    /** Finds the admin port.
+    /**
+     * Finds the admin port.
      * Searches  under environment variables, QBIT_ADMIN_PORT, ADMIN_PORT, PORT1
      */
     public String findAdminPort() {

qbit/build.gradle

@@ -133,7 +133,7 @@ project('core') {
         compile "$boonGroup:boon-reflekt:$boonVersion"
         compile "$boonGroup:boon-json:$boonVersion"
         compile 'org.reactivestreams:reactive-streams:1.0.0'
-        compile 'io.advantageous.reakt:reakt:2.1.0.RELEASE'
+        compile 'io.advantageous.reakt:reakt:2.3.0.RELEASE'
         testCompile project(':qbit:test-support')
     }
 

qbit/core/src/main/java/io/advantageous/qbit/service/ServiceContext.java

@@ -29,7 +29,7 @@
 public class ServiceContext {
 
 
-    static final ServiceContext serviceContext = new ServiceContext();
+    protected static final ServiceContext serviceContext = new ServiceContext();
 
     public static ServiceContext serviceContext() {
         return serviceContext;

qbit/core/src/main/java/io/advantageous/qbit/service/health/ServiceHealthManager.java

@@ -0,0 +1,32 @@
+package io.advantageous.qbit.service.health;
+
+/**
+ * Allows service to notify health system of its status by accessing the low level ServiceQueue health.
+ */
+public interface ServiceHealthManager {
+
+    /**
+     * Checks to see if service is failing.
+     *
+     * @return true if failing
+     */
+    boolean isFailing();
+
+    /**
+     * Check to see if the service is ok.
+     *
+     * @return true if ok
+     */
+    boolean isOk();
+
+    /**
+     * Mark the service as failing.
+     */
+    void setFailing();
+
+    /**
+     * Mark the service as recovered.
+     */
+    void recover();
+
+}

qbit/core/src/main/java/io/advantageous/qbit/service/impl/BaseServiceQueueImpl.java

@@ -83,7 +83,7 @@
  * @author rhightower on 2/18/15.
  */
 public class BaseServiceQueueImpl implements ServiceQueue {
-    private static final ThreadLocal<ServiceQueue> serviceThreadLocal = new ThreadLocal<>();
+    protected static final ThreadLocal<ServiceQueue> serviceThreadLocal = new ThreadLocal<>();
     protected final QBitSystemManager systemManager;
     protected final Logger logger = LoggerFactory.getLogger(ServiceQueueImpl.class);
     protected final boolean debug = GlobalConstants.DEBUG && logger.isDebugEnabled();

qbit/core/src/main/java/io/advantageous/qbit/service/impl/ServiceHealthManagerDefault.java

@@ -0,0 +1,92 @@
+package io.advantageous.qbit.service.impl;
+
+import io.advantageous.qbit.service.ServiceContext;
+import io.advantageous.qbit.service.ServiceQueue;
+import io.advantageous.qbit.service.health.ServiceHealthManager;
+import io.advantageous.reakt.Expected;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static io.advantageous.reakt.Expected.empty;
+import static io.advantageous.reakt.Expected.ofNullable;
+
+public class ServiceHealthManagerDefault implements ServiceHealthManager {
+
+
+    final Logger logger = LoggerFactory.getLogger(ServiceHealthManagerDefault.class);
+    private final Expected<Runnable> failCallback;
+    private final Expected<Runnable> recoverCallback;
+    private Expected<ServiceQueue> serviceQueue = empty();
+    private Expected<ServiceContext> serviceContext = empty();
+
+    public ServiceHealthManagerDefault(final Runnable failCallback,
+                                       final Runnable recoverCallback) {
+        this.failCallback = Expected.ofNullable(failCallback);
+        this.recoverCallback = Expected.ofNullable(recoverCallback);
+    }
+
+
+    @Override
+    public boolean isFailing() {
+
+        serviceQueue.ifEmpty(this::loadIfEmpty);
+
+        if (serviceQueue.isPresent()) {
+            return serviceQueue.get().failing();
+        } else {
+            logger.warn("Service Queue was not found, but isFailing() was called on ServiceHealthManager");
+            return false;
+        }
+
+    }
+
+    private void loadIfEmpty() {
+        serviceContext.ifEmpty(() -> {
+            serviceContext = ofNullable(ServiceContext.serviceContext());
+            serviceContext.ifPresent(serviceContext1 -> serviceQueue =
+                    ofNullable(serviceContext1.currentService()));
+
+        });
+    }
+
+    @Override
+    public boolean isOk() {
+
+        serviceQueue.ifEmpty(this::loadIfEmpty);
+
+        if (serviceQueue.isPresent()) {
+            return !serviceQueue.get().failing();
+        } else {
+            logger.warn("Service Queue was not found, but isOk() was called on ServiceHealthManager");
+            return true;
+        }
+    }
+
+    @Override
+    public void setFailing() {
+
+        serviceQueue.ifEmpty(this::loadIfEmpty);
+
+
+        if (serviceQueue.isPresent()) {
+            serviceQueue.get().setFailing();
+        } else {
+            logger.warn("Service Queue was not found, but setFailing() was called on ServiceHealthManager");
+        }
+        failCallback.ifPresent(Runnable::run);
+    }
+
+    @Override
+    public void recover() {
+
+        serviceQueue.ifEmpty(this::loadIfEmpty);
+
+        if (serviceQueue.isPresent()) {
+            serviceQueue.get().recover();
+        } else {
+            logger.warn("Service Queue was not found, but recover() was called on ServiceHealthManager");
+        }
+        recoverCallback.ifPresent(Runnable::run);
+
+    }
+}

qbit/core/src/test/java/io/advantageous/qbit/service/impl/ServiceHealthManagerDefaultTest.java

@@ -0,0 +1,51 @@
+package io.advantageous.qbit.service.impl;
+
+import io.advantageous.qbit.service.ServiceBuilder;
+import org.junit.Test;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class ServiceHealthManagerDefaultTest {
+
+
+    @Test
+    public void testIsFailing() throws Exception {
+
+        BaseServiceQueueImpl.serviceThreadLocal.set(ServiceBuilder.serviceBuilder().setServiceObject(new Object()).build());
+
+        ServiceHealthManagerDefault serviceHealthManagerDefault = new ServiceHealthManagerDefault(null, null);
+
+        serviceHealthManagerDefault.setFailing();
+        assertTrue(serviceHealthManagerDefault.isFailing());
+        serviceHealthManagerDefault.recover();
+        assertFalse(serviceHealthManagerDefault.isFailing());
+
+
+        serviceHealthManagerDefault = new ServiceHealthManagerDefault(null, null);
+        BaseServiceQueueImpl.serviceThreadLocal.set(null);
+        serviceHealthManagerDefault.setFailing();
+        assertFalse(serviceHealthManagerDefault.isFailing());
+
+    }
+
+    @Test
+    public void testIsOk() throws Exception {
+        BaseServiceQueueImpl.serviceThreadLocal.set(ServiceBuilder.serviceBuilder().setServiceObject(new Object()).build());
+
+        ServiceHealthManagerDefault serviceHealthManagerDefault = new ServiceHealthManagerDefault(null, null);
+
+        serviceHealthManagerDefault.recover();
+        assertTrue(serviceHealthManagerDefault.isOk());
+        serviceHealthManagerDefault.setFailing();
+        assertFalse(serviceHealthManagerDefault.isOk());
+
+
+        serviceHealthManagerDefault = new ServiceHealthManagerDefault(null, null);
+        BaseServiceQueueImpl.serviceThreadLocal.set(null);
+        serviceHealthManagerDefault.recover();
+        assertTrue(serviceHealthManagerDefault.isOk());
+
+    }
+
+}