Skip to content

ipgallery.common.java / microservice

Go to a project
Commit 706a3f14 by amira
Options

Merge branch 'develop' of… 

Merge branch 'develop' of https://municipalitybank.com/ipgallery.common.java/microservice into develop
parents 4821064f 99f80084
Expand all
Inline Side-by-side
Showing with 1173 additions and 468 deletions
README.md
### Microservice Framework in JAVA
## 2.3.1
- add retention policy and ttl for pulsar, move PulsarParams from common
## 2.3.0
- Seperate Services from main lib
- Split to common,clients,servicePubSub and app jars
-
## 2.2.0
- Add Pulsar PubSub Implementation
- Add Either as default for optional
- Divide src to common and main so that app's can use only common where needed
- Note on Pulsar topic: can't be nested so after the domain prefix change the / with @
e.g: rest path like /domain/apiFamiliy/api1 will be in pubsub /domain/apiFamily@api1
## 2.1.4
- connect with undertow 2.0.28
## 2.1.3
- add date time to ILoggerConsoleImpl
## 2.1.2
- add config to okhttp client read/connect timeouts
"microservice.http.client.read.timeout.seconds" (30)
"microservice.http.client.connect.timeout.seconds" (10)
## 2.1.1
- downgrade metrics influxdb to fit the reporter on iot jar
......@@ -27,13 +43,13 @@
- add ZeroMQ support for Rest
## 1.3.8: Add colorful logging
## 1.3.7: Add Pagination - PageBuilder
## 1.3.6: upgrade common-utils 1.2.0
## 1.3.6: upgrade common-common.microservice.utils 1.2.0
## 1.3.2:
- add getMD5Hash, catch unhandled exceptions, add command-params builder
## 1.3.1:
- add rabbotmq command client + change version to 1.3.1
## 1.3.0:
- add service authorization with jwt
- add microservice.service authorization with jwt
- env param override config file
## 1.2.5:
- support rabbit on server side
......@@ -58,10 +74,10 @@
- Add '/_stat' in monitoring to returns stats of metrics, more to come
- Change '/reload' to '/_reload'
## Version '1.1.0':
- Add resolveService in service discovery, used by http command-client to resolve the address
of the dest service
# Env for service discovery:
service.resolver.polling.interval (milli) default is 5000
- Add resolveService in microservice.service discovery, used by http command-client to resolve the address
of the dest microservice.service
# Env for microservice.service discovery:
microservice.service.resolver.polling.interval (milli) default is 5000
# Env for jwt:
jwt.token.in.authorization : true
jwt.salt : "12345678901234567890123456789012"
doc/pulsar.txt 0 → 100644
Running standalone
docker run -it -p 80:80 -p 8080:8080 -p 6650:6650 apachepulsar/pulsar-standalone
[curl]
- show tenant
curl -s http://localhost:8080/admin/v2/tenants/mcz
- show namespaces
curl -s http://localhost:8080/admin/v2/namespaces/mcx/publish
[pulsar-admin]
run "docker exec -it [pulsar-standalone-instance] bash"
and then execute admin commands:
https://pulsar.apache.org/docs/latest/admin-api/overview/
- create tenant: pulsar-admin tenants create mcx
- create namespace in tenant: pulsar-admin namespaces create mcx/namespace1
- set replication clusters (needed to have policies, otherwise we can't publish):
bin/pulsar-admin namespaces set-clusters mcx/namespace1 --clusters standalone
doc/todo.txt
......@@ -3,13 +3,13 @@
+- counters and metrics in the reactor
+- validateRequest in RestService , the jwt issues
+- Add monitoring apis
+- All the validation ,pre/post handling that was were done by the handler will be done by the base service
+- All the validation ,pre/post handling that was were done by the handler will be done by the base microservice.service
- add Runtime Test:
addTest(const char *testName, nsMicroservice_Iface::TestFunction testFunction);
addTest(nsMicroservice_Iface::ITest *p_testClass)
+- Add true async in http rest service, that we can send response after handleRequest ends.
+- Add true async in http rest microservice.service, that we can send response after handleRequest ends.
// exchange.dispatch(() -> {
// new Timer().schedule(new TimerTask() {
// public void run() {
......
gradlew
#!/usr/bin/env bash
#!/usr/bin/env sh
##############################################################################
##
......@@ -6,12 +6,30 @@
##
##############################################################################
# Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
DEFAULT_JVM_OPTS=""
# Attempt to set APP_HOME
# Resolve links: $0 may be a link
PRG="$0"
# Need this for relative symlinks.
while [ -h "$PRG" ] ; do
ls=`ls -ld "$PRG"`
link=`expr "$ls" : '.*-> \(.*\)$'`
if expr "$link" : '/.*' > /dev/null; then
PRG="$link"
else
PRG=`dirname "$PRG"`"/$link"
fi
done
SAVED="`pwd`"
cd "`dirname \"$PRG\"`/" >/dev/null
APP_HOME="`pwd -P`"
cd "$SAVED" >/dev/null
APP_NAME="Gradle"
APP_BASE_NAME=`basename "$0"`
# Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
DEFAULT_JVM_OPTS=""
# Use the maximum available, or set MAX_FD != -1 to use that value.
MAX_FD="maximum"
......@@ -30,6 +48,7 @@ die ( ) {
cygwin=false
msys=false
darwin=false
nonstop=false
case "`uname`" in
CYGWIN* )
cygwin=true
......@@ -40,26 +59,11 @@ case "`uname`" in
MINGW* )
msys=true
;;
NONSTOP* )
nonstop=true
;;
esac
# Attempt to set APP_HOME
# Resolve links: $0 may be a link
PRG="$0"
# Need this for relative symlinks.
while [ -h "$PRG" ] ; do
ls=`ls -ld "$PRG"`
link=`expr "$ls" : '.*-> \(.*\)$'`
if expr "$link" : '/.*' > /dev/null; then
PRG="$link"
else
PRG=`dirname "$PRG"`"/$link"
fi
done
SAVED="`pwd`"
cd "`dirname \"$PRG\"`/" >/dev/null
APP_HOME="`pwd -P`"
cd "$SAVED" >/dev/null
CLASSPATH=$APP_HOME/gradle/wrapper/gradle-wrapper.jar
# Determine the Java command to use to start the JVM.
......@@ -85,7 +89,7 @@ location of your Java installation."
fi
# Increase the maximum file descriptors if we can.
if [ "$cygwin" = "false" -a "$darwin" = "false" ] ; then
if [ "$cygwin" = "false" -a "$darwin" = "false" -a "$nonstop" = "false" ] ; then
MAX_FD_LIMIT=`ulimit -H -n`
if [ $? -eq 0 ] ; then
if [ "$MAX_FD" = "maximum" -o "$MAX_FD" = "max" ] ; then
......@@ -150,11 +154,19 @@ if $cygwin ; then
esac
fi
# Split up the JVM_OPTS And GRADLE_OPTS values into an array, following the shell quoting and substitution rules
function splitJvmOpts() {
JVM_OPTS=("$@")
# Escape application args
save ( ) {
for i do printf %s\\n "$i" | sed "s/'/'\\\\''/g;1s/^/'/;\$s/\$/' \\\\/" ; done
echo " "
}
eval splitJvmOpts $DEFAULT_JVM_OPTS $JAVA_OPTS $GRADLE_OPTS
JVM_OPTS[${#JVM_OPTS[*]}]="-Dorg.gradle.appname=$APP_BASE_NAME"
APP_ARGS=$(save "$@")
# Collect all arguments for the java command, following the shell quoting and substitution rules
eval set -- $DEFAULT_JVM_OPTS $JAVA_OPTS $GRADLE_OPTS "\"-Dorg.gradle.appname=$APP_BASE_NAME\"" -classpath "\"$CLASSPATH\"" org.gradle.wrapper.GradleWrapperMain "$APP_ARGS"
# by default we should be in the correct project dir, but when run from Finder on Mac, the cwd is wrong
if [ "$(uname)" = "Darwin" ] && [ "$HOME" = "$PWD" ]; then
cd "$(dirname "$0")"
fi
exec "$JAVACMD" "${JVM_OPTS[@]}" -classpath "$CLASSPATH" org.gradle.wrapper.GradleWrapperMain "$@"
exec "$JAVACMD" "$@"
gradlew.bat
......@@ -8,14 +8,14 @@
@rem Set local scope for the variables with windows NT shell
if "%OS%"=="Windows_NT" setlocal
@rem Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
set DEFAULT_JVM_OPTS=
set DIRNAME=%~dp0
if "%DIRNAME%" == "" set DIRNAME=.
set APP_BASE_NAME=%~n0
set APP_HOME=%DIRNAME%
@rem Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
set DEFAULT_JVM_OPTS=
@rem Find java.exe
if defined JAVA_HOME goto findJavaFromJavaHome
......@@ -46,10 +46,9 @@ echo location of your Java installation.
goto fail
:init
@rem Get command-line arguments, handling Windowz variants
@rem Get command-line arguments, handling Windows variants
if not "%OS%" == "Windows_NT" goto win9xME_args
if "%@eval[2+2]" == "4" goto 4NT_args
:win9xME_args
@rem Slurp the command line arguments.
......@@ -60,11 +59,6 @@ set _SKIP=2
if "x%~1" == "x" goto execute
set CMD_LINE_ARGS=%*
goto execute
:4NT_args
@rem Get arguments from the 4NT Shell from JP Software
set CMD_LINE_ARGS=%$
:execute
@rem Setup the command line
......
src/main/java/microservice/MicroserviceApp.java → src/main/java/microservice/app/microservice/MicroserviceApp.java
package microservice;
import common.microservice.io.iface.*;
import common.microservice.io.impl.ILoggerConsoleImpl;
import io.undertow.Handlers;
import io.undertow.Undertow;
import io.undertow.Undertow.Builder;
......@@ -7,14 +9,14 @@ import io.undertow.server.HttpHandler;
import io.undertow.server.handlers.PathHandler;
import io.undertow.server.handlers.resource.ClassPathResourceManager;
import io.undertow.util.MimeMappings;
import microservice.defs.Enums;
import common.microservice.defs.Enums;
import common.microservice.params.RMQClientParams;
import common.microservice.params.RestServerParams;
import microservice.handlers.*;
import microservice.io.iface.*;
import microservice.io.iface.IRestServer;
import microservice.io.impl.*;
import microservice.params.RMQClientParams;
import microservice.params.RestServerParams;
import microservice.services.CommonServices;
import common.microservice.services.CommonServices;
import org.eclipse.paho.client.mqttv3.MqttException;
import rabbitmq.common.RMQId;
......@@ -41,7 +43,7 @@ import static io.undertow.Handlers.resource;
* @author amir
*
*/
public class MicroserviceApp
public class MicroserviceApp implements IApp
{
private static MicroserviceApp sInstance = null;
......@@ -629,7 +631,7 @@ public class MicroserviceApp
.filter(p -> p != null)
.forEach(servicesMap ->
servicesMap.forEach((serviceKey, service) -> {
service.init();
service.init(this);
if (enableMetrics)
service.withMetricsFactory(IMetricsFactoryImpl.getInstance());
service.register(serviceDiscovery, id);
......@@ -643,6 +645,7 @@ public class MicroserviceApp
/**
* מי מתחכם זה מיקו!!!
* running only if there are any valid services
*/
if (Arrays.stream(servicesArray)
.filter(p -> p != null)
......@@ -689,10 +692,14 @@ public class MicroserviceApp
}
/**
* logging utils
*/
public void logRcid(String from, String rcid){
getLogger().info(from + " RCID: " + rcid);
}
// /**
// * logging utils
// */
// public void logRcid(String from, String rcid){
// getLogger().info(from + " RCID: " + rcid);
// }
//
// public void logMcid(String from, String mcid){
// getLogger().info(from + " MCID: " + mcid);
// }
}
src/main/java/microservice/MicroserviceClient.java → src/main/java/microservice/app/microservice/MicroserviceClient.java
package microservice;
import microservice.defs.Constants;
import microservice.types.BaseRestResponse;
import clients.microservice.impl.IRestClientHttpImpl;
import common.microservice.defs.Constants;
import common.CacheClient;
import common.JsonHandler;
import common.RedisCacheClient;
import com.fasterxml.jackson.databind.JsonNode;
import microservice.io.iface.IRestClient;
import microservice.io.impl.IRestClientHttpImpl;
import microservice.params.BaseClientParams;
import microservice.params.CommandParams;
import common.microservice.io.iface.IRestClient;
import common.microservice.types.BaseRestResponse;
import common.microservice.params.BaseClientParams;
import common.microservice.params.CommandParams;
/**
*
......
src/main/java/microservice/handlers/BaseHandler.java → src/main/java/microservice/app/microservice/handlers/BaseHandler.java
......@@ -6,11 +6,11 @@ import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import microservice.MicroserviceApp;
import microservice.common.context.RestContext;
import microservice.io.iface.IConfiguration;
import microservice.io.iface.IContainer;
import microservice.io.iface.ILogger;
import microservice.io.iface.INotifyCallback;
import common.microservice.context.RestContext;
import common.microservice.io.iface.IConfiguration;
import common.microservice.io.iface.IContainer;
import common.microservice.io.iface.ILogger;
import common.microservice.io.iface.INotifyCallback;
public abstract class BaseHandler
{
......
src/main/java/microservice/handlers/MBIHandler.java → src/main/java/microservice/app/microservice/handlers/MBIHandler.java
......@@ -5,16 +5,18 @@ import java.util.Deque;
import java.util.HashMap;
import java.util.Map;
import microservice.common.context.RestContext;
import microservice.defs.Constants;
import microservice.defs.Enums.EnumHttpMethod;
import microservice.io.iface.*;
import common.microservice.context.RestContext;
import common.microservice.defs.Constants;
import common.microservice.defs.Enums.EnumHttpMethod;
import common.microservice.io.iface.IContainer;
import common.microservice.io.iface.ILogger;
import common.microservice.io.iface.IRequest;
import common.microservice.io.iface.IResponse;
import common.microservice.types.BaseRestResponse;
import microservice.io.impl.IRequestMBIImpl;
import microservice.io.impl.IResponseMBIImpl;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import microservice.types.BaseRestResponse;
import rabbitmq.common.RMQId;
import rabbitmq.common.RMQMessage;
......
src/main/java/microservice/handlers/MonitorHandler.java → src/main/java/microservice/app/microservice/handlers/MonitorHandler.java
package microservice.handlers;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.SortedMap;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import com.codahale.metrics.Counter;
import com.codahale.metrics.Gauge;
......@@ -16,15 +14,15 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
import com.fasterxml.jackson.databind.node.ObjectNode;
import common.microservice.types.BaseRestResponse;
import microservice.MicroserviceApp;
import microservice.defs.Constants;
import microservice.defs.Enums;
import microservice.io.iface.ILogger;
import microservice.services.CommonServices;
import microservice.io.iface.IConfiguration;
import common.microservice.defs.Constants;
import common.microservice.defs.Enums;
import common.microservice.io.iface.ILogger;
import common.microservice.services.CommonServices;
import common.microservice.io.iface.IConfiguration;
import microservice.io.impl.IMetricsFactoryImpl;
import microservice.types.BaseRestResponse;
import microservice.common.context.RestContext;
import common.microservice.context.RestContext;
/**
* this class is for monitoring the microservice
......
src/main/java/microservice/handlers/Reactor.java → src/main/java/microservice/app/microservice/handlers/Reactor.java
......@@ -2,15 +2,17 @@ package microservice.handlers;
import io.undertow.util.PathMatcher;
import io.undertow.util.PathTemplateMatcher;
import microservice.defs.Constants;
import microservice.defs.Enums;
import microservice.services.CommonServices;
import microservice.io.iface.ILogger;
import microservice.io.iface.IMetricsFactory;
import common.microservice.defs.Constants;
import common.microservice.defs.Enums;
import common.microservice.services.CommonServices;
import common.microservice.io.iface.ILogger;
import common.microservice.io.iface.IMetricsFactory;
import java.util.*;
import java.util.function.BiConsumer;
import static common.microservice.services.CommonServices.buildServiceKey;
/**
* ------------
* Services -> | | -> Methods
......@@ -86,20 +88,20 @@ public class Reactor implements CommonServices.IServiceReactor {
methodKeyList.add(key);
}
public static String buildServiceKey(CommonServices.MethodParams methodParams) {
return Constants.SLASH_SEPERATOR +
methodParams.getServiceType().name() + Constants.TYPE_PREFIX_SEPERATOR +
methodParams.getServiceCommand().toString() + Constants.TYPE_PREFIX_SEPERATOR +
methodParams.getResourceUri();
}
public static String buildServiceKey(Enums.EnumServiceType enumServiceType,
CommonServices.IServiceCommands serviceCommands,
String resourceUri) {
return new StringBuilder(32).append(Constants.SLASH_SEPERATOR).append(enumServiceType.name()).append(Constants.TYPE_PREFIX_SEPERATOR)
.append(serviceCommands.toString()).append(Constants.TYPE_PREFIX_SEPERATOR)
.append(resourceUri).toString();
}
// public static String buildServiceKey(CommonServices.MethodParams methodParams) {
// return Constants.SLASH_SEPERATOR +
// methodParams.getServiceType().name() + Constants.TYPE_PREFIX_SEPERATOR +
// methodParams.getServiceCommand().toString() + Constants.TYPE_PREFIX_SEPERATOR +
// methodParams.getResourceUri();
// }
// public static String buildServiceKey(Enums.EnumServiceType enumServiceType,
// CommonServices.IServiceCommands serviceCommands,
// String resourceUri) {
// return new StringBuilder(32).append(Constants.SLASH_SEPERATOR).append(enumServiceType.name()).append(Constants.TYPE_PREFIX_SEPERATOR)
// .append(serviceCommands.toString()).append(Constants.TYPE_PREFIX_SEPERATOR)
// .append(resourceUri).toString();
// }
/**
* delegating the msg/request from the service to the method if any
......
src/main/java/microservice/handlers/RestHandler.java → src/main/java/microservice/app/microservice/handlers/RestHandler.java
package microservice.handlers;
import com.eaio.uuid.UUID;
import com.fasterxml.jackson.databind.ObjectMapper;
import common.microservice.io.iface.IContainer;
import common.microservice.io.iface.IPubSub;
import common.microservice.io.iface.IRequest;
import common.microservice.io.iface.IResponse;
import common.microservice.types.BaseRestResponse;
import common.microservice.types.UserProfile;
import common.microservice.utils.IDGenerator;
import io.jsonwebtoken.Claims;
import io.jsonwebtoken.ExpiredJwtException;
import io.undertow.server.HttpHandler;
......@@ -10,21 +17,18 @@ import io.undertow.util.HeaderMap;
import io.undertow.util.Headers;
import io.undertow.util.HttpString;
import microservice.MicroserviceApp;
import microservice.common.context.RestContext;
import microservice.common.EncryptionUtils;
import microservice.defs.Constants;
import microservice.defs.Enums.EnumHttpMethod;
import microservice.io.iface.*;
import microservice.io.iface.IMetricsFactory.IMeter;
import microservice.io.iface.IMetricsFactory.ITimer;
import common.microservice.context.RestContext;
import common.microservice.security.EncryptionUtils;
import common.microservice.defs.Constants;
import common.microservice.defs.Enums.EnumHttpMethod;
import common.microservice.io.iface.IMetricsFactory.IMeter;
import common.microservice.io.iface.IMetricsFactory.ITimer;
import microservice.io.impl.IMetricsFactoryImpl;
import microservice.io.impl.IRequestRestImpl;
import microservice.io.impl.IResponseRestImpl;
import microservice.services.CommonServices;
import microservice.types.BaseRestResponse;
import microservice.types.UserProfile;
import common.microservice.services.CommonServices;
import static microservice.defs.Constants.*;
import static common.microservice.defs.Constants.*;
public class RestHandler implements HttpHandler , IContainer
{
......@@ -283,7 +287,7 @@ public class RestHandler implements HttpHandler , IContainer
HeaderMap requestHeaders = exchange.getRequestHeaders();
reqCtx.rcid = requestHeaders.getFirst(Constants.RCID_HEADER);
if (reqCtx.rcid == null) // create a new one
reqCtx.rcid = new UUID().toString();
reqCtx.rcid = IDGenerator.createUUID();
return reqCtx;
}
......
src/main/java/microservice/io/iface/IRestServer.java → src/main/java/microservice/app/microservice/io/iface/IRestServer.java
......@@ -2,6 +2,9 @@ package microservice.io.iface;
import java.util.Map;
import common.microservice.io.iface.IPubSub;
import common.microservice.io.iface.IServiceDiscovery;
import microservice.handlers.BaseHandler;
/**
......@@ -13,7 +16,7 @@ public interface IRestServer {
public boolean build(String appName,
Map<String, BaseHandler> msHandlersMap,
IPubSub pubSub,
IPubSub pubSub,
boolean withMetrics);
public void run();
public void stop();
......
src/main/java/microservice/io/impl/IConfigurationConfigPropImpl.java → src/main/java/microservice/app/microservice/io/impl/IConfigurationConfigPropImpl.java
......@@ -3,7 +3,7 @@ package microservice.io.impl;
import common.configuration.ConfigProperties;
import common.configuration.EnumPropertyType;
import common.configuration.IConfigurationProvider;
import microservice.io.iface.IConfiguration;
import common.microservice.io.iface.IConfiguration;
public class IConfigurationConfigPropImpl implements IConfiguration {
ConfigProperties instance = null;
......
src/main/java/microservice/io/impl/ILogger4jImpl.java → src/main/java/microservice/app/microservice/io/impl/ILogger4jImpl.java
package microservice.io.impl;
import common.Log4jHandler;
import microservice.io.iface.ILogger;
import common.microservice.io.iface.ILogger;
public class ILogger4jImpl implements ILogger
{
......
src/main/java/microservice/io/impl/IMetricsFactoryImpl.java → src/main/java/microservice/app/microservice/io/impl/IMetricsFactoryImpl.java
......@@ -10,7 +10,7 @@ import metrics_influxdb.InfluxdbReporter;
import metrics_influxdb.api.measurements.CategoriesMetricMeasurementTransformer;
import metrics_influxdb.api.protocols.HttpInfluxdbProtocol;
import microservice.MicroserviceApp;
import microservice.io.iface.IMetricsFactory;
import common.microservice.io.iface.IMetricsFactory;
import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;
......@@ -167,7 +167,7 @@ public class IMetricsFactoryImpl implements IMetricsFactory
.convertDurationsTo(TimeUnit.MILLISECONDS)
.filter(MetricFilter.ALL)
.skipIdleMetrics(true)
.tag("service", MicroserviceApp.getsInstance().getAppName())
.tag("microservice/service/microservice/microservice", MicroserviceApp.getsInstance().getAppName())
.tag("instance", MicroserviceApp.getsInstance().getId())
.tag("server", MicroserviceApp.getsInstance().getServerName())
.transformer(new CategoriesMetricMeasurementTransformer("module", "artifact"))
......
src/main/java/microservice/io/impl/IPubSubMQTTImpl.java → src/main/java/microservice/app/microservice/io/impl/IPubSubMQTTImpl.java
......@@ -9,8 +9,8 @@ import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
import com.fasterxml.jackson.databind.JsonNode;
import microservice.io.iface.INotifyCallback;
import microservice.io.iface.IPubSub;
import common.microservice.io.iface.INotifyCallback;
import common.microservice.io.iface.IPubSub;
import common.JsonHandler;
public class IPubSubMQTTImpl implements IPubSub{
......
src/main/java/microservice/io/impl/IPubSubRedisImpl.java → src/main/java/microservice/app/microservice/io/impl/IPubSubRedisImpl.java
......@@ -10,8 +10,8 @@ import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisPubSub;
import common.JsonHandler;
import microservice.io.iface.INotifyCallback;
import microservice.io.iface.IPubSub;
import common.microservice.io.iface.INotifyCallback;
import common.microservice.io.iface.IPubSub;
public class IPubSubRedisImpl implements IPubSub
{
......
src/main/java/microservice/io/impl/IRMQClientRestImpl.java → src/main/java/microservice/app/microservice/io/impl/IRMQClientRestImpl.java
......@@ -4,14 +4,15 @@ import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.netflix.hystrix.contrib.metrics.eventstream.HystrixMetricsPoller;
import common.microservice.io.iface.ILogger;
import http.simpleRestClient.SimpleRestResponse;
import microservice.defs.Enums;
import microservice.io.iface.IRestClient;
import microservice.io.iface.IServiceDiscovery;
import microservice.params.BaseClientParams;
import microservice.params.CommandParams;
import microservice.params.RMQClientParams;
import microservice.types.BaseRestResponse;
import common.microservice.defs.Enums;
import common.microservice.io.iface.IRestClient;
import common.microservice.io.iface.IServiceDiscovery;
import common.microservice.params.BaseClientParams;
import common.microservice.params.CommandParams;
import common.microservice.params.RMQClientParams;
import common.microservice.types.BaseRestResponse;
import rabbitmq.client.RMQRestClient;
import rabbitmq.common.RMQId;
import rabbitmq.common.RMQMessage;
......@@ -28,6 +29,7 @@ import java.util.function.Consumer;
public class IRMQClientRestImpl implements IRestClient
{
private static int REQUEST_TIMEOUT = 30000;
private ILogger logger;
/*********************************************************************************************/
/* JSON LISTENER
......@@ -112,6 +114,11 @@ public class IRMQClientRestImpl implements IRestClient
}
@Override
public void init(ILogger logger) {
this.logger = logger;
}
@Override
public BaseRestResponse create(CommandParams reqCtx)
{
BaseRestResponse brr = null;
......
src/main/java/microservice/io/impl/IRequestMBIImpl.java → src/main/java/microservice/app/microservice/io/impl/IRequestMBIImpl.java
......@@ -4,14 +4,13 @@ import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
//import com.sun.xml.internal.ws.api.message.stream.InputStreamMessage;
import rabbitmq.common.RMQMessage;
import rabbitmq.common.RMQRestRequest;
//import io.undertow.server.HttpServerExchange;
import microservice.io.iface.IRequest;
import common.microservice.io.iface.IRequest;
public class IRequestMBIImpl implements IRequest
{
......
src/main/java/microservice/io/impl/IRequestRestImpl.java → src/main/java/microservice/app/microservice/io/impl/IRequestRestImpl.java
......@@ -2,15 +2,13 @@ package microservice.io.impl;
import java.io.InputStream;
import java.util.Arrays;
import java.util.Deque;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
import io.undertow.server.HttpServerExchange;
import io.undertow.util.HeaderValues;
import microservice.io.iface.IRequest;
import common.microservice.io.iface.IRequest;
public class IRequestRestImpl implements IRequest
{
......
src/main/java/microservice/io/impl/IResponseMBIImpl.java → src/main/java/microservice/app/microservice/io/impl/IResponseMBIImpl.java
package microservice.io.impl;
import java.io.IOException;
import java.nio.ByteBuffer;
import microservice.io.iface.ILogger;
import common.microservice.io.iface.ILogger;
import rabbitmq.common.RMQSender;
import rabbitmq.common.RMQId;
import rabbitmq.common.RMQMessage;
import rabbitmq.common.RMQRestResponse;
import microservice.io.iface.IResponse;
import common.microservice.io.iface.IResponse;
public class IResponseMBIImpl implements IResponse
{
......
src/main/java/microservice/io/impl/IResponseRestImpl.java → src/main/java/microservice/app/microservice/io/impl/IResponseRestImpl.java
......@@ -4,7 +4,7 @@ import io.undertow.server.HttpServerExchange;
import java.nio.ByteBuffer;
import microservice.io.iface.IResponse;
import common.microservice.io.iface.IResponse;
public class IResponseRestImpl implements IResponse
{
......
src/main/java/microservice/io/impl/IRestServerRMQImpl.java → src/main/java/microservice/app/microservice/io/impl/IRestServerRMQImpl.java
......@@ -10,10 +10,10 @@ import rabbitmq.server.RMQHandler;
import rabbitmq.server.RMQServer;
import microservice.handlers.BaseHandler;
import microservice.handlers.MBIHandler;
import microservice.io.iface.IPubSub;
import common.microservice.io.iface.IPubSub;
import microservice.io.iface.IRestServer;
import microservice.io.iface.IServiceDiscovery;
import microservice.params.RMQClientParams;
import common.microservice.io.iface.IServiceDiscovery;
import common.microservice.params.RMQClientParams;
public class IRestServerRMQImpl implements IRestServer {
......
src/main/java/microservice/io/impl/IRestServerUndertowImpl.java → src/main/java/microservice/app/microservice/io/impl/IRestServerUndertowImpl.java
......@@ -16,10 +16,10 @@ import common.Network;
import microservice.MicroserviceApp;
import microservice.handlers.BaseHandler;
import microservice.handlers.RestHandler;
import microservice.io.iface.IPubSub;
import common.microservice.io.iface.IPubSub;
import microservice.io.iface.IRestServer;
import microservice.io.iface.IServiceDiscovery;
import microservice.params.RestServerParams;
import common.microservice.io.iface.IServiceDiscovery;
import common.microservice.params.RestServerParams;
import static io.undertow.Handlers.resource;
public class IRestServerUndertowImpl implements IRestServer {
......
src/main/java/microservice/io/impl/IServiceDiscoveryConsulImpl.java → src/main/java/microservice/app/microservice/io/impl/IServiceDiscoveryConsulImpl.java
......@@ -10,8 +10,8 @@ import com.google.common.collect.Lists;
import common.Network;
import common.configuration.IConfigurationProvider;
import microservice.MicroserviceApp;
import microservice.defs.Enums;
import microservice.io.iface.IServiceDiscovery;
import common.microservice.defs.Enums;
import common.microservice.io.iface.IServiceDiscovery;
import java.util.*;
import java.util.Base64.Decoder;
......
src/main/java/microservice/services/IPubSubServiceMqttImpl.java → src/main/java/microservice/app/microservice/service/pubsub/impl/IPubSubServiceMqttImpl.java
package microservice.services;
package microservice.service.pubsub.impl;
import microservice.services.CommonServices;
import microservice.io.iface.IServiceDiscovery;
import common.microservice.io.iface.IApp;
import common.microservice.services.CommonServices;
import common.microservice.io.iface.IServiceDiscovery;
import org.eclipse.paho.client.mqttv3.*;
import java.util.HashMap;
......@@ -46,7 +47,7 @@ public class IPubSubServiceMqttImpl extends CommonServices.IPubSubService {
}
@Override
public boolean init() {
public boolean init(IApp app) {
try {
......
src/main/java/microservice/services/IRestServiceHttpImpl.java → src/main/java/microservice/app/microservice/service/pubsub/impl/IRestServiceHttpImpl.java
package microservice.services;
package microservice.service.pubsub.impl;
import com.eaio.uuid.UUID;
import com.fasterxml.jackson.databind.ObjectMapper;
import common.Network;
import common.microservice.io.iface.*;
import common.microservice.services.CommonServices;
import common.microservice.types.BaseRestResponse;
import common.microservice.types.UserProfile;
import common.microservice.utils.IDGenerator;
import common.microservice.utils.Logging;
import io.jsonwebtoken.Claims;
import io.jsonwebtoken.ExpiredJwtException;
import io.undertow.Handlers;
......@@ -16,37 +22,33 @@ import io.undertow.util.Headers;
import io.undertow.util.HttpString;
import io.undertow.util.MimeMappings;
import microservice.MicroserviceApp;
import microservice.common.context.RestContext;
import microservice.common.EncryptionUtils;
import microservice.defs.Constants;
import microservice.defs.Enums;
import common.microservice.context.RestContext;
import common.microservice.security.EncryptionUtils;
import common.microservice.defs.Constants;
import common.microservice.defs.Enums;
import microservice.handlers.BaseHandler;
import microservice.handlers.Reactor;
import microservice.io.iface.*;
import microservice.io.impl.IRequestRestImpl;
import microservice.io.impl.IResponseRestImpl;
import microservice.params.CommandParams;
import microservice.params.RestServerParams;
import microservice.types.BaseRestResponse;
import microservice.types.UserProfile;
import common.microservice.params.CommandParams;
import common.microservice.params.RestServerParams;
import java.util.function.BooleanSupplier;
import java.util.function.Consumer;
import java.util.function.Supplier;
import static common.microservice.services.CommonServices.buildServiceKey;
import static io.undertow.Handlers.resource;
import static microservice.defs.Constants.*;
import static common.microservice.defs.Constants.*;
/**
* Created by amir on 08/05/17.
*/
public class IRestServiceHttpImpl extends CommonServices.IRestService implements HttpHandler , IContainer{
public class IRestServiceHttpImpl extends CommonServices.IRestService implements HttpHandler , IContainer {
ILogger logger = null;
RestServerParams restServerParams = null;
Undertow restServer = null;
Thread restThread = null;
MicroserviceApp msAppInstance = null;
private String appName;
public ObjectMapper objMapper = null;
protected Enums.EnumAuthenticationType authType = Enums.EnumAuthenticationType.DEFAULT;
......@@ -78,7 +80,7 @@ public class IRestServiceHttpImpl extends CommonServices.IRestService implements
try {
if (restClient != null) {
if (cmdParams.getRcid() != null) // log rcid
msAppInstance.logRcid("handleSyncRespCommand." + name,cmdParams.getRcid());
Logging.logRcid(logger,"handleSyncRespCommand." + name,cmdParams.getRcid());
resp = command.get();
};
} catch (Exception e) {
......@@ -94,7 +96,7 @@ public class IRestServiceHttpImpl extends CommonServices.IRestService implements
boolean retstat;
try {
if (cmdParams.getRcid() != null) // log rcid
msAppInstance.logRcid("handleAsyncRespCommand." + name,cmdParams.getRcid());
Logging.logRcid(logger,"handleAsyncRespCommand." + name,cmdParams.getRcid());
retstat = command.getAsBoolean();
} catch (Exception e) {
retstat = false;
......@@ -143,19 +145,21 @@ public class IRestServiceHttpImpl extends CommonServices.IRestService implements
}
@Override
public boolean init() {
public boolean init(IApp app) {
boolean stat = true;
if (MicroserviceApp.getsInstance() != null) {
msAppInstance = MicroserviceApp.getsInstance();
logger = msAppInstance.getLogger();
this.appName = msAppInstance.getAppName();
this.app = app;
if (app != null) {
logger = app.getLogger();
this.appName = app.getAppName();
}
switch (getServiceMode()){
case E_CLIENT:
initRestClient();
break;
case E_CLIENT_SERVER:
case E_SERVER:
initRestClient();
stat = buildRestServer();
break;
default:
......@@ -166,6 +170,12 @@ public class IRestServiceHttpImpl extends CommonServices.IRestService implements
return stat;
}
private void initRestClient() {
if (restClient != null) {
restClient.init(logger);
}
}
private boolean buildRestServer() {
String host = this.restServerParams.getHost();
if (host == null || Network.LOCALHOST.equals(host))
......@@ -233,7 +243,7 @@ public class IRestServiceHttpImpl extends CommonServices.IRestService implements
/**
* finally we can delegate
*/
String key = Reactor.buildServiceKey(Enums.EnumServiceType.E_REST,reqContext.enumRestCommands,reqContext.request.getRelativePath());
String key = buildServiceKey(Enums.EnumServiceType.E_REST,reqContext.enumRestCommands,reqContext.request.getRelativePath());
reactor.delegate(this, key ,reqContext);
}
} else {
......@@ -285,9 +295,9 @@ public class IRestServiceHttpImpl extends CommonServices.IRestService implements
HeaderMap requestHeaders = exchange.getRequestHeaders();
reqCtx.rcid = requestHeaders.getFirst(Constants.RCID_HEADER);
if (reqCtx.rcid == null) // create a new one
reqCtx.rcid = new UUID().toString();
reqCtx.rcid = IDGenerator.createUUID();
else // log it
msAppInstance.logRcid("getRequestContext",reqCtx.rcid);
Logging.logRcid(logger,"getRequestContext",reqCtx.rcid);
return reqCtx;
}
......@@ -343,7 +353,7 @@ public class IRestServiceHttpImpl extends CommonServices.IRestService implements
apiDocMimeMappingsBuilder.addMapping("yaml", "text/x-yaml");
apiDocMimeMappingsBuilder.addMapping("html", "text/html");
pathHandler.addPrefixPath("/static", resource(new ClassPathResourceManager(getClass().getClassLoader(),getClass().getPackage()))//"static"))
.addWelcomeFiles("microservice/services/metrics.html")
.addWelcomeFiles("microservice/service/pubsub/metrics.html")
.setMimeMappings(apiDocMimeMappingsBuilder.build())
.setDirectoryListingEnabled(false));
}
......@@ -427,7 +437,7 @@ public class IRestServiceHttpImpl extends CommonServices.IRestService implements
}
else
{
msAppInstance.getLogger().error(NO_TOKEN_FOR_REQUEST);
logger.error(NO_TOKEN_FOR_REQUEST);
sendErrorResp(restContext.response,NO_TOKEN_FOR_REQUEST);
valid = false;
}
......
src/main/java/microservice/services/protocol/zmq/IMsgQServiceZmqImpl.java → src/main/java/microservice/app/microservice/service/pubsub/impl/protocol/zmq/IMsgQServiceZmqImpl.java
package microservice.services.protocol.zmq;
package microservice.service.pubsub.impl.protocol.zmq;
import com.eaio.uuid.UUID;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.flatbuffers.FlatBufferBuilder;
import microservice.MicroserviceApp;
import microservice.common.context.ParamValue;
import microservice.common.context.QueueMsg;
import microservice.defs.Constants;
import microservice.defs.Enums;
import microservice.handlers.Reactor;
import microservice.io.iface.ILogger;
import microservice.io.iface.IServiceDiscovery;
import microservice.params.ZMQParams;
import microservice.services.CommonServices;
import common.microservice.io.iface.IApp;
import common.microservice.utils.IDGenerator;
import common.microservice.context.ParamValue;
import common.microservice.context.QueueMsg;
import common.microservice.defs.Constants;
import common.microservice.defs.Enums;
import common.microservice.io.iface.ILogger;
import common.microservice.io.iface.IServiceDiscovery;
import common.microservice.params.ZMQParams;
import common.microservice.services.CommonServices;
import microservice.utils.ZSocketPool;
import org.zeromq.ZMQ;
import org.zeromq.ZSocket;
......@@ -23,8 +22,8 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import static common.microservice.services.CommonServices.buildServiceKey;
/**
* MsgQueue implementation in ZMQ
......@@ -58,7 +57,7 @@ public class IMsgQServiceZmqImpl extends CommonServices.IMsgQService {
QueueMsg queueMsg = QueueMsg.getRootAsQueueMsg(bb);
if (queueMsg != null){
MsgQContext msgQContext = getMsgQContext(queueMsg);
String key = Reactor.buildServiceKey(Enums.EnumServiceType.E_MSGQ,
String key = buildServiceKey(Enums.EnumServiceType.E_MSGQ,
CommonServices.EnumMsgQueueCommands.E_QUEUE,
msgQContext.topic);
reactor.delegate(parentService, key ,msgQContext);
......@@ -84,11 +83,12 @@ public class IMsgQServiceZmqImpl extends CommonServices.IMsgQService {
@Override
public boolean init() {
public boolean init(IApp app) {
boolean retstat = true;
if (MicroserviceApp.getsInstance() != null) {
this.appName = MicroserviceApp.getsInstance().getAppName();
this.logger = MicroserviceApp.getsInstance().getLogger();
this.app = app;
if (app != null) {
this.appName = app.getAppName();
this.logger = app.getLogger();
}
if (getServiceMode() != null) {
......@@ -206,9 +206,9 @@ public class IMsgQServiceZmqImpl extends CommonServices.IMsgQService {
*/
if (msgCtx.parameters == null) {
msgCtx.parameters = new HashMap<>();
msgCtx.parameters.put(Constants.MSGQ_ID, new UUID().toString());
msgCtx.parameters.put(Constants.MSGQ_ID, IDGenerator.createUUID());
} else
msgCtx.parameters.putIfAbsent(Constants.MSGQ_ID, new UUID().toString());
msgCtx.parameters.putIfAbsent(Constants.MSGQ_ID, IDGenerator.createUUID());
int topicOffset = msgBuilder.createString(msgCtx.topic);
int msgOffset = msgBuilder.createString(msgCtx.msg);
......
src/main/java/microservice/services/protocol/zmq/RestImpl.java → src/main/java/microservice/app/microservice/service/pubsub/impl/protocol/zmq/RestImpl.java
package microservice.services.protocol.zmq;
package microservice.service.pubsub.impl.protocol.zmq;
import microservice.common.context.RestMsg;
import microservice.io.iface.IRequest;
import microservice.io.iface.IResponse;
import microservice.services.IRestServiceZmqImpl;
import common.microservice.context.RestMsg;
import common.microservice.io.iface.IRequest;
import common.microservice.io.iface.IResponse;
import microservice.service.pubsub.impl.IRestServiceZmqImpl;
import microservice.utils.ZSocketPool;
import org.apache.commons.lang.SerializationUtils;
import org.zeromq.ZMQ;
......
src/main/java/microservice/services/protocol/zmq/ZmqWorkers.java → src/main/java/microservice/app/microservice/service/pubsub/impl/protocol/zmq/ZmqWorkers.java
package microservice.services.protocol.zmq;
package microservice.service.pubsub.impl.protocol.zmq;
import com.eaio.uuid.UUID;
import com.fasterxml.jackson.databind.ObjectMapper;
import common.microservice.utils.IDGenerator;
import microservice.MicroserviceApp;
import microservice.common.context.RestContext;
import microservice.common.context.RestMsg;
import microservice.io.iface.ILogger;
import microservice.params.ZMQParams;
import microservice.services.CommonServices;
import common.microservice.context.RestContext;
import common.microservice.io.iface.ILogger;
import common.microservice.params.ZMQParams;
import common.microservice.services.CommonServices;
import org.zeromq.ZMQ;
import org.zeromq.ZSocket;
import java.nio.ByteBuffer;
import java.util.List;
import static microservice.defs.Constants.EXIT_MSG;
import static microservice.defs.Constants.EXIT_MSG_LEN;
import static common.microservice.defs.Constants.EXIT_MSG;
import static common.microservice.defs.Constants.EXIT_MSG_LEN;
/**
* all workers common to rest amd msgQ
......@@ -108,7 +107,7 @@ public class ZmqWorkers {
this.parentService = parentService;
this.workerNumber = workerNumber;
bindAddress = ADDRESS + '/' +
new UUID().toString() + '/' +
IDGenerator.createUUID() + '/' +
String.valueOf(workerNumber);
}
......
src/main/java/microservice/utils/ServiceBuilderFactory.java → src/main/java/microservice/app/microservice/utils/ServiceBuilderFactory.java
package microservice.utils;
import microservice.services.CommonServices;
import microservice.io.iface.IRestClient;
import microservice.io.impl.IRestClientHttpImpl;
import microservice.services.IRestServiceHttpImpl;
import microservice.services.IRestServiceZmqImpl;
import microservice.params.RestClientParams;
import microservice.params.RestServerParams;
import microservice.params.ZMQParams;
import microservice.services.protocol.zmq.IMsgQServiceZmqImpl;
import clients.microservice.impl.IRestClientHttpImpl;
import common.microservice.services.CommonServices;
import common.microservice.io.iface.IRestClient;
import common.microservice.params.PulsarParams;
import microservice.service.pubsub.impl.IRestServiceHttpImpl;
import microservice.service.pubsub.impl.IRestServiceZmqImpl;
import common.microservice.params.RestClientParams;
import common.microservice.params.RestServerParams;
import common.microservice.params.ZMQParams;
import microservice.service.pubsub.impl.protocol.zmq.IMsgQServiceZmqImpl;
import service.microservice.IPubSubServicePulsarImpl;
/**
* Created by amir on 09/05/17.
......@@ -16,6 +18,9 @@ import microservice.services.protocol.zmq.IMsgQServiceZmqImpl;
public class ServiceBuilderFactory {
public static final String FAILED_IN_VALIDATING_PARAMS = " >> Failed in validating params";
public static final String EXCEPTION = "Exception >> ";
public static RestServiceHttpBuilder createRestServiceHttpBuilder(CommonServices.EnumRestServiceMode serviceMode){
return new RestServiceHttpBuilder(serviceMode);
}
......@@ -28,6 +33,10 @@ public class ServiceBuilderFactory {
return new MsgQServiceZmqBuilder(serviceMode);
}
public static PubSubServicePulsarBuilder createPubSubServicePulsarBuilder(CommonServices.EnumPubSubServiceMode serviceMode){
return new PubSubServicePulsarBuilder(serviceMode);
}
public interface IBuilder {
CommonServices.IService build();
}
......@@ -89,11 +98,11 @@ public class ServiceBuilderFactory {
break;
}
} catch (Exception exp){
System.err.println(this.getClass().getName().toString() + "Exception >> " + exp);
System.err.println(this.getClass().getName().toString() + EXCEPTION + exp);
restServiceHttp = null;
}
} else {
System.err.println(this.getClass().getName().toString() + " >> Failed in validating params");
System.err.println(this.getClass().getName().toString() + FAILED_IN_VALIDATING_PARAMS);
}
return restServiceHttp;
}
......@@ -203,11 +212,11 @@ public class ServiceBuilderFactory {
break;
}
} catch (Exception exp){
System.err.println(this.getClass().getName().toString() + "Exception >> " + exp);
System.err.println(this.getClass().getName().toString() + EXCEPTION + exp);
restServiceZmq = null;
}
} else {
System.err.println(this.getClass().getName().toString() + " >> Failed in validating params");
System.err.println(this.getClass().getName().toString() + FAILED_IN_VALIDATING_PARAMS);
}
return restServiceZmq;
}
......@@ -297,11 +306,11 @@ public class ServiceBuilderFactory {
break;
}
} catch (Exception exp){
System.err.println(this.getClass().getName().toString() + "Exception >> " + exp);
System.err.println(this.getClass().getName().toString() + EXCEPTION + exp);
msgQServiceZmq = null;
}
} else {
System.err.println(this.getClass().getName().toString() + " >> Failed in validating params");
System.err.println(this.getClass().getName().toString() + FAILED_IN_VALIDATING_PARAMS);
}
return msgQServiceZmq;
}
......@@ -317,4 +326,89 @@ public class ServiceBuilderFactory {
}
}
public static class PubSubServicePulsarBuilder implements IBuilder {
IPubSubServicePulsarImpl pubSubServicePulsar = null;
PulsarParams.PulsarParamsBuilder pulsarParamsBuilder = new PulsarParams.PulsarParamsBuilder();
CommonServices.EnumPubSubServiceMode serviceMode = CommonServices.EnumPubSubServiceMode.E_BOTH;
public PubSubServicePulsarBuilder(CommonServices.EnumPubSubServiceMode serviceMode) {
this.serviceMode = serviceMode;
}
public PubSubServicePulsarBuilder setConsumerPoolSize(int consumerPoolSize) {
this.pulsarParamsBuilder.setConsumersThreadPoolSize(consumerPoolSize);
return this;
}
public PubSubServicePulsarBuilder setServiceUrl(String serviceUrl) {
this.pulsarParamsBuilder.setServiceUrl(serviceUrl);
return this;
}
public PubSubServicePulsarBuilder setAdminUrl(String adminUrl) {
this.pulsarParamsBuilder.setAdminUrl(adminUrl);
return this;
}
public PubSubServicePulsarBuilder setClusters(String clusters) {
this.pulsarParamsBuilder.setClusters(clusters);
return this;
}
public PubSubServicePulsarBuilder setMsgTtl(int msgTtl) {
this.pulsarParamsBuilder.setMsgTtl(msgTtl);
return this;
}
public PubSubServicePulsarBuilder setRetentionTimeMin(int retentionTimeMin) {
this.pulsarParamsBuilder.setRetentionTimeMin(retentionTimeMin);
return this;
}
public PubSubServicePulsarBuilder setRetentionSizeMB(int retentionSizeMB) {
this.pulsarParamsBuilder.setRetentionSizeMB(retentionSizeMB);
return this;
}
/**
* msgs published to all services has 'public' namespace
* @param createPublicNamespace
* @return
*/
public PubSubServicePulsarBuilder setCreatePublicNamespace(boolean createPublicNamespace) {
this.pulsarParamsBuilder.setCreatePublicNamespace(createPublicNamespace);
return this;
}
private boolean validateParams() {
if(pulsarParamsBuilder.getServiceUrl() == null)
return false;
return true;
}
@Override
public CommonServices.IService build() {
if (validateParams()) {
try {
PulsarParams pulsarParams = pulsarParamsBuilder.build();
if (pulsarParams != null){
pubSubServicePulsar = new IPubSubServicePulsarImpl(pulsarParams);
pubSubServicePulsar.setServiceMode(serviceMode);
}
} catch (Exception exp){
System.err.println(this.getClass().getName().toString() + EXCEPTION + exp);
pubSubServicePulsar = null;
}
} else {
System.err.println(this.getClass().getName().toString() + FAILED_IN_VALIDATING_PARAMS);
}
return pubSubServicePulsar;
}
}
}
src/main/java/com/google/flatbuffers/FlatBufferBuilder.java → src/main/java/microservice/common/com/google/flatbuffers/FlatBufferBuilder.java
......@@ -16,8 +16,6 @@
package com.google.flatbuffers;
import static com.google.flatbuffers.Constants.*;
import java.nio.CharBuffer;
import java.nio.charset.CharacterCodingException;
import java.nio.charset.CharsetEncoder;
......@@ -27,6 +25,10 @@ import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.charset.Charset;
import static com.google.flatbuffers.Constants.FILE_IDENTIFIER_LENGTH;
import static com.google.flatbuffers.Constants.SIZEOF_INT;
import static com.google.flatbuffers.Constants.SIZEOF_SHORT;
/// @file
/// @addtogroup flatbuffers_java_api
/// @{
......@@ -204,7 +206,7 @@ public class FlatBufferBuilder {
*
* @param x A `short` to put into the buffer.
*/
public void putShort (short x) { bb.putShort (space -= Constants.SIZEOF_SHORT, x); }
public void putShort (short x) { bb.putShort (space -= SIZEOF_SHORT, x); }
/**
* Add an `int` to the buffer, backwards from the current location. Doesn't align nor
......@@ -212,7 +214,7 @@ public class FlatBufferBuilder {
*
* @param x An `int` to put into the buffer.
*/
public void putInt (int x) { bb.putInt (space -= Constants.SIZEOF_INT, x); }
public void putInt (int x) { bb.putInt (space -= SIZEOF_INT, x); }
/**
* Add a `long` to the buffer, backwards from the current location. Doesn't align nor
......@@ -258,14 +260,14 @@ public class FlatBufferBuilder {
*
* @param x A `short` to put into the buffer.
*/
public void addShort (short x) { prep(Constants.SIZEOF_SHORT, 0); putShort (x); }
public void addShort (short x) { prep(SIZEOF_SHORT, 0); putShort (x); }
/**
* Add an `int` to the buffer, properly aligned, and grows the buffer (if necessary).
*
* @param x An `int` to put into the buffer.
*/
public void addInt (int x) { prep(Constants.SIZEOF_INT, 0); putInt (x); }
public void addInt (int x) { prep(SIZEOF_INT, 0); putInt (x); }
/**
* Add a `long` to the buffer, properly aligned, and grows the buffer (if necessary).
......@@ -397,7 +399,7 @@ public class FlatBufferBuilder {
*/
public int createVectorOfTables(int[] offsets) {
notNested();
startVector(Constants.SIZEOF_INT, offsets.length, Constants.SIZEOF_INT);
startVector(SIZEOF_INT, offsets.length, SIZEOF_INT);
for(int i = offsets.length - 1; i >= 0; i--) addOffset(offsets[i]);
return endVector();
}
......
src/main/java/microservice/common/context/CrudMethod.java → src/main/java/microservice/common/common/microservice/context/CrudMethod.java
// automatically generated by the FlatBuffers compiler, do not modify
package microservice.common.context;
package common.microservice.context;
public final class CrudMethod {
private CrudMethod() { }
......
src/main/java/microservice/common/context/ParamValue.java → src/main/java/microservice/common/common/microservice/context/ParamValue.java
// automatically generated by the FlatBuffers compiler, do not modify
package microservice.common.context;
package common.microservice.context;
import java.nio.*;
import java.lang.*;
import java.util.*;
import com.google.flatbuffers.*;
@SuppressWarnings("unused")
......
src/main/java/microservice/common/common/microservice/context/PubSubContext.java 0 → 100644
package common.microservice.context;
import common.microservice.services.CommonServices;
import java.util.Deque;
import java.util.Map;
public class PubSubContext implements CommonServices.IMsgContext {
public String[] params = null;
public Map<String, Deque<String>> queryParameters = null;
public Map<String, String> pathParameters = null;
public String rcid; // request correlation id
public String getParameter(String paramName) {
if (queryParameters != null && queryParameters.containsKey(paramName))
return queryParameters.get(paramName).getFirst();
else
return null;
}
public String getPathParameter(String paramName) {
if (pathParameters != null )
return pathParameters.get(paramName);
else
return null;
}
@Override
public void setParameters(Map<String, String> parameters) {
pathParameters = parameters;
}
}
src/main/java/microservice/common/common/microservice/context/PubSubMsg.java 0 → 100644
package common.microservice.context;
import java.util.HashMap;
import java.util.Map;
public class PubSubMsg {
String content;
String mcid; // msg context id
Map<String, String> parameters = null;
public PubSubMsg(String content, String mcid) {
this.content = content;
this.mcid = mcid;
}
public PubSubMsg(String content, String mcid, Map<String, String> parameters) {
this.content = content;
this.mcid = mcid;
this.parameters = parameters;
}
public PubSubMsg() {
}
PubSubMsg addParameter(String param, String value){
if (parameters == null)
parameters = new HashMap<>();
parameters.put(param,value);
return this;
}
public Map<String, String> getParameters() {
return parameters;
}
public String getContent() {
return content;
}
public String getMcid() {
return mcid;
}
@Override
public String toString() {
return "mcid: " + mcid + ", content: " + content;
}
}
src/main/java/microservice/common/context/QueueMsg.java → src/main/java/microservice/common/common/microservice/context/QueueMsg.java
// automatically generated by the FlatBuffers compiler, do not modify
package microservice.common.context;
package common.microservice.context;
import java.nio.*;
import java.lang.*;
import java.util.*;
import com.google.flatbuffers.*;
@SuppressWarnings("unused")
......
src/main/java/microservice/common/context/RestContext.java → src/main/java/microservice/common/common/microservice/context/RestContext.java
package microservice.common.context;
package common.microservice.context;
import java.util.Deque;
import java.util.Map;
import com.fasterxml.jackson.databind.ObjectMapper;
import microservice.services.CommonServices;
import microservice.io.iface.IContainer;
import microservice.io.iface.IRequest;
import microservice.io.iface.IResponse;
import microservice.types.UserProfile;
import common.microservice.services.CommonServices;
import common.microservice.io.iface.IContainer;
import common.microservice.io.iface.IRequest;
import common.microservice.io.iface.IResponse;
import common.microservice.types.UserProfile;
public class RestContext implements CommonServices.IMsgContext
......
src/main/java/microservice/common/context/RestMsg.java → src/main/java/microservice/common/common/microservice/context/RestMsg.java
// automatically generated by the FlatBuffers compiler, do not modify
package microservice.common.context;
package common.microservice.context;
import java.nio.*;
import java.lang.*;
import java.util.*;
import com.google.flatbuffers.*;
@SuppressWarnings("unused")
......
src/main/java/microservice/common/context/RestResponse.java → src/main/java/microservice/common/common/microservice/context/RestResponse.java
// automatically generated by the FlatBuffers compiler, do not modify
package microservice.common.context;
package common.microservice.context;
import java.nio.*;
import java.lang.*;
import java.util.*;
import com.google.flatbuffers.*;
@SuppressWarnings("unused")
......
src/main/java/microservice/defs/Constants.java → src/main/java/microservice/common/common/microservice/defs/Constants.java
package microservice.defs;
package common.microservice.defs;
public class Constants
{
......@@ -25,4 +25,5 @@ public class Constants
public static final String TIMER = "Timer:";
public static final String HTTP_SCHEME = "http";
public static final String MSGQ_ID = "mqid";
public static final String DEFAULT_TENANT = "mcx";
}
src/main/java/microservice/common/common/microservice/io/iface/IApp.java 0 → 100644
package common.microservice.io.iface;
public interface IApp {
ILogger getLogger();
String getAppName();
String getServerName();
String getId();
IConfiguration getConfiguration();
}
src/main/java/microservice/io/iface/IConfiguration.java → src/main/java/microservice/common/common/microservice/io/iface/IConfiguration.java
package microservice.io.iface;
package common.microservice.io.iface;
import common.configuration.IConfigurationProvider;
......
src/main/java/microservice/io/iface/IContainer.java → src/main/java/microservice/common/common/microservice/io/iface/IContainer.java
package microservice.io.iface;
package common.microservice.io.iface;
import java.util.regex.Pattern;
import com.fasterxml.jackson.databind.JsonNode;
import microservice.handlers.BaseHandler;
/**
* this interface defines the basic operations
* required from the container implementor
......
src/main/java/microservice/io/iface/ILogger.java → src/main/java/microservice/common/common/microservice/io/iface/ILogger.java
package microservice.io.iface;
import java.util.Optional;
import java.util.stream.Stream;
package common.microservice.io.iface;
/**
* logger interface to use in the MS
......
src/main/java/microservice/io/iface/IMetricsFactory.java → src/main/java/microservice/common/common/microservice/io/iface/IMetricsFactory.java
package microservice.io.iface;
package common.microservice.io.iface;
/**
* metrics interface for adding/incerementing metrics counters (collectd/...)
......
src/main/java/microservice/io/iface/INotifyCallback.java → src/main/java/microservice/common/common/microservice/io/iface/INotifyCallback.java
package microservice.io.iface;
package common.microservice.io.iface;
import com.fasterxml.jackson.databind.JsonNode;
......
src/main/java/microservice/io/iface/IPubSub.java → src/main/java/microservice/common/common/microservice/io/iface/IPubSub.java
package microservice.io.iface;
package common.microservice.io.iface;
public interface IPubSub
{
......
src/main/java/microservice/io/iface/IRequest.java → src/main/java/microservice/common/common/microservice/io/iface/IRequest.java
package microservice.io.iface;
package common.microservice.io.iface;
import java.io.InputStream;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
public interface IRequest
{
......
src/main/java/microservice/io/iface/IResponse.java → src/main/java/microservice/common/common/microservice/io/iface/IResponse.java
package microservice.io.iface;
package common.microservice.io.iface;
import java.nio.ByteBuffer;
......
src/main/java/microservice/io/iface/IRestClient.java → src/main/java/microservice/common/common/microservice/io/iface/IRestClient.java
package microservice.io.iface;
package common.microservice.io.iface;
import microservice.params.CommandParams;
import common.microservice.params.CommandParams;
import com.fasterxml.jackson.databind.JsonNode;
import com.netflix.hystrix.HystrixCommand;
import com.netflix.hystrix.HystrixCommandGroupKey;
import microservice.types.BaseRestResponse;
import common.microservice.types.BaseRestResponse;
import java.util.function.Consumer;
public interface IRestClient
{
static final String COMMAND_ERROR = "Command Error: ";
void init(ILogger logger);
public abstract class Command extends HystrixCommand<BaseRestResponse> {
protected CommandParams reqCtx = null;
......
src/main/java/microservice/io/iface/IServiceDiscovery.java → src/main/java/microservice/common/common/microservice/io/iface/IServiceDiscovery.java
package microservice.io.iface;
package common.microservice.io.iface;
import microservice.defs.Enums;
import common.microservice.defs.Enums;
import common.configuration.IConfigurationProvider;
/**
......
src/main/java/microservice/io/impl/ILoggerConsoleImpl.java → src/main/java/microservice/common/common/microservice/io/impl/ILoggerConsoleImpl.java
package microservice.io.impl;
package common.microservice.io.impl;
import microservice.io.iface.ILogger;
import common.microservice.io.iface.ILogger;
import java.time.LocalDateTime;
......
src/main/java/microservice/common/paging/Href.java → src/main/java/microservice/common/common/microservice/paging/Href.java
package microservice.common.paging;
package common.microservice.paging;
/**
* Created by amir on 23/03/17.
......
src/main/java/microservice/common/paging/Links.java → src/main/java/microservice/common/common/microservice/paging/Links.java
package microservice.common.paging;
package common.microservice.paging;
/**
* Created by amir on 23/03/17.
......
src/main/java/microservice/common/paging/Page.java → src/main/java/microservice/common/common/microservice/paging/Page.java
package microservice.common.paging;
package common.microservice.paging;
/**
* Created by amir on 23/03/17.
......
src/main/java/microservice/common/paging/PageBuilder.java → src/main/java/microservice/common/common/microservice/paging/PageBuilder.java
package microservice.common.paging;
package common.microservice.paging;
import com.fasterxml.jackson.databind.node.ArrayNode;
......
src/main/java/microservice/common/paging/PageBuilderJsonNodeImpl.java → src/main/java/microservice/common/common/microservice/paging/PageBuilderJsonNodeImpl.java
package microservice.common.paging;
package common.microservice.paging;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ArrayNode;
......
src/main/java/microservice/common/paging/Sort.java → src/main/java/microservice/common/common/microservice/paging/Sort.java
package microservice.common.paging;
package common.microservice.paging;
/**
* Created by amir on 23/03/17.
......
src/main/java/microservice/params/CommandParamsBuilder.java → src/main/java/microservice/common/common/microservice/params/CommandParamsBuilder.java
package microservice.params;
package common.microservice.params;
import java.util.HashMap;
import java.util.Map;
......
src/main/java/microservice/params/RMQClientParams.java → src/main/java/microservice/common/common/microservice/params/RMQClientParams.java
package microservice.params;
package common.microservice.params;
public class RMQClientParams extends BaseClientParams
{
......
src/main/java/microservice/params/RestClientParams.java → src/main/java/microservice/common/common/microservice/params/RestClientParams.java
package microservice.params;
package common.microservice.params;
public class RestClientParams extends BaseClientParams
{
......
src/main/java/microservice/params/ZMQParams.java → src/main/java/microservice/common/common/microservice/params/ZMQParams.java
package microservice.params;
package common.microservice.params;
import jdk.nashorn.internal.ir.EmptyNode;
//import jdk.nashorn.internal.ir.EmptyNode;
/**
* Created by amir on 14/05/17.
......
src/main/java/microservice/common/BCrypt.java → src/main/java/microservice/common/common/microservice/security/BCrypt.java
......@@ -12,7 +12,7 @@
// ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
// OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
package microservice.common;
package common.microservice.security;
import java.io.UnsupportedEncodingException;
import java.security.SecureRandom;
......
src/main/java/microservice/common/EncryptionUtils.java → src/main/java/microservice/common/common/microservice/security/EncryptionUtils.java
package microservice.common;
package common.microservice.security;
import common.microservice.types.UserProfile;
import io.jsonwebtoken.*;
import microservice.types.UserProfile;
import javax.crypto.spec.SecretKeySpec;
import javax.xml.bind.DatatypeConverter;
......
src/main/java/microservice/services/CommonServices.java → src/main/java/microservice/common/common/microservice/services/CommonServices.java
package microservice.services;
import microservice.common.context.CrudMethod;
import microservice.defs.Constants;
import microservice.defs.Enums;
import microservice.io.iface.IMetricsFactory;
import microservice.io.iface.IRequest;
import microservice.io.iface.IResponse;
import microservice.io.iface.IServiceDiscovery;
import microservice.params.CommandParams;
import microservice.types.BaseRestResponse;
package common.microservice.services;
import common.microservice.context.CrudMethod;
import common.microservice.defs.Constants;
import common.microservice.defs.Enums;
import common.microservice.io.iface.*;
import common.microservice.params.CommandParams;
import common.microservice.types.BaseRestResponse;
import java.util.List;
import java.util.Map;
......@@ -35,11 +32,12 @@ public class CommonServices {
public static abstract class IService {
protected IServiceReactor reactor = null;
protected IMetricsFactory metricsFactory = null;
public abstract boolean init();
protected IApp app = null;
public abstract boolean init(IApp app);
public abstract void run();
public abstract void shutdown();
public abstract void handleNotImplmented(IMsgContext msgContext);
public abstract Enums.EnumServiceType getServiceType();
/// when relevant
public abstract void register(IServiceDiscovery serviceDiscovery, String id);
......@@ -138,14 +136,15 @@ public class CommonServices {
protected static String DELETE_METRIC_PREFIX = Constants.METER + SEND + Enums.EnumServiceType.E_REST.name() + ':' + CrudMethod.name(CrudMethod.Delete) + ':';
EnumRestServiceMode serviceMode = EnumRestServiceMode.E_UNKNOWN;
protected microservice.io.iface.IRestClient restClient = null;
protected common.microservice.io.iface.IRestClient restClient = null;
public EnumRestServiceMode getServiceMode() { return serviceMode; }
public void setServiceMode(EnumRestServiceMode serviceMode) { this.serviceMode = serviceMode; }
public void setRestClient(microservice.io.iface.IRestClient restClient) {
public void setRestClient(common.microservice.io.iface.IRestClient restClient) {
this.restClient = restClient;
}
@Override
public Enums.EnumServiceType getServiceType() { return Enums.EnumServiceType.E_REST; }
///////////// SYNC //////////////////////////
public abstract BaseRestResponse create(CommandParams cmdParams);
......@@ -166,29 +165,88 @@ public class CommonServices {
public abstract void startAsync(IRequest request,Runnable asyncFunc);
}
enum EnumPubSubCommands implements IServiceCommands {
public enum EnumPubSubCommands implements IServiceCommands {
E_SUBSCRIBE,
E_PUBLISH,
E_NOTIFY
}
public enum EnumPubSubServiceMode {
E_PUBLISHER,
E_SUBSCRIBER,
E_BOTH
}
public static abstract class IPubSubService extends IService {
public class PubSubMsgContext implements IMsgContext {
public String topic;
public String msg;
public static class PubSubMsgContext implements IMsgContext {
private String topic;
private String msg;
private String mcid = null;
public Map<String, String> parameters = null;
public PubSubMsgContext(String topic, String msg) {
this.topic = topic;
this.msg = msg;
}
public PubSubMsgContext(String topic,
String msg,
String mcid,
Map<String, String> parameters) {
this.topic = topic;
this.msg = msg;
this.mcid = mcid;
this.parameters = parameters;
}
public String getTopic() {
return topic;
}
public void setTopic(String topic) {
this.topic = topic;
}
public String getMsg() {
return msg;
}
public void setMsg(String msg) {
this.msg = msg;
}
public String getMcid() {
return mcid;
}
public PubSubMsgContext setMcid(String mcid) {
this.mcid = mcid;
return this;
}
public Map<String, String> getParameters() {
return parameters;
}
@Override
public void setParameters(Map<String, String> parameters) {
this.parameters = parameters;
}
}
EnumPubSubServiceMode serviceMode = EnumPubSubServiceMode.E_BOTH;
public EnumPubSubServiceMode getServiceMode() {
return serviceMode;
}
public void setServiceMode(EnumPubSubServiceMode serviceMode) {
this.serviceMode = serviceMode;
}
@Override
public Enums.EnumServiceType getServiceType() { return Enums.EnumServiceType.E_PUBSUB; }
public abstract void subscribe(String topic, Consumer<PubSubMsgContext> notifyFunc,Consumer<String> errorFunc);
public abstract void unsubscribe(String topic);
public abstract void publish(PubSubMsgContext pubSubMsgContext);
......@@ -227,7 +285,8 @@ public class CommonServices {
public EnumMsgQServiceMode getServiceMode() {
return serviceMode;
}
@Override
public Enums.EnumServiceType getServiceType() { return Enums.EnumServiceType.E_MSGQ; }
public void setServiceMode(EnumMsgQServiceMode serviceMode) {
this.serviceMode = serviceMode;
}
......@@ -277,4 +336,21 @@ public class CommonServices {
public abstract void stop() throws InterruptedException;
}
public static String buildServiceKey(CommonServices.MethodParams methodParams) {
return Constants.SLASH_SEPERATOR +
methodParams.getServiceType().name() + Constants.TYPE_PREFIX_SEPERATOR +
methodParams.getServiceCommand().toString() + Constants.TYPE_PREFIX_SEPERATOR +
methodParams.getResourceUri();
}
public static String buildServiceKey(Enums.EnumServiceType enumServiceType,
CommonServices.IServiceCommands serviceCommands,
String resourceUri) {
return new StringBuilder(32).append(Constants.SLASH_SEPERATOR).append(enumServiceType.name()).append(Constants.TYPE_PREFIX_SEPERATOR)
.append(serviceCommands.toString()).append(Constants.TYPE_PREFIX_SEPERATOR)
.append(resourceUri).toString();
}
}
src/main/java/microservice/types/BaseRestResponse.java → src/main/java/microservice/common/common/microservice/types/BaseRestResponse.java
package microservice.types;
package common.microservice.types;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonIgnore;
......
src/main/java/microservice/types/UserProfile.java → src/main/java/microservice/common/common/microservice/types/UserProfile.java
package microservice.types;
package common.microservice.types;
import io.jsonwebtoken.Claims;
import io.jsonwebtoken.impl.DefaultClaims;
import microservice.services.CommonServices;
import common.microservice.services.CommonServices;
import java.util.HashMap;
import java.util.Map;
......
src/main/java/microservice/utils/CheckedFunction.java → src/main/java/microservice/common/common/microservice/utils/CheckedFunction.java
package microservice.utils;
package common.microservice.utils;
@FunctionalInterface
public interface CheckedFunction<T,R> {
......
src/main/java/microservice/utils/Either.java → src/main/java/microservice/common/common/microservice/utils/Either.java
package microservice.utils;
package common.microservice.utils;
import com.google.gdata.util.common.base.Pair;
......
src/main/java/microservice/utils/ICacheClientGuavaImpl.java → src/main/java/microservice/common/common/microservice/utils/ICacheClientGuavaImpl.java
package microservice.utils;
package common.microservice.utils;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import microservice.services.CommonServices;
import common.microservice.services.CommonServices;
import java.util.concurrent.TimeUnit;
......
src/main/java/microservice/common/common/microservice/utils/IDGenerator.java 0 → 100644
package common.microservice.utils;
import com.eaio.uuid.UUID;
public class IDGenerator {
public static String createUUID(){
return new UUID().toString();
}
public static long createUUIDLong(){
return new UUID().getTime();
}
}
src/main/java/microservice/common/common/microservice/utils/Logging.java 0 → 100644
package common.microservice.utils;
import common.microservice.io.iface.ILogger;
public class Logging {
/**
* logging utils
*/
public static void logRcid(ILogger logger,String from, String rcid){
logger.info(from + " RCID: " + rcid);
}
public static void logMcid(ILogger logger,String from, String mcid){
logger.info(from + " MCID: " + mcid);
}
}
src/main/java/microservice/service/pubsub/common/microservice/params/PulsarParams.java 0 → 100644
package common.microservice.params;
import org.apache.pulsar.shade.org.apache.commons.lang.StringUtils;
/**
* parameters to init pulsar pub-sub
* namepaces and topic - admin
*/
public class PulsarParams {
private final int retentionTimeMin;
private final int retentionSizeMB;
int msgTtl;
String clusters;
String serviceUrl;
String adminUrl;
int consumersThreadPoolSize;
boolean createPublicNamespace;
public static class PulsarParamsBuilder {
public static final int DEFAULT_ADMIN_PORT = 8080;
public static final int DEFAULT_SERVICE_PORT = 6650;
public static final String STANDALONE_CLUSTER = "standalone";
int msgTtl = 3600;
String clusters = STANDALONE_CLUSTER;
String serviceUrl;
String adminUrl = null;
int consumersThreadPoolSize = 0;
boolean createPublicNamespace = false;
int retentionTimeMin = 1440; // a day
int retentionSizeMB = 100; // 100MB
public PulsarParamsBuilder() {
// empty
}
public String getServiceUrl() {
return serviceUrl;
}
public PulsarParamsBuilder setServiceUrl(String serviceUrl) {
final String[] parts = serviceUrl.split(":");
if(parts.length < 2 ) { // || !StringUtils.isNumeric(parts[1]))
this.serviceUrl = String.format("%s:%u",serviceUrl, DEFAULT_SERVICE_PORT);
} else if (!StringUtils.isNumeric(parts[1])){
this.serviceUrl = String.format("%s:%u",parts[0], DEFAULT_SERVICE_PORT);
} else {
this.serviceUrl = serviceUrl;
}
return this;
}
public PulsarParamsBuilder setMsgTtl(int msgTtl) {
this.msgTtl = msgTtl;
return this;
}
public PulsarParamsBuilder setClusters(String clusters) {
this.clusters = clusters;
return this;
}
public PulsarParamsBuilder setAdminUrl(String adminUrl) {
this.adminUrl = adminUrl;
return this;
}
public void setConsumersThreadPoolSize(int consumersThreadPoolSize) {
this.consumersThreadPoolSize = consumersThreadPoolSize;
}
public PulsarParamsBuilder setCreatePublicNamespace(boolean createPublicNamespace) {
this.createPublicNamespace = createPublicNamespace;
return this;
}
public PulsarParamsBuilder setRetentionTimeMin(int retentionTimeMin) {
this.retentionTimeMin = retentionTimeMin;
return this;
}
public PulsarParamsBuilder setRetentionSizeMB(int retentionSizeMB) {
this.retentionSizeMB = retentionSizeMB;
return this;
}
public PulsarParams build(){
try {
if (adminUrl == null) {
String host = serviceUrl.substring(0, serviceUrl.indexOf(':'));
adminUrl = String.format("%s:%d",host, DEFAULT_ADMIN_PORT);
}
if (consumersThreadPoolSize == 0){
consumersThreadPoolSize = Runtime.getRuntime().availableProcessors();
}
return new PulsarParams(msgTtl,
retentionTimeMin,
retentionSizeMB,
clusters,
serviceUrl,
adminUrl,
consumersThreadPoolSize,
createPublicNamespace);
} catch (Exception e){
System.err.println("PulsarParamsBuilder > " + e.toString());
}
return null;
}
}
public PulsarParams(int msgTtl,
int retentionTimeMin,
int retentionSizeMB,
String clusters,
String serviceUrl,
String adminUrl,
int consumersThreadPoolSize,
boolean createPublicNamespace) {
this.msgTtl = msgTtl;
this.retentionTimeMin = retentionTimeMin;
this.retentionSizeMB = retentionSizeMB;
this.clusters = clusters;
this.serviceUrl = serviceUrl;
this.adminUrl = adminUrl;
this.consumersThreadPoolSize = consumersThreadPoolSize;
this.createPublicNamespace = createPublicNamespace;
}
public int getMsgTtl() {
return msgTtl;
}
public int getRetentionTimeMin() {
return retentionTimeMin;
}
public int getRetentionSizeMB() {
return retentionSizeMB;
}
public String getClusters() {
return clusters;
}
public String getServiceUrl() {
return serviceUrl;
}
public void setServiceUrl(String serviceUrl) {
this.serviceUrl = serviceUrl;
}
public String getAdminUrl() {
return adminUrl;
}
public void setAdminUrl(String adminUrl) {
this.adminUrl = adminUrl;
}
public int getConsumersThreadPoolSize() {
return consumersThreadPoolSize;
}
public boolean isCreatePublicNamespace() {
return createPublicNamespace;
}
}
src/test/java/microservice/TestCommandClient.java
......@@ -2,13 +2,14 @@ package microservice;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import microservice.io.iface.IRestClient;
import common.microservice.io.iface.ILogger;
import common.microservice.io.iface.IRestClient;
import microservice.io.impl.IRMQClientRestImpl;
import microservice.params.BaseClientParams;
import microservice.params.CommandParams;
import microservice.params.CommandParamsBuilder;
import microservice.params.RMQClientParams;
import microservice.types.BaseRestResponse;
import common.microservice.params.BaseClientParams;
import common.microservice.params.CommandParams;
import common.microservice.params.CommandParamsBuilder;
import common.microservice.params.RMQClientParams;
import common.microservice.types.BaseRestResponse;
import org.junit.Test;
import rx.Observable;
......@@ -48,6 +49,11 @@ public class TestCommandClient {
}
@Override
public void init(ILogger logger) {
}
@Override
public BaseRestResponse create(CommandParams reqCtx) {
BaseRestResponse brr = new BaseRestResponse(true, null);
Optional<Observable<BaseRestResponse>> brro = Optional.empty();
......
src/test/java/microservice/TestConfiguration.java
package microservice;
import microservice.io.iface.IConfiguration;
import common.microservice.io.iface.IConfiguration;
import microservice.io.impl.IConfigurationConfigPropImpl;
import microservice.io.impl.IServiceDiscoveryConsulImpl;
......
src/test/java/microservice/TestEncryption.java
package microservice;
import io.jsonwebtoken.ExpiredJwtException;
import microservice.common.EncryptionUtils;
import microservice.types.UserProfile;
import common.microservice.security.EncryptionUtils;
import common.microservice.types.UserProfile;
import org.junit.Test;
import static microservice.common.EncryptionUtils.checkPassword;
import static common.microservice.security.EncryptionUtils.checkPassword;
/**
* Created by amir on 03/08/16.
......
src/test/java/microservice/TestLogger.java
package microservice;
import microservice.io.iface.ILogger;
import microservice.io.impl.ILoggerConsoleImpl;
import common.microservice.io.iface.ILogger;
import common.microservice.io.impl.ILoggerConsoleImpl;
import org.junit.Test;
import java.time.LocalDateTime;
......
src/test/java/microservice/TestMQTT.java
......@@ -3,8 +3,8 @@ package microservice;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.junit.Test;
import microservice.io.iface.INotifyCallback;
import microservice.io.iface.IPubSub;
import common.microservice.io.iface.INotifyCallback;
import common.microservice.io.iface.IPubSub;
import microservice.io.impl.IPubSubMQTTImpl;
import com.fasterxml.jackson.databind.JsonNode;
......
src/test/java/microservice/TestMicroClient.java
package microservice;
import clients.microservice.impl.IRestClientHttpImpl;
import clients.microservice.utils.RestHttpClient;
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
import microservice.common.context.CrudMethod;
import microservice.common.context.RestResponse;
import microservice.defs.Constants;
import microservice.defs.Enums;
import microservice.io.iface.IRestClient;
import microservice.io.impl.IRestClientHttpImpl;
import common.microservice.defs.Enums;
import common.microservice.io.iface.IRestClient;
import microservice.io.impl.IServiceDiscoveryConsulImpl;
import microservice.utils.RestHttpClient;
import okhttp3.Call;
import okhttp3.Callback;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import org.apache.http.HttpEntity;
import org.apache.http.HttpHost;
import org.apache.http.HttpResponse;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpRequestBase;
import org.apache.http.client.utils.URIBuilder;
import org.apache.http.concurrent.FutureCallback;
import org.apache.http.entity.ContentType;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager;
import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
import org.apache.http.nio.entity.NStringEntity;
import org.apache.http.nio.reactor.ConnectingIOReactor;
import org.apache.http.util.EntityUtils;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.ResponseListener;
import org.elasticsearch.client.RestClient;
import org.junit.Test;
import microservice.MicroserviceClient.EnumRestClientType;
import microservice.params.CommandParams;
import microservice.params.RestClientParams;
import common.microservice.params.CommandParams;
import common.microservice.params.RestClientParams;
import com.fasterxml.jackson.databind.JsonNode;
import microservice.types.BaseRestResponse;
import rx.Observable;
import rx.apache.http.ObservableHttp;
import rx.apache.http.ObservableHttpResponse;
import common.microservice.types.BaseRestResponse;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import static microservice.utils.RestHttpClient.DEFAULT_CONNECTION_REQUEST_TIMEOUT_MILLIS;
import static microservice.utils.RestHttpClient.DEFAULT_CONNECT_TIMEOUT_MILLIS;
import static microservice.utils.RestHttpClient.DEFAULT_MAX_CONN_PER_ROUTE;
import static microservice.utils.RestHttpClient.DEFAULT_MAX_CONN_TOTAL;
import static microservice.utils.RestHttpClient.DEFAULT_SOCKET_TIMEOUT_MILLIS;
import static clients.microservice.utils.RestHttpClient.*;
public class TestMicroClient
......
src/test/java/microservice/TestMicroserviceApp.java
package microservice;
import clients.microservice.impl.IRestClientHttpImpl;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
import com.fasterxml.jackson.databind.node.ObjectNode;
import microservice.common.context.RestContext;
import microservice.defs.Enums;
import microservice.services.CommonServices;
import microservice.io.iface.IRestClient;
import microservice.io.impl.*;
import microservice.services.IRestServiceHttpImpl;
import microservice.params.*;
import microservice.services.IRestServiceZmqImpl;
import microservice.types.BaseRestResponse;
import common.microservice.context.RestContext;
import common.microservice.defs.Enums;
import common.microservice.io.iface.IRestClient;
import common.microservice.params.*;
import common.microservice.services.CommonServices;
import common.microservice.types.BaseRestResponse;
import microservice.io.impl.IRestServerRMQImpl;
import microservice.io.impl.IRestServerUndertowImpl;
import microservice.io.impl.IServiceDiscoveryConsulImpl;
import microservice.service.pubsub.impl.IRestServiceHttpImpl;
import microservice.service.pubsub.impl.IRestServiceZmqImpl;
import microservice.utils.ServiceBuilderFactory;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.junit.Test;
......@@ -37,7 +39,7 @@ public class TestMicroserviceApp {
String appName = "testApp";
//ILogger logger = new ILogger4jImpl(appName);
microservice.MicroserviceApp msApp = new microservice.MicroserviceApp(appName);
MicroserviceApp msApp = new microservice.MicroserviceApp(appName);
msApp.withMetrics()
//.withDefaultServiceAuthorization()
.withPubSub(new microservice.io.impl.IPubSubMQTTImpl("tcp://localhost",0,null,0))
......@@ -52,30 +54,30 @@ public class TestMicroserviceApp {
}
@Test
public void testRMQApp() throws MqttException, Exception
{
System.setProperty("configFile.location","/opt/mcx/config/config.properties");
BaseClientParams clientParams = new RestClientParams("other-service", true, 10, "localhost:32010","localhost:6379");
final IServiceDiscoveryConsulImpl serDisco = new IServiceDiscoveryConsulImpl("localhost", 8500);
IRestClient cmdClient = new IRestClientHttpImpl(clientParams).withServiceDiscovery(serDisco);
String appName = "testApp";
//ILogger logger = new ILogger4jImpl(appName);
microservice.MicroserviceApp msApp = new microservice.MicroserviceApp(appName);
msApp.withMetrics()
//.withDefaultServiceAuthorization()
.withPubSub(new microservice.io.impl.IPubSubMQTTImpl("tcp://localhost",0,null,0))
//.withServiceDiscovery(serDisco)
.withMonitoring()
// .addHandler("/test",new TestMicroserviceHandler())
.addMicroserviceClient(new MicroserviceClient(cmdClient,clientParams))
//.addMicroserviceClient("rabbit-service",new MicroserviceClient(EnumRestClientType.E_RABBITMQ,clientParams))
.addRestServer(new IRestServerUndertowImpl(new RestServerParams(32000, "localhost", 2)))
.addRestServer(new IRestServerRMQImpl(new RMQClientParams(appName, "myFirstQ@localhost", null, 1, 200, "/logs")))
.build()
.run();
}
// @Test
// public void testRMQApp() throws MqttException, Exception
// {
// System.setProperty("configFile.location","/opt/mcx/config/config.properties");
// BaseClientParams clientParams = new RestClientParams("other-service", true, 10, "localhost:32010","localhost:6379");
// final IServiceDiscoveryConsulImpl serDisco = new IServiceDiscoveryConsulImpl("localhost", 8500);
// IRestClient cmdClient = new IRestClientHttpImpl(clientParams).withServiceDiscovery(serDisco);
//
// String appName = "testApp";
// //ILogger logger = new ILogger4jImpl(appName);
// microservice.MicroserviceApp msApp = new microservice.MicroserviceApp(appName);
// msApp.withMetrics()
// //.withDefaultServiceAuthorization()
// .withPubSub(new microservice.io.impl.IPubSubMQTTImpl("tcp://localhost",0,null,0))
// //.withServiceDiscovery(serDisco)
// .withMonitoring()
// // .addHandler("/test",new TestMicroserviceHandler())
// .addMicroserviceClient(new MicroserviceClient(cmdClient,clientParams))
// //.addMicroserviceClient("rabbit-service",new MicroserviceClient(EnumRestClientType.E_RABBITMQ,clientParams))
// .addRestServer(new IRestServerUndertowImpl(new RestServerParams(32000, "localhost", 2)))
// .addRestServer(new IRestServerRMQImpl(new RMQClientParams(appName, "myFirstQ@localhost", null, 1, 200, "/logs")))
// .build()
// .run();
// }
@Test
......@@ -84,7 +86,7 @@ public class TestMicroserviceApp {
System.setProperty("configFile.location","/opt/mcx/config/config.properties");
System.setProperty("influxdb.hostport","172.16.1.244:8086");
String appName = "testApp";
String appName = "activities";
/**
* creating the services
*/
......@@ -100,6 +102,14 @@ public class TestMicroserviceApp {
CommonServices.IService zmqMsgQ = ServiceBuilderFactory.createMsgQServiceZmqBuilder(CommonServices.EnumMsgQServiceMode.E_SERVER)
.setServerParams(new ZMQParams.ServerParams(ZMQParams.ServerParams.EnumProtocol.eTcp,32020,"localhost"))
.build();
CommonServices.IService pulsarPubSub = ServiceBuilderFactory.createPubSubServicePulsarBuilder(CommonServices.EnumPubSubServiceMode.E_BOTH)
.setServiceUrl("localhost:6650")
.setMsgTtl(60)
.setRetentionSizeMB(20)
.setRetentionTimeMin(60)
.setAdminUrl(null)
.setClusters("standalone")
.build();
microservice.MicroserviceApp msApp = new microservice.MicroserviceApp(appName);
msApp.withMetrics()
......@@ -108,6 +118,7 @@ public class TestMicroserviceApp {
.addService(Enums.EnumServiceType.E_REST,httpRestService,"undertowRestService")
.addService(Enums.EnumServiceType.E_REST,zmqRestService,"zmqRestService")
.addService(Enums.EnumServiceType.E_MSGQ,zmqMsgQ,"zmqMsgQService")
.addService(Enums.EnumServiceType.E_PUBSUB,pulsarPubSub,"pulsarPubSub")
.addMethodClass(new MethodClass())
.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");
......@@ -164,12 +175,28 @@ public class TestMicroserviceApp {
(msgCtx,orgService) -> {
testZmqMsgQueue((CommonServices.IMsgQService.MsgQContext)msgCtx);
}));
methodParamsList.add(new CommonServices.MethodParams(Enums.EnumServiceType.E_PUBSUB,
CommonServices.EnumPubSubCommands.E_NOTIFY,
"/activity",
(msgCtx,orgService) -> {
testPubSubMsg((CommonServices.IPubSubService.PubSubMsgContext)msgCtx);
}));
}
long startTime = 0;
ObjectMapper objectMapper = new ObjectMapper();
private void testPubSubMsg(CommonServices.IPubSubService.PubSubMsgContext msgCtx) {
System.out.println("Recieved mcid: " + msgCtx.getMcid());
try {
JsonNode jsonNode = objectMapper.readValue(msgCtx.getMsg(), JsonNode.class);
handleJsonTestStartStop(jsonNode);
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* around 300k/s
* @param msgCtx
......@@ -177,22 +204,26 @@ public class TestMicroserviceApp {
private void testZmqMsgQueue(CommonServices.IMsgQService.MsgQContext msgCtx) {
try {
JsonNode jsonNode = objectMapper.readValue(msgCtx.msg, JsonNode.class);
String state = jsonNode.path("state").asText();
switch (state){
case "start":
startTime = System.currentTimeMillis();
break;
case "end":
System.out.println("Test took: " + String.valueOf(System.currentTimeMillis() - startTime) + "ms");
break;
case "msg":
break;
}
handleJsonTestStartStop(jsonNode);
} catch (IOException e) {
e.printStackTrace();
}
}
private void handleJsonTestStartStop(JsonNode jsonNode) {
String state = jsonNode.path("state").asText();
switch (state){
case "start":
startTime = System.currentTimeMillis();
break;
case "end":
System.out.println("Test took: " + String.valueOf(System.currentTimeMillis() - startTime) + "ms");
break;
case "msg":
break;
}
}
private void testZmqRead(RestContext msgCtx) {
RestContext restContext = (RestContext)msgCtx;
CommonServices.IRestService inRestService = (CommonServices.IRestService)MicroserviceApp.getsInstance().getService(Enums.EnumServiceType.E_REST,"undertowRestService");
......
src/test/java/microservice/TestPulsar.java 0 → 100644
package microservice;
import common.microservice.context.PubSubMsg;
import common.microservice.utils.IDGenerator;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminBuilder;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.admin.Tenants;
import org.apache.pulsar.client.admin.internal.PulsarAdminBuilderImpl;
import org.apache.pulsar.client.api.*;
import org.apache.pulsar.client.api.url.URL;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
import org.apache.pulsar.client.impl.schema.JSONSchema;
import org.apache.pulsar.common.policies.data.Policies;
import org.apache.pulsar.common.policies.data.TenantInfo;
import org.junit.Test;
import java.net.MalformedURLException;
import java.net.URISyntaxException;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.regex.Pattern;
public class TestPulsar {
@Test
public void testTopicProducer() throws PulsarClientException {
String localClusterUrl = "pulsar://localhost:6650";
String namespace = "mcx/testApp"; // This namespace is created automatically
String topic = String.format("persistent://%s/activity", namespace);
PulsarClient client = PulsarClient.builder()
.serviceUrl(localClusterUrl)
.build();
Producer<PubSubMsg> producer = client
.newProducer(JSONSchema.of(PubSubMsg.class))
.topic(topic)
.create();
// final Producer<byte[]> producer = client.newProducer().topic(topic).create();
// Publish 10 messages to the topic
for (int i = 0; i < 100; i++) {
PubSubMsg pubSubMsg = new PubSubMsg("hello - " + String.valueOf(System.currentTimeMillis()),String.valueOf(i));
producer.send(pubSubMsg);
// producer.send(String.format("Message number %d", i).getBytes());
System.out.println("Sending message");
// final Message message = MessageBuilder.create()
// .setContent(String.format("Message number %d", i).getBytes())
// .build();
// producer.send(message);
}
producer.close();
client.close();
}
@Test
public void testTopicConsumer() throws PulsarClientException {
String localClusterUrl = "pulsar://localhost:6650";
String namespace = "mcx/testApp";
String topic = String.format("persistent://%s/my-topic", namespace);
PulsarClient client = PulsarClient.builder()
.serviceUrl(localClusterUrl)
.build();
//final Consumer consumer = client.newConsumer().subscriptionName("my-sub-1").topic(topic).subscribe();
Pattern allTopicsInNamespace = Pattern.compile("persistent://mcx/testApp/.*");
final Consumer<PubSubMsg> consumer = client.newConsumer(JSONSchema.of(PubSubMsg.class))
.subscriptionName(IDGenerator.createUUID())
.subscriptionType(SubscriptionType.Shared) // enable for multi-instance
.topicsPattern(allTopicsInNamespace)
.subscribe();
// final Consumer consumer = client.newConsumer()
// .subscriptionName(IDGenerator.createUUID())
// .subscriptionType(SubscriptionType.Shared) // enable for multi-instance
// .topicsPattern(allTopicsInNamespace)
// .subscribe();
for (int i = 0; i < 10; i++) {
// Wait for a message
Message<PubSubMsg> msg = consumer.receive();
System.out.printf("Message received: %s\n", msg.getValue().toString());
// Acknowledge the message so that it can be deleted by the message broker
consumer.acknowledge(msg);
}
consumer.close();
client.close();
}
@Test
public void testPulsarAdmin() throws PulsarClientException, PulsarAdminException {
Set<String> clusters = new HashSet<>(Arrays.asList("standalone"));
PulsarAdmin admin = new PulsarAdminBuilderImpl().serviceHttpUrl("http://localhost:8080").build();
final Tenants tenants = admin.tenants();
final List<String> tenantList = tenants.getTenants();
final TenantInfo tenantInfo = new TenantInfo();
tenantInfo.setAllowedClusters(clusters);
if (!tenantList.contains("mcx")){
// creating tenant
tenants.createTenant("mcx", tenantInfo);
} else {
tenants.updateTenant("mcx",tenantInfo);
}
final List<String> namespaces = admin.namespaces().getNamespaces("mcx");
String namespaceStr = "mcx/testApp";
if (!namespaces.contains(namespaceStr)){
admin.namespaces().createNamespace(namespaceStr);
admin.namespaces().setNamespaceMessageTTL(namespaceStr,3600);
admin.namespaces().setNamespaceReplicationClusters(namespaceStr,clusters);
} else {
admin.namespaces().deleteNamespace(namespaceStr);
}
admin.close();
}
}
src/test/java/microservice/TestServiceDiscovery.java
package microservice;
import microservice.defs.Enums;
import microservice.io.iface.IServiceDiscovery;
import common.microservice.defs.Enums;
import common.microservice.io.iface.IServiceDiscovery;
import microservice.io.impl.IServiceDiscoveryConsulImpl;
import org.junit.Test;
......
src/test/java/microservice/TestServicesAndMethods.java
......@@ -14,13 +14,10 @@ import com.google.common.cache.CacheBuilder;
import common.JsonHandler;
import io.undertow.predicate.Predicate;
import io.undertow.util.PathMatcher;
import io.undertow.util.PathTemplateMatcher;
import microservice.params.ZMQParams;
import microservice.services.CommonServices;
import microservice.services.IRestServiceZmqImpl;
import microservice.types.BaseRestResponse;
import common.microservice.params.ZMQParams;
import common.microservice.services.CommonServices;
import microservice.utils.ServiceBuilderFactory;
import org.apache.commons.lang.SerializationUtils;
import org.junit.Assert;
import org.junit.Test;
......@@ -39,7 +36,7 @@ import java.util.stream.Collectors;
*/
public class TestServicesAndMethods {
public static final int ITERATIONS = 10000000;
public static final int ITERATIONS = 10000;
private static class RoutingMatch {
......@@ -264,7 +261,7 @@ public class TestServicesAndMethods {
.setClientParams(new ZMQParams.ServerParams(ZMQParams.ServerParams.EnumProtocol.eTcp, 32020, "localhost"))
.build();
CommonServices.IMsgQService msgQService = (CommonServices.IMsgQService)iService;
msgQService.init();
msgQService.init(null);
msgQService.run();
ObjectNode objectNode = JsonNodeFactory.instance.objectNode().put("state", "start").put("iterations", ITERATIONS);
System.out.println("Testing " + String.valueOf(ITERATIONS) + " iterations");
......@@ -278,4 +275,32 @@ public class TestServicesAndMethods {
msgQService.send(new CommonServices.IMsgQService.MsgQContext("/test/msgQ/zmq", objectNode.toString()));
//Thread.sleep(1000);
}
@Test
public void testPubSubPulsar() throws InterruptedException {
CommonServices.IService iService = ServiceBuilderFactory.createPubSubServicePulsarBuilder(CommonServices.EnumPubSubServiceMode.E_PUBLISHER)
.setServiceUrl("localhost:6650")
.setAdminUrl("localhost:8080")
// .setCreatePublicNamespace(true)
.build();
CommonServices.IPubSubService pubSubService = (CommonServices.IPubSubService)iService;
pubSubService.init(null);
pubSubService.run();
//String topic = "/testApp/activity"; // '[domain]/[method]'
String topic = "/activities/activity"; // '[domain]/[method]'
ObjectNode objectNode = JsonNodeFactory.instance.objectNode().put("state", "start").put("iterations", ITERATIONS);
System.out.println("Testing " + String.valueOf(ITERATIONS) + " iterations");
long start = System.currentTimeMillis();
pubSubService.publish(new CommonServices.IPubSubService.PubSubMsgContext(topic, objectNode.toString()));
objectNode.put("state", "msg");
for (int i = 0; i < TestServicesAndMethods.ITERATIONS; i++) {
objectNode.put("msg","hello" + String.valueOf(i));
pubSubService.publish(new CommonServices.IPubSubService.PubSubMsgContext(topic, objectNode.toString()));
}
objectNode.put("state", "end");
pubSubService.publish(new CommonServices.IPubSubService.PubSubMsgContext(topic, objectNode.toString()));
System.out.println("Async publish Test of: " + String.valueOf(ITERATIONS) +" took (msec): " + String.valueOf(System.currentTimeMillis() - start));
Thread.sleep(10000);
pubSubService.shutdown();
}
}
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