diff options
Diffstat (limited to '')
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;
+ }
+}
|