full async service and add monitoring

doc/todo.txt

@@ -2,7 +2,7 @@
 # new services design
 +- counters and metrics in the reactor
 +- validateRequest in RestService , the jwt issues
-- Add monitoring apis
++- Add monitoring apis
 +- All the validation ,pre/post handling that was were done by the handler will be done by the base service
 - add Runtime Test:
     addTest(const char *testName, nsMicroservice_Iface::TestFunction testFunction);

src/main/java/microservice/MicroserviceApp.java

@@ -27,6 +27,7 @@ import java.util.function.Consumer;
 import java.util.stream.Stream;
 
 import static io.undertow.Handlers.resource;
+import static microservice.handlers.MonitorHandler.MON_PREFIX;
 
 /*
  * TO DO:
@@ -44,7 +45,7 @@ public class MicroserviceApp
 {
 	private static MicroserviceApp sInstance = null;
 
-	private static final String MON_PREFIX = "_mon";
+
 	RestServerParams rsiParams = null;
 	RMQClientParams mbiParams = null;
 	String appName;
@@ -159,8 +160,9 @@ public class MicroserviceApp
 	}
 
 	public MicroserviceApp withMonitoring() {
-		optMonitorHandler = Optional.of(new MonitorHandler());
-		this.addHandler(MON_PREFIX, optMonitorHandler.get());
+//		optMonitorHandler = Optional.of(new MonitorHandler());
+//		this.addHandler(MON_PREFIX, optMonitorHandler.get());
+		addMethodClass(new MonitorHandler());
 		return this;
 	}
 

src/main/java/microservice/handlers/MonitorHandler.java

@@ -3,7 +3,13 @@ package microservice.handlers;
 import java.util.LinkedList;
 import java.util.List;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.JsonNodeFactory;
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import microservice.MicroserviceApp;
+import microservice.defs.Enums;
+import microservice.io.iface.CommonServices;
+import microservice.io.iface.IConfiguration;
 import microservice.io.impl.IMetricsFactoryImpl;
 import microservice.types.BaseRestResponse;
 import microservice.RestContext;
@@ -13,12 +19,13 @@ import microservice.RestContext;
  * @author amir
  *
  */
-public class MonitorHandler extends BaseHandler
+public class MonitorHandler implements CommonServices.IMethodClass
 {
+	public static final String MON_PREFIX = "_mon";
 	private static final String RELOAD = "_reload";
 	public static final String STAT = "_stat";
 	List<BaseHandler> containers = null;
-
+	protected ObjectMapper objMapper = new ObjectMapper();
 
 
 
@@ -32,25 +39,25 @@ public class MonitorHandler extends BaseHandler
 		containers.add(handler);
 	}
 
-	@Override
-	public void doCreate(RestContext reqCtx)
-	{
-		switch(reqCtx.params[0])
-		{
-			case RELOAD:
-				reloadServiceApp(reqCtx);
-				break;
-			case STAT:
-				printStats(reqCtx);
-				break;
-		}
-
-	}
+//	@Override
+//	public void doCreate(RestContext reqCtx)
+//	{
+//		switch(reqCtx.params[0])
+//		{
+//			case RELOAD:
+//				reloadServiceApp(reqCtx);
+//				break;
+//			case STAT:
+//				printStats(reqCtx);
+//				break;
+//		}
+//
+//	}
 
 	private void printStats(RestContext reqCtx) {
 		BaseRestResponse brr = new BaseRestResponse(true, null);
 		IMetricsFactoryImpl factoryImpl = (IMetricsFactoryImpl) IMetricsFactoryImpl.getInstance();
-		ObjectNode objectNode = this.objMapper.createObjectNode();
+		ObjectNode objectNode = JsonNodeFactory.instance.objectNode();
 		try {
 			objectNode.put("counters",this.objMapper.valueToTree(factoryImpl.getMetrics().getCounters()));
 			objectNode.put("meters",this.objMapper.valueToTree(factoryImpl.getMetrics().getMeters()));
@@ -59,46 +66,65 @@ public class MonitorHandler extends BaseHandler
 			brr.setError(exp.toString());
 		}
 		brr.objectNode = objectNode;
-		writeObjectToResponse(reqCtx, brr);
+		reqCtx.container.writeObjectToResponse(reqCtx.response, brr);
 	}
 
 	private void reloadServiceApp(RestContext reqCtx) {
-
-		this.optConfiguration.ifPresent(cfg -> cfg.reload());
-		containers.forEach(handler -> handler.doReload());
-		writeObjectToResponse(reqCtx,new BaseRestResponse(true,null));
-	}
-
-	@Override
-	public void doRead(RestContext reqCtx)
-	{
-		switch(reqCtx.params[0])
-		{
-			case STAT:
-				printStats(reqCtx);
-				break;
+		final IConfiguration cfg = MicroserviceApp.getsInstance().getConfiguration();
+		if (cfg != null)
+			cfg.reload();
+		//containers.forEach(handler -> handler.doReload());
+		reqCtx.container.writeObjectToResponse(reqCtx.response,new BaseRestResponse(true,null));
 	}
 
-	}
+//	@Override
+//	public void doRead(RestContext reqCtx)
+//	{
+//		switch(reqCtx.params[0])
+//		{
+//			case STAT:
+//				printStats(reqCtx);
+//				break;
+//		}
+//
+//	}
+//
+//	@Override
+//	public void doUpdate(RestContext reqCtx)
+//	{
+//		// TODO Auto-generated method stub
+//
+//	}
+//
+//	@Override
+//	public void doDelete(RestContext reqCtx)
+//	{
+//		// TODO Auto-generated method stub
+//
+//	}
+//
+//	@Override
+//	public void init() {
+//		// TODO Auto-generated method stub
+////		getApp().getMSClient("service");
+//	}
 
 	@Override
-	public void doUpdate(RestContext reqCtx)
-	{
-		// TODO Auto-generated method stub
-
-	}
-
-	@Override
-	public void doDelete(RestContext reqCtx)
-	{
-		// TODO Auto-generated method stub
-
-	}
-
-	@Override
-	public void init() {
-		// TODO Auto-generated method stub
-//		getApp().getMSClient("service");
+	public void getMethods(List<CommonServices.MethodParams> methodParamsList) {
+		/**
+		 * adding methods
+		 */
+		methodParamsList.add(new CommonServices.MethodParams(Enums.EnumServiceType.E_REST,
+				CommonServices.EnumRestCommands.E_READ,
+				"/" + MON_PREFIX + "/" + RELOAD,
+				(msgCtx,orgService) -> {
+					reloadServiceApp((RestContext)msgCtx);
+		}));
+		methodParamsList.add(new CommonServices.MethodParams(Enums.EnumServiceType.E_REST,
+				CommonServices.EnumRestCommands.E_READ,
+				"/" + MON_PREFIX + "/" + STAT,
+				(msgCtx,orgService) -> {
+					printStats((RestContext)msgCtx);
+		}));
 	}
-
 }

src/main/java/microservice/io/iface/CommonServices.java

@@ -124,6 +124,10 @@ public class CommonServices {
 
         public EnumRestServiceMode getServiceMode() { return serviceMode;  }
         public void setServiceMode(EnumRestServiceMode serviceMode) { this.serviceMode = serviceMode; }
+        public void setRestClient(ICommandClient restClient) {
+            this.restClient = restClient;
+        }
+
         ///////////// SYNC //////////////////////////
         public abstract BaseRestResponse create(CommandParams cmdParams);
         public abstract BaseRestResponse read(CommandParams cmdParams);
@@ -135,9 +139,12 @@ public class CommonServices {
         public abstract boolean asyncUpdate(CommandParams reqCtx, Consumer<BaseRestResponse> cbFunc);
         public abstract boolean asyncDelete(CommandParams reqCtx, Consumer<BaseRestResponse> cbFunc);
 
-        public void setRestClient(ICommandClient restClient) {
-            this.restClient = restClient;
-        }
+        /// helpers
+        public abstract void sendErrorResp(IResponse response, String error);
+        public abstract void writeObjectToResponse(IResponse response, Object value);
+        public abstract Object readObjectFromRequest(IRequest request, Class<?> ObjClass);
+
+        public abstract void startAsync(IRequest request,Runnable asyncFunc);
     }
 
     enum EnumPubSubCommands implements IServiceCommands {

src/main/java/microservice/io/iface/IRequest.java

@@ -2,6 +2,7 @@ package microservice.io.iface;
 
 import java.io.InputStream;
 import java.util.List;
+import java.util.function.Consumer;
 
 public interface IRequest
 {
@@ -24,4 +25,7 @@ public interface IRequest
 	 * @return list of the header values
      */
 	List<String> getHeader(String headerName);
+
+	public boolean startAsync(Runnable asyncFunc);
+
 }

src/main/java/microservice/io/impl/IRequestMBIImpl.java

@@ -3,6 +3,7 @@ package microservice.io.impl;
 import java.io.ByteArrayInputStream;
 import java.io.InputStream;
 import java.util.List;
+import java.util.function.Consumer;
 
 //import com.sun.xml.internal.ws.api.message.stream.InputStreamMessage;
 
@@ -59,5 +60,9 @@ public class IRequestMBIImpl implements IRequest
 		return null;
 	}
 
+	@Override
+	public boolean startAsync(Runnable asyncFunc) {
+		return false;
+	}
 
 }

src/main/java/microservice/io/impl/IRequestRestImpl.java

@@ -2,6 +2,7 @@ package microservice.io.impl;
 
 import java.io.InputStream;
 import java.util.List;
+import java.util.function.Consumer;
 
 import io.undertow.server.HttpServerExchange;
 import microservice.io.iface.IRequest;
@@ -48,5 +49,14 @@ public class IRequestRestImpl implements IRequest
 		return exchange.getRequestHeaders().get(headerName);
 	}
 
-
+	@Override
+	public boolean startAsync(Runnable asyncFunc) {
+		if (exchange != null){
+			exchange.dispatch(() -> {
+				asyncFunc.run();
+			});
+			return true;
+		}
+		return false;
+	}
 }

src/main/java/microservice/io/impl/service/IRestServiceHttpImpl.java

@@ -456,4 +456,9 @@ public class IRestServiceHttpImpl extends CommonServices.IRestService implements
         return jwt;
     }
 
+    @Override
+    public void startAsync(IRequest request,Runnable asyncFunc){
+        request.startAsync(asyncFunc);
+    }
+
 }

src/test/java/microservice/TestMicroserviceApp.java

@@ -94,7 +94,8 @@ public class TestMicroserviceApp {
             if (restContext.pathParameters != null)
                 restContext.pathParameters.forEach((key,value) -> objectNode.put(key,value));
             brr.objectNode = objectNode;
-            restContext.container.writeObjectToResponse(restContext.response,brr);
+            final CommonServices.IRestService restService1 = (CommonServices.IRestService) orgService;
+            restService1.writeObjectToResponse(restContext.response,brr);
         })
         .addMethod(Enums.EnumServiceType.E_REST,CommonServices.EnumRestCommands.E_READ,"/registry/{query}",(msgCtx,orgService) -> {
             CommonServices.IRestService inRestService = (CommonServices.IRestService)MicroserviceApp.getsInstance().getService(Enums.EnumServiceType.E_REST,"undertowRestService");
@@ -102,15 +103,17 @@ public class TestMicroserviceApp {
             String query = restContext.getPathParameter("query");
             CommandParams cmdParams = new CommandParams().setEntity("172.16.1.132:5000").setParamsString("/v1/search").setRequestParams("q=" + query);
             BaseRestResponse brr = inRestService.read(cmdParams);
-            restContext.container.writeObjectToResponse(restContext.response,brr);
+            inRestService.writeObjectToResponse(restContext.response,brr);
         })
         .addMethod(Enums.EnumServiceType.E_REST,CommonServices.EnumRestCommands.E_READ,"/asyncRegistry/{query}",(msgCtx,orgService) -> {
             CommonServices.IRestService inRestService = (CommonServices.IRestService)MicroserviceApp.getsInstance().getService(Enums.EnumServiceType.E_REST,"undertowRestService");
             RestContext restContext = (RestContext)msgCtx;
             String query = restContext.getPathParameter("query");
             CommandParams cmdParams = new CommandParams().setEntity("172.16.1.132:5000").setParamsString("/v1/search").setRequestParams("q=" + query);
-            boolean retstat = inRestService.asyncRead(cmdParams, brr -> System.out.println(brr));
-            restContext.container.writeObjectToResponse(restContext.response,new BaseRestResponse(retstat,null));
+            ((IRestServiceHttpImpl)inRestService).startAsync(restContext.request,() -> {
+                boolean retstat = inRestService.asyncRead(cmdParams, brr -> inRestService.writeObjectToResponse(restContext.response,brr));
+            });
+//            restContext.container.writeObjectToResponse(restContext.response,new BaseRestResponse(retstat,null));
         })
         ._build()
         ._run();