starting state

DeshanKTD · Sep 21, 2017 · 53d44de · 53d44de
1 parent 7f1137f
commit 53d44de
Show file tree

Hide file tree

Showing 3 changed files with 62 additions and 55 deletions.
diff --git a/...src/main/java/lambda/netty/loadbalancer/core/SysService/SysServiceHostResolveHandler.java b/...src/main/java/lambda/netty/loadbalancer/core/SysService/SysServiceHostResolveHandler.java
@@ -89,11 +89,18 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
                 if (stateImpl.getState() == InstanceStates.DOWN) {
                     logger.info("No instance is up ! informing Sys-service ");
                     requestIp();
+                } else if (stateImpl.getState() == InstanceStates.STARTING) {
+                    String state = EtcdUtil.setWatcher(instanceID);
+                    State tmp = StateImplJsonHelp.getObject(state);
+                    String remoteIp = tmp.getHosts().peek();
+                    ProxyEvent proxyEvent = new ProxyEvent(remoteIp);
+                    proxyEvent.setDomain(instanceID);
+                    ctx.fireUserEventTriggered(proxyEvent);
                 } else if (stateImpl.getState() == InstanceStates.RUNNING) {
                     logger.info("These instances are up and running");
                     String remoteIp = LoadBalanceUtil.getRemoteHost(stateImpl);
                     try {
-                        EtcdUtil.putValue("localhost", StateImplJsonHelp.toString(stateImpl));
+                        EtcdUtil.putValue(instanceID, StateImplJsonHelp.toString(stateImpl));
                     } catch (EtcdClientException e) {
                         logger.error("Cannot connect to ETCD !", e);
                     }

diff --git a/lambdanettyloadbalancer/src/main/java/lambda/netty/loadbalancer/core/etcd/EtcdUtil.java b/lambdanettyloadbalancer/src/main/java/lambda/netty/loadbalancer/core/etcd/EtcdUtil.java
@@ -126,44 +126,41 @@ public static void txngetValue(String s_key, long version) {
 //        System.out.println(txnResponse.getGetResponses());
     }
 
-//    public static void main(String[] args) throws ExecutionException, InterruptedException {
-//        Watch watch = null;
-//        Watch.Watcher watcher = null;
-//
-//        try {
-//            watch = client.getWatchClient();
-//            watcher = watch.watch(ByteSequence.fromString("aaa"));
-//
-//            for (int i = 0; i < 5; i++) {
-//                WatchResponse response = watcher.listen();
-//
-//                for (WatchEvent event : response.getEvents()) {
-//
-//                    System.out.println(event.getEventType());
-//                    System.out.println(Optional.ofNullable(event.getKeyValue().getKey())
-//                            .map(ByteSequence::toStringUtf8)
-//                            .orElse(""));
-//                    System.out.println(Optional.ofNullable(event.getKeyValue().getValue())
-//                            .map(ByteSequence::toStringUtf8)
-//                            .orElse(""));
-//
-//                }
-//            }
-//        } catch (Exception e) {
-//            if (watcher != null) {
-//                watcher.close();
-//            }
-//
-//            if (watch != null) {
-//                watch.close();
-//            }
-//
-//            if (client != null) {
-//                client.close();
-//            }
-//        }
-//
-//    }
-//
+    public static String setWatcher(String key) {
+        Watch watch = null;
+        Watch.Watcher watcher = null;
+        String result=null;
+        try {
+            watch = client.getWatchClient();
+            watcher = watch.watch(ByteSequence.fromString(key));
+            WatchResponse response = watcher.listen();
+            WatchEvent event = response.getEvents().get(0);
+            logger.info("Watcher event for key: " + key + " action: " + event.getEventType());
+            result = Optional.ofNullable(event.getKeyValue().getValue())
+                    .map(ByteSequence::toStringUtf8)
+                    .orElse("");
+
+
+        } catch (Exception e) {
+            if (watcher != null) {
+                watcher.close();
+            }
+
+            if (watch != null) {
+                watch.close();
+            }
+
+            if (client != null) {
+                client.close();
+            }
+        } finally {
+            watcher.close();
+            watch.close();
+        }
+        return result;
+    }
+
+
+
 
 }
diff --git a/lambdanettyloadbalancer/src/main/java/lambda/netty/loadbalancer/core/launch/Launcher.java b/lambdanettyloadbalancer/src/main/java/lambda/netty/loadbalancer/core/launch/Launcher.java
@@ -27,6 +27,8 @@
 import lambda.netty.loadbalancer.core.Server;
 import lambda.netty.loadbalancer.core.SysService.SysServiceConnection;
 import lambda.netty.loadbalancer.core.SysService.SysServiceHandlersInit;
+import lambda.netty.loadbalancer.core.etcd.EtcdClientException;
+import lambda.netty.loadbalancer.core.etcd.EtcdUtil;
 import lambda.netty.loadbalancer.core.scalability.ScalabilityManager;
 import org.apache.commons.configuration2.XMLConfiguration;
 import org.apache.commons.configuration2.builder.fluent.Configurations;
@@ -99,7 +101,7 @@ public static long getLong(String s) {
         return xmlConfiguration.getLong(s);
     }
 
-    public static void main(String[] args) throws InterruptedException {
+    public static void main(String[] args) throws InterruptedException, EtcdClientException {
 
 //        State  state = new StateImpl();
 //        state.pushHost("127.0.0.1:8082");
@@ -117,21 +119,22 @@ public static void main(String[] args) throws InterruptedException {
 //        } catch (ExecutionException e) {
 //            e.printStackTrace();
 //        }
+//
+//        try {
+//            ConfigLogger.printFields();
+//        } catch (Exception e) {
+//            logger.error("Cannot print Configurations !", e);
+//        }
+//
+//
+//        if (SCALABILITY_ENABLED) {
+//            service.submit(new ScalabilityManager());
+//        } else {
+//            logger.info("Scalability is not enabled !");
+//        }
+//        service.submit(new Server());
 
-        try {
-            ConfigLogger.printFields();
-        } catch (Exception e) {
-            logger.error("Cannot print Configurations !", e);
-        }
-
-
-        if (SCALABILITY_ENABLED) {
-            service.submit(new ScalabilityManager());
-        } else {
-            logger.info("Scalability is not enabled !");
-        }
-        service.submit(new Server());
-
+        EtcdUtil.putValue("aaa","klk");
     }
 
 }