Commit 687b0d44 by amir

full async service and add monitoring

parent 71ca7d92
......@@ -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);
......
......@@ -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;
}
......
......@@ -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));
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 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
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);
}));
}
@Override
public void doDelete(RestContext reqCtx)
{
// TODO Auto-generated method stub
}
@Override
public void init() {
// TODO Auto-generated method stub
// getApp().getMSClient("service");
}
}
......@@ -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 {
......
......@@ -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);
}
......@@ -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;
}
}
......@@ -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;
}
}
......@@ -456,4 +456,9 @@ public class IRestServiceHttpImpl extends CommonServices.IRestService implements
return jwt;
}
@Override
public void startAsync(IRequest request,Runnable asyncFunc){
request.startAsync(asyncFunc);
}
}
......@@ -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();
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment