Update hazelcast registry to use runtime properties

git-svn-id: http://svn.us.apache.org/repos/asf/tuscany@937890 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 258 insertions, 57 deletions

diff --git a/sca-java-2.x/trunk/modules/endpoint-hazelcast/src/main/java/org/apache/tuscany/sca/endpoint/hazelcast/HazelcastDomainRegistryFactory.java b/sca-java-2.x/trunk/modules/endpoint-hazelcast/src/main/java/org/apache/tuscany/sca/endpoint/hazelcast/HazelcastDomainRegistryFactory.java
index a4e2da8be3..110982a1bd 100644
--- a/sca-java-2.x/trunk/modules/endpoint-hazelcast/src/main/java/org/apache/tuscany/sca/endpoint/hazelcast/HazelcastDomainRegistryFactory.java
+++ b/sca-java-2.x/trunk/modules/endpoint-hazelcast/src/main/java/org/apache/tuscany/sca/endpoint/hazelcast/HazelcastDomainRegistryFactory.java
@@ -19,9 +19,13 @@
 
 package org.apache.tuscany.sca.endpoint.hazelcast;
 
+import java.util.Properties;
+
 import org.apache.tuscany.sca.core.ExtensionPointRegistry;
+import org.apache.tuscany.sca.core.UtilityExtensionPoint;
 import org.apache.tuscany.sca.runtime.BaseDomainRegistryFactory;
 import org.apache.tuscany.sca.runtime.EndpointRegistry;
+import org.apache.tuscany.sca.runtime.RuntimeProperties;
 
 /**
  * The utility responsible for finding the endpoint regstry by the scheme and creating instances for the
@@ -30,17 +34,13 @@ import org.apache.tuscany.sca.runtime.EndpointRegistry;
 public class HazelcastDomainRegistryFactory extends BaseDomainRegistryFactory {
     private final static String[] schemes = new String[] {"multicast", "wka", "tuscany", "hazelcast"};
 
-    /**
-     * @param extensionRegistry
-     */
     public HazelcastDomainRegistryFactory(ExtensionPointRegistry registry) {
         super(registry);
     }
 
     protected EndpointRegistry createEndpointRegistry(String endpointRegistryURI, String domainURI) {
-        EndpointRegistry endpointRegistry =
-            new HazelcastEndpointRegistry(registry, null, endpointRegistryURI, domainURI);
-        return endpointRegistry;
+        Properties properties = registry.getExtensionPoint(UtilityExtensionPoint.class).getUtility(RuntimeProperties.class).getProperties();
+        return new HazelcastEndpointRegistry(registry, properties, domainURI);
     }
 
     public String[] getSupportedSchemes() {
diff --git a/sca-java-2.x/trunk/modules/endpoint-hazelcast/src/main/java/org/apache/tuscany/sca/endpoint/hazelcast/HazelcastEndpointRegistry.java b/sca-java-2.x/trunk/modules/endpoint-hazelcast/src/main/java/org/apache/tuscany/sca/endpoint/hazelcast/HazelcastEndpointRegistry.java
index 10b2362bd7..420bcfa3a2 100644
--- a/sca-java-2.x/trunk/modules/endpoint-hazelcast/src/main/java/org/apache/tuscany/sca/endpoint/hazelcast/HazelcastEndpointRegistry.java
+++ b/sca-java-2.x/trunk/modules/endpoint-hazelcast/src/main/java/org/apache/tuscany/sca/endpoint/hazelcast/HazelcastEndpointRegistry.java
@@ -19,12 +19,13 @@
 

 package org.apache.tuscany.sca.endpoint.hazelcast;

 

+import java.io.FileNotFoundException;

 import java.net.UnknownHostException;

 import java.util.ArrayList;

 import java.util.Collection;

-import java.util.HashMap;

 import java.util.List;

 import java.util.Map;

+import java.util.Properties;

 import java.util.concurrent.ConcurrentHashMap;

 import java.util.logging.Logger;

 

@@ -33,10 +34,11 @@ import org.apache.tuscany.sca.assembly.Endpoint;
 import org.apache.tuscany.sca.core.ExtensionPointRegistry;

 import org.apache.tuscany.sca.core.FactoryExtensionPoint;

 import org.apache.tuscany.sca.core.LifeCycleListener;

+import org.apache.tuscany.sca.core.UtilityExtensionPoint;

 import org.apache.tuscany.sca.runtime.BaseEndpointRegistry;

-import org.apache.tuscany.sca.runtime.DomainRegistryURI;

 import org.apache.tuscany.sca.runtime.EndpointRegistry;

 import org.apache.tuscany.sca.runtime.RuntimeEndpoint;

+import org.apache.tuscany.sca.runtime.RuntimeProperties;

 import org.oasisopen.sca.ServiceRuntimeException;

 

 import com.hazelcast.config.Config;

@@ -62,22 +64,27 @@ import com.hazelcast.nio.Address;
 public class HazelcastEndpointRegistry extends BaseEndpointRegistry implements EndpointRegistry, LifeCycleListener, EntryListener<String, Endpoint>, MembershipListener {

     private final static Logger logger = Logger.getLogger(HazelcastEndpointRegistry.class.getName());

 

-    protected DomainRegistryURI configURI;

-

     private HazelcastInstance hazelcastInstance;

     protected Map<Object, Object> endpointMap;

-    private Map<String, Endpoint> localEndpoints = new ConcurrentHashMap<String, Endpoint>();

+    protected Map<String, Endpoint> localEndpoints = new ConcurrentHashMap<String, Endpoint>();

     protected MultiMap<String, String> endpointOwners;

-    private AssemblyFactory assemblyFactory;

-    private Object shutdownMutex = new Object();

+    protected AssemblyFactory assemblyFactory;

+    protected Object shutdownMutex = new Object();

+    protected Properties properties;

+

+    public HazelcastEndpointRegistry(ExtensionPointRegistry registry, Properties properties, String domainURI) {

+        super(registry, null, null, domainURI);

+        this.assemblyFactory = registry.getExtensionPoint(FactoryExtensionPoint.class).getFactory(AssemblyFactory.class);

+        this.properties = registry.getExtensionPoint(UtilityExtensionPoint.class).getUtility(RuntimeProperties.class).getProperties();

+    }

 

     public HazelcastEndpointRegistry(ExtensionPointRegistry registry,

                                      Map<String, String> attributes,

                                      String domainRegistryURI,

                                      String domainURI) {

         super(registry, attributes, domainRegistryURI, domainURI);

-        this.configURI = new DomainRegistryURI(domainRegistryURI);

         this.assemblyFactory = registry.getExtensionPoint(FactoryExtensionPoint.class).getFactory(AssemblyFactory.class);

+        this.properties = registry.getExtensionPoint(UtilityExtensionPoint.class).getUtility(RuntimeProperties.class).getProperties();

     }

     

     public HazelcastInstance getHazelcastInstance() {

@@ -88,18 +95,18 @@ public class HazelcastEndpointRegistry extends BaseEndpointRegistry implements E
         if (endpointMap != null) {

             throw new IllegalStateException("The registry has already been started");

         }

-        if (configURI.toString().startsWith("tuscany:vm:")) {

-            endpointMap = new HashMap<Object, Object>();

-        } else {

+//        if (configURI.toString().startsWith("tuscany:vm:")) {

+//            endpointMap = new HashMap<Object, Object>();

+//        } else {

             initHazelcastInstance();

-            IMap imap = hazelcastInstance.getMap(configURI.getDomainName() + "/Endpoints");

+            IMap imap = hazelcastInstance.getMap(domainURI + "/Endpoints");

             imap.addEntryListener(this, true);

             endpointMap = imap;

             

-            endpointOwners = hazelcastInstance.getMultiMap(configURI.getDomainName() + "/EndpointOwners");

+            endpointOwners = hazelcastInstance.getMultiMap(domainURI + "/EndpointOwners");

 

             hazelcastInstance.getCluster().addMembershipListener(this);

-        }

+//        }

     }

 

     public void stop() {

@@ -114,45 +121,12 @@ public class HazelcastEndpointRegistry extends BaseEndpointRegistry implements E
     }

 

     private void initHazelcastInstance() {

-        Config config = new XmlConfigBuilder().build();

-

-        config.setPort(configURI.getListenPort());

-        //config.setPortAutoIncrement(false);

-

-        if (configURI.getBindAddress() != null) {

-            config.getNetworkConfig().getInterfaces().setEnabled(true);

-            config.getNetworkConfig().getInterfaces().clear();

-            config.getNetworkConfig().getInterfaces().addInterface(configURI.getBindAddress());

-        }

-

-        config.getGroupConfig().setName(configURI.getDomainName());

-        config.getGroupConfig().setPassword(configURI.getPassword());

+        Config config = getHazelcastConfig();

 

-        if (configURI.isMulticastDisabled()) {

-            config.getNetworkConfig().getJoin().getMulticastConfig().setEnabled(false);

-        } else {

-            config.getNetworkConfig().getJoin().getMulticastConfig().setEnabled(true);

-            config.getNetworkConfig().getJoin().getMulticastConfig().setMulticastPort(configURI.getMulticastPort());

-            config.getNetworkConfig().getJoin().getMulticastConfig().setMulticastGroup(configURI.getMulticastAddress());

-        }

-        

+        // do this when theres a way to have adders be the key owners

         // config.getMapConfig(configURI.getDomainName() + "/Endpoints").setBackupCount(0);

 

-        if (configURI.getRemotes().size() > 0) {

-            TcpIpConfig tcpconfig = config.getNetworkConfig().getJoin().getTcpIpConfig();

-            tcpconfig.setEnabled(true);

-            List<Address> lsMembers = tcpconfig.getAddresses();

-            lsMembers.clear();

-            for (String addr : configURI.getRemotes()) {

-                String[] ipNPort = addr.split(":");

-                try {

-                    lsMembers.add(new Address(ipNPort[0], Integer.parseInt(ipNPort[1])));

-                } catch (UnknownHostException e) {

-                    throw new RuntimeException(e);

-                }

-            }

-        }

-        

+        // this caches reads locally

         config.getMapConfig("default").setNearCacheConfig(new NearCacheConfig(0, 0, "NONE", 0, true));

 

         // Disable the Hazelcast shutdown hook as Tuscany has its own and with both there are race conditions

@@ -169,10 +143,61 @@ public class HazelcastEndpointRegistry extends BaseEndpointRegistry implements E
         this.hazelcastInstance = Hazelcast.newHazelcastInstance(config);

     }

 

+    protected Config getHazelcastConfig() {

+        Config config;

+        this.properties = registry.getExtensionPoint(UtilityExtensionPoint.class).getUtility(RuntimeProperties.class).getProperties();

+        String configFile = properties.getProperty("hazelcastConfig");

+        if (configFile != null) {

+            try {

+                config = new XmlConfigBuilder(configFile).build();

+            } catch (FileNotFoundException e) {

+                throw new IllegalArgumentException(configFile, e);

+            }

+        } else {

+            config = new XmlConfigBuilder().build();

+            RegistryConfig rc = new RegistryConfig(properties);

+            config.setPort(rc.getBindPort());

+            //config.setPortAutoIncrement(false);

+

+            if (!rc.getBindAddress().equals("*")) {

+                config.getNetworkConfig().getInterfaces().setEnabled(true);

+                config.getNetworkConfig().getInterfaces().clear();

+                config.getNetworkConfig().getInterfaces().addInterface(rc.getBindAddress());

+            }

+

+            config.getGroupConfig().setName(rc.getUserid());

+            config.getGroupConfig().setPassword(rc.getPassword());

+

+            if (rc.isMulticastDisabled()) {

+                config.getNetworkConfig().getJoin().getMulticastConfig().setEnabled(false);

+            } else {

+                config.getNetworkConfig().getJoin().getMulticastConfig().setEnabled(true);

+                config.getNetworkConfig().getJoin().getMulticastConfig().setMulticastPort(rc.getMulticastPort());

+                config.getNetworkConfig().getJoin().getMulticastConfig().setMulticastGroup(rc.getMulticastAddress());

+            }

+            

+            if (rc.getWKAs().size() > 0) {

+                TcpIpConfig tcpconfig = config.getNetworkConfig().getJoin().getTcpIpConfig();

+                tcpconfig.setEnabled(true);

+                List<Address> lsMembers = tcpconfig.getAddresses();

+                lsMembers.clear();

+                for (String addr : rc.getWKAs()) {

+                    String[] ipNPort = addr.split(":");

+                    try {

+                        lsMembers.add(new Address(ipNPort[0], Integer.parseInt(ipNPort[1])));

+                    } catch (UnknownHostException e) {

+                        throw new RuntimeException(e);

+                    }

+                }

+            }

+        }

+        return config;

+    }

+

     public void addEndpoint(Endpoint endpoint) {

         if (findEndpoint(endpoint.getURI()).size() > 0) {

             Member m = getOwningMember(endpoint.getURI());

-            throw new IllegalStateException("Endpoint " + endpoint.getURI() + " already exists in domain " + configURI.getDomainName() + " at " + m.getInetSocketAddress());

+            throw new IllegalStateException("Endpoint " + endpoint.getURI() + " already exists in domain " + domainURI + " at " + m.getInetSocketAddress());

         }

             

         String localMemberAddr = hazelcastInstance.getCluster().getLocalMember().getInetSocketAddress().toString();

diff --git a/sca-java-2.x/trunk/modules/endpoint-hazelcast/src/main/java/org/apache/tuscany/sca/endpoint/hazelcast/RegistryConfig.java b/sca-java-2.x/trunk/modules/endpoint-hazelcast/src/main/java/org/apache/tuscany/sca/endpoint/hazelcast/RegistryConfig.java
new file mode 100644
index 0000000000..da604d717b
--- /dev/null
+++ b/sca-java-2.x/trunk/modules/endpoint-hazelcast/src/main/java/org/apache/tuscany/sca/endpoint/hazelcast/RegistryConfig.java
@@ -0,0 +1,176 @@
+/*

+ * Licensed to the Apache Software Foundation (ASF) under one

+ * or more contributor license agreements.  See the NOTICE file

+ * distributed with this work for additional information

+ * regarding copyright ownership.  The ASF licenses this file

+ * to you under the Apache License, Version 2.0 (the

+ * "License"); you may not use this file except in compliance

+ * with the License.  You may obtain a copy of the License at

+ * 

+ *   http://www.apache.org/licenses/LICENSE-2.0

+ * 

+ * Unless required by applicable law or agreed to in writing,

+ * software distributed under the License is distributed on an

+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

+ * KIND, either express or implied.  See the License for the

+ * specific language governing permissions and limitations

+ * under the License.    

+ */

+

+package org.apache.tuscany.sca.endpoint.hazelcast;

+

+import java.net.InetAddress;

+import java.net.NetworkInterface;

+import java.util.ArrayList;

+import java.util.Enumeration;

+import java.util.List;

+import java.util.Properties;

+

+

+/**

+ * Utility to parse the config properties.

+ * 

+ * bind - ip[:port] - defines the local bind address and port, it defaults to the network interface with the widest 

+ *                    class (ie class A is wider than class B) on port 14820 and if that port in use it will try 

+ *                    incrementing by one till a free port is found.

+ *             

+ * multicast - groupip:port | off - defines if multicast discovery is used and if so what multicast ip group and port is used 

+ *             defaults to 224.5.12.10:51482. A value of off means multicast is disabled.

+ *             

+ * wka - ip[:port] - a comma separated list of ip address and port for remote nodes in the domain group. The port defaults to 14820.

+ *             

+ * userid -  is the userid other nodes must use to connect to this domain group. The default is the default domain name.

+ * 

+ * password -  is the password other nodes must use to connect to this domain group. The default is 'tuscany'.

+ *             

+ * client - true means this is an SCAClient call           

+ *             

+ */

+public class RegistryConfig {

+    

+    private String bindAddress;

+    private int bindPort = 14820;

+    private boolean multicastDisabled = false;

+    private String multicastAddress = "224.5.12.10";

+    private int multicastPort = 51482;

+    private List<String> wkas = new ArrayList<String>();

+    private String userid;

+    private String password;

+    boolean client;

+    

+    public RegistryConfig(Properties properties) {

+        init(properties);

+    }

+

+    private void init(Properties properties) {

+   

+        String bindValue = properties.getProperty("bind");

+        if (bindValue == null) {

+            this.bindAddress = chooseLocalAddress().getHostAddress();

+        } else {

+            if (bindValue.indexOf(":") == -1) {

+                this.bindAddress = bindValue;

+            } else {

+                String[] addr = bindValue.split(":");

+                this.bindAddress = addr[0];

+                this.bindPort = Integer.parseInt(addr[1]);

+            }

+        }

+

+        String multicastValue = properties.getProperty("multicast");

+        if (multicastValue != null) {

+            if ("off".equalsIgnoreCase(multicastValue)) {

+                this.multicastDisabled = true;

+            } else {

+                if (multicastValue.indexOf(":") == -1) {

+                    this.multicastAddress = multicastValue;

+                } else {

+                    String[] addr = multicastValue.split(":");

+                    this.multicastAddress = addr[0];

+                    this.multicastPort = Integer.parseInt(addr[1]);

+                }

+            }

+        }

+        

+        String wkaValue = properties.getProperty("wka");

+        if (wkaValue != null) {

+            String[] ips = wkaValue.split(",");

+            for (String ip : ips) {

+                if (ip.indexOf(":") == -1) {

+                    wkas.add(ip + ":14820");

+                } else {

+                    wkas.add(ip);

+                }

+            }

+        }

+

+        this.client = Boolean.parseBoolean(properties.getProperty("client", "false"));

+        this.password = properties.getProperty("password", "tuscany");

+        this.userid = properties.getProperty("userid", properties.getProperty("defaultDomainName", "default"));

+

+    }

+

+    public String getBindAddress() {

+        return bindAddress;

+    }

+

+    public int getBindPort() {

+        return bindPort;

+    }

+

+    public boolean isMulticastDisabled() {

+        return multicastDisabled;

+    }

+

+    public String getMulticastAddress() {

+        return multicastAddress;

+    }

+

+    public int getMulticastPort() {

+        return multicastPort;

+    }

+

+    public List<String> getWKAs() {

+        return wkas;

+    }

+    

+    public String getUserid() {

+        return userid;

+    }

+    public String getPassword() {

+        return password;

+    }

+

+    /**

+     * Use the NIC address with the widest class, ie class A instead of class B or C.

+     * Bit crude but in a lot of environments a class A address (eg 10.x.x.x) is likely

+     * a better choice than a class C address (eg 192.x.x.x). And the alternative to 

+     * this is to just choose the first address of the first network interface which 

+     * likely isn't a better choice than this approach.

+     */

+    protected InetAddress chooseLocalAddress() {

+        InetAddress chosen = null;

+        try {

+            Enumeration<NetworkInterface> nis = NetworkInterface.getNetworkInterfaces();

+            while (nis.hasMoreElements()) {

+                NetworkInterface ni = nis.nextElement();

+                Enumeration<InetAddress> ips = ni.getInetAddresses();

+                while (ips.hasMoreElements()) {

+                    InetAddress addr = ips.nextElement();

+                    if (!addr.isLoopbackAddress()) {

+                        if (chosen == null) {

+                            chosen = addr;

+                        } else {

+                            if (((int) addr.getAddress()[0] & 0xFF) < ((int) chosen.getAddress()[0] & 0xFF)) {

+                                chosen = addr;

+                            }

+                        }

+                    }

+                }

+            }

+        } catch (Exception e) {

+            // ignore

+        }

+        return chosen;

+    }

+}