Added health check option against subscriptions (#6)

Author: getorymckeag

java-client/src/main/java/energy/trolie/client/RequestSubscription.java

@@ -17,4 +17,11 @@ public interface RequestSubscription {
 	 * @return true if the subscription is running and has not been stopped
 	 */
 	boolean isSubscribed();
+
+	/**
+	 *
+	 * @return true if the subscription is currently healthy, meaning that the last poll
+	 * occurred successfully without error.
+	 */
+	boolean isHealthy();
 }

java-client/src/main/java/energy/trolie/client/impl/request/AbstractStreamingGet.java

@@ -45,10 +45,15 @@ public abstract class AbstractStreamingGet<T extends StreamingResponseReceiver>
 	int bufferSize;
 	ThreadPoolExecutor threadPoolExecutor;
 	Map<String, String> httpHeaders;
+	protected boolean lastRequestFailed = false;
 
 	protected JsonFactory jsonFactory;
 	protected T receiver;
-	
+
+	protected boolean didLastRequestFail() {
+		return lastRequestFailed;
+	}
+
 	protected abstract String getPath();
 	protected abstract String getContentType();
 
@@ -100,9 +105,11 @@ protected boolean handleResponse(ClassicHttpResponse response) {
 			try {
 				return threadPoolExecutor.submit(new HandlerExecutor(response.getEntity().getContent())).get();
 			} catch (IOException e) {
+				lastRequestFailed = true;
 				logger.error("I/O error initiating request",e);
 				receiver.error(new StreamingGetConnectionException(e));
 			} catch (Exception e) {
+				lastRequestFailed = true;
 				logger.error("Internal error handling response",e);
 				receiver.error(new SubscriberInternalException(e));
 			}
@@ -111,6 +118,7 @@ protected boolean handleResponse(ClassicHttpResponse response) {
 			return true;
 		} else {
 			String s = "Server responded with status code " + response.getCode();
+			lastRequestFailed = true;
 			logger.error(s);
 			receiver.error(new StreamingGetResponseException(s, response.getCode()));
 		}
@@ -131,14 +139,16 @@ protected HttpGet createRequest() throws URISyntaxException {
 
 	public void executeRequest() {
 		try {
-			
+			lastRequestFailed = false;
 			HttpGet get = createRequest();
 			httpClient.execute(host.getHost(), get, createResponseHandler());
 
 		} catch (IOException e) {
+			lastRequestFailed = true;
 			logger.error("I/O error initiating request",e);
 			receiver.error(new StreamingGetConnectionException(e));
 		} catch (Exception e) {
+			lastRequestFailed = true;
 			receiver.error(new SubscriberInternalException(e));
 		}
 	}

java-client/src/main/java/energy/trolie/client/impl/request/AbstractStreamingSubscribedGet.java

@@ -83,6 +83,11 @@ public boolean isActive() {
 		return active.get();
 	}
 
+	@Override
+	public boolean isHealthy() {
+		return !didLastRequestFail();
+	}
+
 	@Override
 	protected boolean handleResponse(ClassicHttpResponse response) {
 		boolean success = super.handleResponse(response);
@@ -91,6 +96,7 @@ protected boolean handleResponse(ClassicHttpResponse response) {
 			try {
 				eTagStore.putETag(getPath(), response.getHeader(HttpHeaders.ETAG).getValue());
 			} catch (ProtocolException e) {
+				lastRequestFailed = true;
 				logger.error("Error handling server response",e);
 				receiver.error(new StreamingGetHandlingException(e));
 				return false;

java-client/src/test/java/energy/trolie/client/TrolieClientIT.java

@@ -78,6 +78,7 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 @Slf4j
 @SuppressWarnings("unchecked")
@@ -543,6 +544,7 @@ public void setSubscription(RequestSubscription subscription) {
 
 			while (subscription.isSubscribed()) {
 				Thread.sleep(100);
+				assertTrue(subscription.isHealthy());
 			}
 
 			//we should have received 2 snapshots, 1 304 code and 1 500 code