end of day commits

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

@@ -222,4 +222,15 @@ public class CommonServices {
         void unsubscribe(String topic);
 
     }
+
+
+    public interface ICacheClient <K,V>{
+        void set(K key, V val);
+        void set(K key, V val, int expiration);
+        void setExpiration(K key, int expiration);
+        V get(K key);
+        void delete(K key);
+        void deleteByPattern(K pattern);
+        V[] getByPattern(K pattern);
+    }
 }

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

@@ -5,6 +5,7 @@ import io.undertow.server.HttpServerExchange;
 import microservice.MicroserviceApp;
 import microservice.common.context.RestContext;
 import microservice.defs.Constants;
+import microservice.handlers.Reactor;
 import microservice.io.iface.*;
 import microservice.params.CommandParams;
 import microservice.params.ZMQParams;
@@ -12,6 +13,11 @@ import microservice.types.BaseRestResponse;
 import org.zeromq.ZMQ;
 import org.zeromq.ZSocket;
 
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 
 /**
@@ -39,26 +45,237 @@ import java.util.function.Consumer;
  *
  *  </p>
  */
-public class IRestServiceZmqImpl extends CommonServices.IRestService implements HttpHandler, IContainer {
+public class IRestServiceZmqImpl extends CommonServices.IRestService implements IContainer {
     static final String MAINT_CHANNEL = "inproc://maint";
 
     private String appName;
-    private String serverBindAddress = null;
     private ZMQParams.ServerParams serverParams = null;
-    private ZSocket serverReceive = null;
-    private ZSocket serverCommandsSend = null; // to send commands to serverReceive
-    private ZSocket clientReceive = null;
-    private ZSocket clientCommandSend = null; // to send commands to clientReceive
-    private Thread serverReceiveThread = null;
-    private Thread clientReceiveThread = null;
-    private ZMQParams.ServerParams clientParams = null;
-    private String clientBindAddress = null;
+    private ZMQParams.ServerParams clientReceiveParams = null;
+    private ZMQParams.ServerParams clientSendParams = null;
+    private CommonServices.ICacheClient<String,CacheEntry> responseCacheClient = null;
+    private int numOfClientWorkers = 0;
+    private int numOfServerWorkers = 0;
+
+    List<ClientWorker> clientWorkerList = new ArrayList<>();
+    IWorker clientReceive = null;
+    IWorker serverReceive = null;
+    IWorker serverWorker = null;
+    IWorker serverReply = null;
+    ClientSend clientSend = null;
+    List<IWorker> allWorkersList = new ArrayList<>();
+
+    public static class CacheEntry {
+        BiConsumer<String,Integer> onResponseFunc;
+        public int cmid;
+    }
+
+    public static abstract class IWorker {
+        protected Thread runThread = null;
+        abstract boolean init();
+        abstract void start();
+        abstract void stop() throws InterruptedException;
+
+    }
+    public static class ClientSend {
+        private ThreadLocal<ZSocket> clientSend = null;
+        public ZSocket getSocket() {
+            return clientSend.get();
+        }
+
+        public ClientSend(final String address) {
+            clientSend = new ThreadLocal<ZSocket>(){
+                @Override protected ZSocket initialValue() {
+                    ZSocket socket = new ZSocket(ZMQ.PUSH);
+                    System.out.println("Connectring to:" + address);
+                    if (socket.connect(address) == false) {
+                        System.err.println("Failed Connecting to:" + address);
+                        socket = null;
+                    }
+                    return socket;
+                }
+            };
+        }
+    }
+
+    public static class ClientWorker extends IWorker {
+        public static final String ADDRESS = "inproc://ClientWorker";
+        private ZSocket pull = null;
+
+
+        @Override
+        boolean init() {
+
+            return true;
+        }
+
+        @Override
+        void start() {
+            runThread = new Thread(() -> {
+
+            });
+            runThread.start();
+
+        }
+
+        @Override
+        void stop() throws InterruptedException {
+            if (runThread != null){
+                ZSocket exitSocket = new ZSocket(ZMQ.PUSH);
+                exitSocket.connect(ADDRESS);
+                exitSocket.send(Constants.EXIT_MSG.getBytes());
+                runThread.join();
+            }
+        }
+    }
+
+    public static class ClientReceive extends IWorker {
+        private ZMQParams.ServerParams zmqParams;
+        private CommonServices.ICacheClient<String,CacheEntry> responseCacheClient = null;
+        private ZSocket pull = null;
+        private ZSocket push = null;
+
+        public ClientReceive(ZMQParams.ServerParams zmqParams, CommonServices.ICacheClient<String, CacheEntry> responseCacheClient) {
+            this.zmqParams = zmqParams;
+            this.responseCacheClient = responseCacheClient;
+        }
+
+
+        @Override
+        boolean init() {
+
+            return true;
+        }
+
+        @Override
+        void start() {
+            runThread = new Thread(() -> {
+
+            });
+            runThread.start();
+        }
+
+        @Override
+        void stop() throws InterruptedException {
+            if (runThread != null){
+                ZSocket exitSocket = new ZSocket(ZMQ.PUSH);
+                exitSocket.connect(zmqParams.bindAddress());
+                exitSocket.send(Constants.EXIT_MSG.getBytes());
+                runThread.join();
+            }
+        }
+    }
+
+    public static class ServerReceive extends IWorker {
+        private ZMQParams.ServerParams zmqParams;
+        private ZSocket pull = null;
+        private ZSocket push = null;
+
+        public ServerReceive(ZMQParams.ServerParams zmqParams) {
+            this.zmqParams = zmqParams;
+        }
+
+
+        @Override
+        boolean init() {
+            push = new ZSocket(ZMQ.PUSH);
+            pull = new ZSocket(ZMQ.PULL);
+            final String bindAddress = zmqParams.bindAddress();
+            if (pull.bind(bindAddress) == false) {
+                System.err.println(this.getClass().getName() + " >> Failed in binding to: " + bindAddress);
+                return false;
+            }
+            push.connect(ServerWorker.ADDRESS);
+
+            return true;
+        }
+
+        @Override
+        void start() {
+            runThread = new Thread(() -> {
+
+            });
+            runThread.start();
+        }
+
+        @Override
+        void stop() throws InterruptedException {
+            if (runThread != null){
+                ZSocket exitSocket = new ZSocket(ZMQ.PUSH);
+                exitSocket.connect(zmqParams.bindAddress());
+                exitSocket.send(Constants.EXIT_MSG.getBytes());
+                runThread.join();
+            }
+        }
+    }
+
+    public static class ServerWorker extends IWorker {
+        public static final String ADDRESS = "inproc://ServerWorker";
+        private ZSocket pull = null;
+        private ZSocket push = null;
+        private CommonServices.IServiceReactor reactor = null;
+
+        public ServerWorker(CommonServices.IServiceReactor reactor) {
+            this.reactor = reactor;
+        }
+
+        @Override
+        boolean init() {
+            push = new ZSocket(ZMQ.PUSH);
+            pull = new ZSocket(ZMQ.PULL);
+
+            return pull.bind(ADDRESS) && push.connect(ServerReply.ADDRESS);
+        }
+
+        @Override
+        void start() {
+            runThread = new Thread(() -> {
+
+            });
+            runThread.start();
+        }
+
+        @Override
+        void stop() throws InterruptedException {
+            if (runThread != null){
+                ZSocket exitSocket = new ZSocket(ZMQ.PUSH);
+                exitSocket.connect(ADDRESS);
+                exitSocket.send(Constants.EXIT_MSG.getBytes());
+                runThread.join();
+            }
+        }
+    }
+
+    public static class ServerReply extends IWorker {
+        public static final String ADDRESS = "inproc://ServerReply";
+        private ZSocket pull = null;
+        private Map<String,ZSocket> connectionsMap = new HashMap<>();
 
-    @Override
-    public void handleRequest(HttpServerExchange exchange) throws Exception {
 
+        @Override
+        boolean init() {
+            return false;
+        }
+
+        @Override
+        void start() {
+            runThread = new Thread(() -> {
+
+            });
+            runThread.start();
+        }
+
+        @Override
+        void stop() throws InterruptedException {
+            if (runThread != null){
+                ZSocket exitSocket = new ZSocket(ZMQ.PUSH);
+                exitSocket.connect(ADDRESS);
+                exitSocket.send(Constants.EXIT_MSG.getBytes());
+                runThread.join();
+            }
+        }
     }
 
+
     @Override
     public BaseRestResponse create(CommandParams cmdParams) {
         return null;
@@ -126,83 +343,64 @@ public class IRestServiceZmqImpl extends CommonServices.IRestService implements
 
         switch (getServiceMode()){
             case E_CLIENT:
+                retstat = allocateClient();
                 break;
             case E_SERVER:
-                retstat = initServer();
+                allocateServer();
                 break;
             case E_CLIENT_SERVER:
-                retstat = initServer() & initClient();
+                allocateServer();
+                retstat = allocateClient();
                 break;
             case E_UNKNOWN:
                 System.err.println(getClass().getName() + " >> unknown service mode");
                 retstat = false;
                 break;
         }
-        return retstat;
-    }
 
-    public boolean initServer() {
-        serverBindAddress = serverParams.bindAddress();
-        serverReceive = new ZSocket(ZMQ.PULL);
-        serverCommandsSend = new ZSocket(ZMQ.PUSH );
-        return serverReceive.bind(serverBindAddress) &&
-                serverReceive.bind(MAINT_CHANNEL) &&
-                serverCommandsSend.connect(MAINT_CHANNEL);
+        return (retstat &&
+                !allWorkersList.isEmpty() &&
+                allWorkersList.stream().allMatch(IWorker::init));
     }
 
-    public boolean initClient() {
+    public void allocateServer() {
+        serverReceive = new ServerReceive(serverParams);
+        allWorkersList.add(serverReceive);
+        for (int i = 0; i < numOfServerWorkers; i++){
+            allWorkersList.add(new ServerWorker(reactor));
+        }
 
-        clientBindAddress = clientParams.bindAddress();
-        clientReceive = new ZSocket(ZMQ.PULL);
-        clientCommandSend = new ZSocket(ZMQ.PUSH);
-        return clientReceive.bind(serverBindAddress) &&
-                clientReceive.bind(MAINT_CHANNEL) &&
-                clientCommandSend.connect(MAINT_CHANNEL);
     }
 
-    @Override
-    public void run() {
-        switch (getServiceMode()){
-            case E_CLIENT:
-                runClientThread();
-                break;
-            case E_SERVER:
-                runServerThread();
-                break;
-            case E_CLIENT_SERVER:
-                runServerThread();
-                runClientThread();
-                break;
+    public boolean allocateClient() {
+        clientReceive = new ClientReceive(clientReceiveParams,responseCacheClient);
+        allWorkersList.add(clientReceive);
+        for (int i = 0; i < numOfClientWorkers; i++){
+            allWorkersList.add(new ClientWorker());
         }
-    }
-
-    private void runServerThread() {
-        serverReceiveThread = new Thread(() -> {
 
-        });
+       /**
+        * init client send
+        */
+        clientSend = new ClientSend(clientSendParams.bindAddress());
+        return clientSend.getSocket() != null;
     }
 
-    private void runClientThread() {
-        clientReceiveThread = new Thread(() -> {
-
-        });
-
+    @Override
+    public void run() {
+        allWorkersList.forEach(IWorker::start);
     }
 
+
     @Override
     public void shutdown() {
-        try {
-        if (clientReceiveThread!= null) {
-            clientCommandSend.send(Constants.EXIT_MSG.getBytes());
-            clientReceiveThread.join();
-        }
-        if (serverReceiveThread != null){
-            serverCommandsSend.send(Constants.EXIT_MSG.getBytes());
-            serverReceiveThread.join();
-        }
-        } catch (InterruptedException e) {
-            System.err.println(e.toString());
-        }
+        allWorkersList.forEach((iWorker) -> {
+            try {
+                iWorker.stop();
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+        });
     }
 
     @Override
@@ -220,7 +418,17 @@ public class IRestServiceZmqImpl extends CommonServices.IRestService implements
         this.serverParams = serverParams;
     }
 
-    public void setClientParams(ZMQParams.ServerParams clientParams) {
-        this.clientParams = clientParams;
+    public void setClientReceiveParams(ZMQParams.ServerParams clientReceiveParams) {
+        this.clientReceiveParams = clientReceiveParams;
+    }
+
+    public void setClientSendParams(ZMQParams.ServerParams clientSendParams) { this.clientSendParams = clientSendParams;  }
+
+    public void setNumOfClientWorkers(int numOfClientWorkers) {
+        this.numOfClientWorkers = numOfClientWorkers;
+    }
+
+    public void setNumOfServerWorkers(int numOfServerWorkers) {
+        this.numOfServerWorkers = numOfServerWorkers;
     }
 }

src/main/java/microservice/utils/ServiceBuilderFactory.java

@@ -119,8 +119,12 @@ public class ServiceBuilderFactory {
 
         CommonServices.EnumRestServiceMode serviceMode = CommonServices.EnumRestServiceMode.E_UNKNOWN;
         ZMQParams.ServerParams serverParams = null;
-        ZMQParams.ServerParams clientParams = null;
+        ZMQParams.ServerParams clientReceiveParams = null;
+        ZMQParams.ServerParams clientSendParams = null;
         IRestServiceZmqImpl restServiceZmq = null;
+        int numOfClientWorkers = 0;
+        int numOfServerWorkers = 0;
+
         public RestServiceZmqBuilder(CommonServices.EnumRestServiceMode serviceMode) {
             this.serviceMode = serviceMode;
         }
@@ -128,8 +132,19 @@ public class ServiceBuilderFactory {
         public void setServerParams(ZMQParams.ServerParams serverParams) {
             this.serverParams = serverParams;
         }
-        public void setClientParams(ZMQParams.ServerParams clientParams) {
-            this.clientParams = clientParams;
+        public void setClientReceiveParams(ZMQParams.ServerParams clientReceiveParams) {
+            this.clientReceiveParams = clientReceiveParams;
+        }
+        public void setClientSendParams(ZMQParams.ServerParams clientSendParams) { this.clientSendParams = clientSendParams; }
+
+        public RestServiceZmqBuilder setNumOfClientWorkers(int num) {
+            numOfClientWorkers = num;
+            return this;
+        }
+
+        public RestServiceZmqBuilder setNumOfServerWorkers(int numOfServerWorkers) {
+            this.numOfServerWorkers = numOfServerWorkers;
+            return this;
         }
 
         private boolean validateParams() {
@@ -141,14 +156,22 @@ public class ServiceBuilderFactory {
                         return false;
                     break;
                 case E_CLIENT:
-                    if (this.clientParams == null)
+                    if (this.clientReceiveParams == null || this.clientSendParams == null)
                         return false;
                     break;
                 case E_CLIENT_SERVER:
-                    if (this.serverParams == null || this.clientParams == null)
+                    if (this.serverParams == null || this.clientReceiveParams == null || this.clientSendParams == null)
                         return false;
                     break;
             }
+            /**
+             * defaulting to number of processors
+             */
+            if (numOfClientWorkers == 0){
+                numOfClientWorkers = Runtime.getRuntime().availableProcessors();
+            }
+            if (numOfServerWorkers == 0)
+                numOfServerWorkers = Runtime.getRuntime().availableProcessors();
             return true;
         }
 
@@ -161,14 +184,14 @@ public class ServiceBuilderFactory {
 
                     switch (serviceMode) {
                         case E_SERVER:
-                            restServiceZmq.setServerParams(serverParams);
+                            buildServer();
                             break;
                         case E_CLIENT:
-                            restServiceZmq.setClientParams(clientParams);
+                            buildClient();
                             break;
                         case E_CLIENT_SERVER:
-                            restServiceZmq.setServerParams(serverParams);
-                            restServiceZmq.setClientParams(clientParams);
+                            buildClient();
+                            buildServer();
                             break;
                     }
                 } catch (Exception exp){
@@ -181,5 +204,16 @@ public class ServiceBuilderFactory {
             return restServiceZmq;
         }
 
+        private void buildServer() {
+            restServiceZmq.setServerParams(serverParams);
+            restServiceZmq.setNumOfServerWorkers(numOfServerWorkers);
+        }
+
+        private void buildClient() {
+            restServiceZmq.setClientReceiveParams(clientReceiveParams);
+            restServiceZmq.setClientSendParams(clientSendParams);
+            restServiceZmq.setNumOfClientWorkers(numOfClientWorkers);
+        }
+
     }
 }