Merge pull request #9 from chinmoy12c/enhancement/async-health-endpoints

Modified health checks to work asynchronously.

src/main/java/com/uci/dao/service/HealthService.java

@@ -1,107 +1,72 @@
 package com.uci.dao.service;
 
-import java.io.IOException;
-import java.util.Arrays;
+import java.util.function.Function;
 
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
-
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.JsonMappingException;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.uci.dao.repository.XMessageRepository;
 import com.uci.utils.UtilHealthService;
-
 import reactor.core.publisher.Mono;
 
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
 @Service
-public class HealthService extends UtilHealthService{
+public class HealthService extends UtilHealthService {
 	@Autowired
     private XMessageRepository xMessageRepository;
-
-
+
 	/**
-	 * Returns health json node for kafka, campaign url and cassandra
+	 * Returns health node
 	 * 
-	 * @return JsonNode
-	 * @throws JsonMappingException
-	 * @throws JsonProcessingException
-	 * @throws IOException
+	 * @return Mono<JsonNode>
 	 */
-	public JsonNode getAllHealthNode() throws JsonMappingException, JsonProcessingException, IOException {
-		ObjectMapper mapper = new ObjectMapper();
-        JsonNode jsonNode = mapper.readTree("{\"checks\":[{\"name\":\"kafka\",\"healthy\":false},{\"name\":\"campaign\",\"healthy\":false}],\"healthy\":true}");
-
-        /* Cassandra health info */
-        JsonNode cassandraHealthNode = getCassandraHealthNode();
-        JsonNode cassandraNode = mapper.createObjectNode();
-        ((ObjectNode) cassandraNode).put("name", "Cassandra");
-        ((ObjectNode) cassandraNode).put("healthy", cassandraHealthNode.get("healthy").asBoolean());
-
-        /* Kafka health info */
-        JsonNode kafkaHealthNode = getKafkaHealthNode();
-        JsonNode kafkaNode = mapper.createObjectNode();
-        ((ObjectNode) kafkaNode).put("name", "Kafka");
-        ((ObjectNode) kafkaNode).put("healthy", kafkaHealthNode.get("healthy").asBoolean());
-        ((ObjectNode) kafkaNode).put("details", kafkaHealthNode.get("details"));
-
-        /* create `ArrayNode` object */
-        ArrayNode arrayNode = mapper.createArrayNode();
-
-        /* add JSON users to array */
-        arrayNode.addAll(Arrays.asList(cassandraNode, kafkaNode));
-
-        ((ObjectNode) jsonNode).putArray("checks").addAll(arrayNode);
-
-        /* System overall health */
-        if(cassandraNode.get("healthy").booleanValue() && kafkaHealthNode.get("healthy").booleanValue()) {
-        	((ObjectNode) jsonNode).put("healthy", true);
-        } else {
-        	((ObjectNode) jsonNode).put("healthy", false);
-        }
-
-        return jsonNode;
+	public Mono<JsonNode> getCassandraHealthNode() {
+		return getIsCassandraHealthy().map(healthy -> new ObjectMapper().createObjectNode().put("healthy", healthy));
 	}
-
-
+
 	/**
-	 * Returns health node
-	 * 
-	 * @return
-	 * @throws JsonMappingException
-	 * @throws JsonProcessingException
+	 * Returns the combined health of kafka, campaign and cassandra.
+	 *
+	 * @return Returns details of each service in `checks` key and
+	 * the overall health status in `healthy` key.
 	 */
-	@SuppressWarnings("deprecation")
-	public JsonNode getCassandraHealthNode() throws IOException, JsonMappingException, JsonProcessingException {
-		Boolean cassandraHealth = getIsCassandraHealthy();
-
-		/* Result node */
-		ObjectMapper mapper = new ObjectMapper();
-        JsonNode jsonNode = mapper.readTree("{\"healthy\":false}");
-        /* Add data in result node */
-        ((ObjectNode) jsonNode).put("healthy", cassandraHealth);
-
-        return jsonNode;
+	@Override
+	public Mono<JsonNode> getAllHealthNode() {
+		ObjectNode resultNode = new ObjectMapper().createObjectNode();
+		Mono<JsonNode> externalComponentHealth = super.getAllHealthNode();
+		Mono<JsonNode> cassandraHealth = getCassandraHealthNode().map(result -> {
+			ObjectNode objectNode = new ObjectMapper().createObjectNode();
+			objectNode.put("name", "cassandra");
+			objectNode.set("healthy", result.get("healthy"));
+			return  objectNode;
+		});
+		return Mono.zip(externalComponentHealth, cassandraHealth).map(healths -> {
+			resultNode.putArray("checks")
+					.add(healths.getT1().get("checks"))
+					.add(healths.getT2());
+			resultNode.put("healthy",
+					healths.getT1().get("healthy").booleanValue() &&
+					healths.getT2().get("healthy").booleanValue()
+			);
+			return resultNode;
+		});
 	}
-	
+
 	/**
 	 * Check if Cassandra connecting or not
 	 * 
-	 * @return Boolean
+	 * @return Mono<Boolean>
 	 */
-	public Boolean getIsCassandraHealthy() {
-		Boolean isHealthy = false;
-		try {
-			Mono<Boolean> exists = xMessageRepository.existsByUserId("Test");
-	    	exists.subscribe(System.out::println);
-	    	isHealthy = true;
-		} catch(Exception e) {
-			isHealthy = false;
-		}
-
-		return isHealthy;
+	private Mono<Boolean> getIsCassandraHealthy() {
+		return Mono.fromCallable(() -> {
+			try {
+				xMessageRepository.existsByUserId("Test");
+				return true;
+			} catch(Exception e) {
+				return false;
+			}
+		});
 	}
 }