summaryrefslogtreecommitdiffstats
path: root/sca-java-2.x/trunk/modules/endpoint-hazelcast
diff options
context:
space:
mode:
Diffstat (limited to 'sca-java-2.x/trunk/modules/endpoint-hazelcast')
-rw-r--r--sca-java-2.x/trunk/modules/endpoint-hazelcast/src/main/java/org/apache/tuscany/sca/endpoint/hazelcast/HazelcastEndpointRegistry.java133
-rw-r--r--sca-java-2.x/trunk/modules/endpoint-hazelcast/src/test/java/org/apache/tuscany/sca/endpoint/hazelcast/MultiRegTestCase.java183
2 files changed, 265 insertions, 51 deletions
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 a65c1a9c23..f9ec30011e 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
@@ -38,14 +38,19 @@ import org.apache.tuscany.sca.runtime.RuntimeEndpoint;
import com.hazelcast.config.Config;
import com.hazelcast.config.TcpIpConfig;
import com.hazelcast.config.XmlConfigBuilder;
+import com.hazelcast.core.EntryEvent;
+import com.hazelcast.core.EntryListener;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.core.IMap;
+import com.hazelcast.core.MembershipEvent;
+import com.hazelcast.core.MembershipListener;
import com.hazelcast.nio.Address;
/**
* An EndpointRegistry using a Hazelcast
*/
-public class HazelcastEndpointRegistry implements EndpointRegistry, LifeCycleListener {
+public class HazelcastEndpointRegistry implements EndpointRegistry, LifeCycleListener, EntryListener<String, Endpoint>, MembershipListener {
private final static Logger logger = Logger.getLogger(HazelcastEndpointRegistry.class.getName());
private List<EndpointReference> endpointreferences = new CopyOnWriteArrayList<EndpointReference>();
@@ -54,14 +59,14 @@ public class HazelcastEndpointRegistry implements EndpointRegistry, LifeCycleLis
private ExtensionPointRegistry registry;
private ConfigURI configURI;
- private HazelcastInstance hazelcastInstance;
- private Map<Object, Object> map;
+ HazelcastInstance hazelcastInstance;
+ Map<Object, Object> map;
private List<String> localEndpoints = new ArrayList<String>();;
public HazelcastEndpointRegistry(ExtensionPointRegistry registry,
- Map<String, String> attributes,
- String domainRegistryURI,
- String domainURI) {
+ Map<String, String> attributes,
+ String domainRegistryURI,
+ String domainURI) {
this.registry = registry;
this.configURI = new ConfigURI(domainRegistryURI);
}
@@ -74,22 +79,27 @@ public class HazelcastEndpointRegistry implements EndpointRegistry, LifeCycleLis
map = new HashMap<Object, Object>();
} else {
initHazelcastInstance();
- map = hazelcastInstance.getMap(configURI.getDomainName() + "Endpoints");
+ IMap imap = hazelcastInstance.getMap(configURI.getDomainName() + "/Endpoints");
+ imap.addEntryListener(this, true);
+ map = imap;
+ hazelcastInstance.getCluster().addMembershipListener(this);
}
}
public void stop() {
if (hazelcastInstance != null) {
hazelcastInstance.shutdown();
+ hazelcastInstance = null;
+ map = null;
}
}
- private void initHazelcastInstance() {
+ 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();
@@ -107,6 +117,8 @@ public class HazelcastEndpointRegistry implements EndpointRegistry, LifeCycleLis
config.getNetworkConfig().getJoin().getMulticastConfig().setMulticastGroup(configURI.getMulticastAddress());
}
+ // config.getMapConfig(configURI.getDomainName() + "/Endpoints").setBackupCount(0);
+
if (configURI.getRemotes().size() > 0) {
TcpIpConfig tcpconfig = config.getNetworkConfig().getJoin().getJoinMembers();
tcpconfig.setEnabled(true);
@@ -200,15 +212,15 @@ public class HazelcastEndpointRegistry implements EndpointRegistry, LifeCycleLis
endpoint.setRemote(true);
}
// if (!entry.isPrimary()) {
- ((RuntimeEndpoint) endpoint).bind(registry, this);
+ ((RuntimeEndpoint)endpoint).bind(registry, this);
// }
foundEndpoints.add(endpoint);
logger.fine("Found endpoint with matching service - " + endpoint);
- }
+ }
// else the service name doesn't match
}
}
-
+
return foundEndpoints;
}
@@ -252,46 +264,65 @@ public class HazelcastEndpointRegistry implements EndpointRegistry, LifeCycleLis
}
public void updateEndpoint(String uri, Endpoint endpoint) {
-// // TODO: is updateEndpoint needed?
-// throw new UnsupportedOperationException();
+ // // TODO: is updateEndpoint needed?
+ // throw new UnsupportedOperationException();
+ }
+
+ public void entryAdded(EntryEvent<String, Endpoint> event) {
+ entryAdded(event.getKey(), event.getValue());
}
-// public void entryAdded(Object key, Object value) {
-// MapEntry entry = (MapEntry)value;
-// Endpoint newEp = (Endpoint)entry.getValue();
-// if (!isLocal(entry)) {
-// logger.info(id + " Remote endpoint added: " + entry.getValue());
-// newEp.setRemote(true);
-// }
-// ((RuntimeEndpoint) newEp).bind(registry, this);
-// for (EndpointListener listener : listeners) {
-// listener.endpointAdded(newEp);
-// }
-// }
-//
-// public void entryRemoved(Object key, Object value) {
-// MapEntry entry = (MapEntry)value;
-// if (!isLocal(entry)) {
-// logger.info(id + " Remote endpoint removed: " + entry.getValue());
-// }
-// Endpoint oldEp = (Endpoint)entry.getValue();
-// for (EndpointListener listener : listeners) {
-// listener.endpointRemoved(oldEp);
-// }
-// }
-//
-// public void entryUpdated(Object key, Object oldValue, Object newValue) {
-// MapEntry oldEntry = (MapEntry)oldValue;
-// MapEntry newEntry = (MapEntry)newValue;
-// if (!isLocal(newEntry)) {
-// logger.info(id + " Remote endpoint updated: " + newEntry.getValue());
-// }
-// Endpoint oldEp = (Endpoint)oldEntry.getValue();
-// Endpoint newEp = (Endpoint)newEntry.getValue();
-// ((RuntimeEndpoint) newEp).bind(registry, this);
-// for (EndpointListener listener : listeners) {
-// listener.endpointUpdated(oldEp, newEp);
-// }
-// }
+ public void entryEvicted(EntryEvent<String, Endpoint> event) {
+ // Should not happen
+ }
+
+ public void entryRemoved(EntryEvent<String, Endpoint> event) {
+ entryRemoved(event.getKey(), event.getValue());
+ }
+
+ public void entryUpdated(EntryEvent<String, Endpoint> event) {
+ entryUpdated(event.getKey(), null, event.getValue());
+ }
+
+ public void entryAdded(Object key, Object value) {
+ Endpoint newEp = (Endpoint)value;
+ if (!isLocal(newEp)) {
+ logger.info(" Remote endpoint added: " + newEp);
+ newEp.setRemote(true);
+ }
+ ((RuntimeEndpoint)newEp).bind(registry, this);
+ for (EndpointListener listener : listeners) {
+ listener.endpointAdded(newEp);
+ }
+ }
+
+ public void entryRemoved(Object key, Object value) {
+ Endpoint oldEp = (Endpoint)value;
+ if (!isLocal(oldEp)) {
+ logger.info(" Remote endpoint removed: " + value);
+ }
+ ((RuntimeEndpoint) oldEp).bind(registry, this);
+ for (EndpointListener listener : listeners) {
+ listener.endpointRemoved(oldEp);
+ }
+ }
+
+ public void entryUpdated(Object key, Object oldValue, Object newValue) {
+ Endpoint oldEp = (Endpoint)oldValue;
+ Endpoint newEp = (Endpoint)newValue;
+ if (!isLocal(newEp)) {
+ logger.info(" Remote endpoint updated: " + newEp);
+ }
+ ((RuntimeEndpoint)newEp).bind(registry, this);
+ for (EndpointListener listener : listeners) {
+ listener.endpointUpdated(oldEp, newEp);
+ }
+ }
+
+ public void memberAdded(MembershipEvent event) {
+ }
+
+ public void memberRemoved(MembershipEvent event) {
+ }
}
diff --git a/sca-java-2.x/trunk/modules/endpoint-hazelcast/src/test/java/org/apache/tuscany/sca/endpoint/hazelcast/MultiRegTestCase.java b/sca-java-2.x/trunk/modules/endpoint-hazelcast/src/test/java/org/apache/tuscany/sca/endpoint/hazelcast/MultiRegTestCase.java
new file mode 100644
index 0000000000..32e49dbe41
--- /dev/null
+++ b/sca-java-2.x/trunk/modules/endpoint-hazelcast/src/test/java/org/apache/tuscany/sca/endpoint/hazelcast/MultiRegTestCase.java
@@ -0,0 +1,183 @@
+/*
+ * 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.util.HashMap;
+import java.util.Map;
+
+import org.apache.tuscany.sca.assembly.AssemblyFactory;
+import org.apache.tuscany.sca.assembly.Binding;
+import org.apache.tuscany.sca.assembly.Component;
+import org.apache.tuscany.sca.assembly.Endpoint;
+import org.apache.tuscany.sca.assembly.SCABindingFactory;
+import org.apache.tuscany.sca.core.DefaultExtensionPointRegistry;
+import org.apache.tuscany.sca.core.ExtensionPointRegistry;
+import org.apache.tuscany.sca.core.FactoryExtensionPoint;
+import org.apache.tuscany.sca.runtime.EndpointListener;
+import org.apache.tuscany.sca.runtime.RuntimeEndpoint;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import com.hazelcast.core.IMap;
+
+// Ignore so its not run in the build yet till its working
+@Ignore("Hazelcast doesn't support the map entry management by members")
+public class MultiRegTestCase implements EndpointListener {
+ private static ExtensionPointRegistry extensionPoints;
+ private static AssemblyFactory assemblyFactory;
+ private static SCABindingFactory scaBindingFactory;
+
+ @BeforeClass
+ public static void init() {
+ extensionPoints = new DefaultExtensionPointRegistry();
+ FactoryExtensionPoint factories = extensionPoints.getExtensionPoint(FactoryExtensionPoint.class);
+ assemblyFactory = factories.getFactory(AssemblyFactory.class);
+ scaBindingFactory = factories.getFactory(SCABindingFactory.class);
+ }
+
+ @Test
+ public void testReplication() throws Exception {
+ RuntimeEndpoint ep1 = createEndpoint("ep1uri");
+
+ String host = InetAddress.getLocalHost().getHostAddress();
+ String bind = null; // "9.65.158.31";
+ String port1 = "8085";
+ String port2 = "8086";
+ String port3 = "8087";
+ String range = "1";
+
+ Map<String, String> attrs1 = new HashMap<String, String>();
+ // attrs1.put("nomcast", "true");
+ attrs1.put("bind", bind);
+ attrs1.put("receiverPort", port1);
+ attrs1.put("receiverAutoBind", range);
+ // attrs1.put("routes", host + ":" + port2 + " " + host + ":" + port3);
+ HazelcastEndpointRegistry reg1 = new HazelcastEndpointRegistry(extensionPoints, attrs1, "tuscany:foo", "bar");
+ reg1.addListener(this);
+ reg1.start();
+
+ Map<String, String> attrs2 = new HashMap<String, String>();
+ // attrs2.put("nomcast", "true");
+ attrs1.put("bind", bind);
+ attrs2.put("receiverPort", port2);
+ attrs2.put("receiverAutoBind", range);
+ // attrs2.put("routes", host + ":"+port1);
+ HazelcastEndpointRegistry reg2 = new HazelcastEndpointRegistry(extensionPoints, attrs2, "tuscany:foo", "bar");
+ reg2.addListener(this);
+ reg2.start();
+
+ Map<String, String> attrs3 = new HashMap<String, String>();
+ // attrs3.put("nomcast", "true");
+ attrs1.put("bind", bind);
+ attrs3.put("receiverPort", port3);
+ attrs3.put("receiverAutoBind", range);
+ // attrs3.put("routes", host + ":"+port1);
+ HazelcastEndpointRegistry reg3 = new HazelcastEndpointRegistry(extensionPoints, attrs3, "tuscany:foo", "bar");
+ reg3.addListener(this);
+ reg3.start();
+
+ ep1.bind(extensionPoints, reg1);
+ reg1.addEndpoint(ep1);
+ assertExists(reg1, "ep1uri");
+ assertExists(reg2, "ep1uri");
+ assertExists(reg3, "ep1uri");
+
+ RuntimeEndpoint ep2 = createEndpoint("ep2uri");
+ ep2.bind(extensionPoints, reg2);
+ reg2.addEndpoint(ep2);
+ assertExists(reg2, "ep2uri");
+ assertExists(reg1, "ep2uri");
+ assertExists(reg3, "ep2uri");
+
+ System.out.println(((IMap)reg1.map).localKeySet().size());
+ System.out.println(((IMap)reg2.map).localKeySet().size());
+ System.out.println(((IMap)reg3.map).localKeySet().size());
+
+ reg1.stop();
+ Thread.sleep(6000);
+ Assert.assertNull(reg2.getEndpoint("ep1uri"));
+ Assert.assertNull(reg3.getEndpoint("ep1uri"));
+
+ System.out.println(((IMap)reg2.map).localKeySet().size());
+ System.out.println(((IMap)reg3.map).localKeySet().size());
+
+ assertExists(reg2, "ep2uri");
+ assertExists(reg3, "ep2uri");
+
+ reg1.start();
+ ep1.bind(extensionPoints, reg1);
+ reg1.addEndpoint(ep1);
+ assertExists(reg1, "ep1uri");
+ assertExists(reg2, "ep1uri");
+ assertExists(reg3, "ep1uri");
+
+ reg1.stop();
+ reg2.stop();
+ reg3.stop();
+ System.out.println(); // closed
+ }
+
+ private Endpoint assertExists(HazelcastEndpointRegistry reg, String uri) throws InterruptedException {
+ Endpoint ep = null;
+ int count = 0;
+ while (ep == null && count < 15) {
+ ep = reg.getEndpoint(uri);
+ if (ep == null) {
+ Thread.sleep(1000);
+ System.out.println(reg + ": tries=" + count);
+ }
+ count++;
+ }
+ Assert.assertNotNull(ep);
+ Assert.assertEquals(uri, ep.getURI());
+ return ep;
+ }
+
+ private RuntimeEndpoint createEndpoint(String uri) {
+ RuntimeEndpoint ep = (RuntimeEndpoint) assemblyFactory.createEndpoint();
+ Component comp = assemblyFactory.createComponent();
+ ep.setComponent(comp);
+ ep.setService(assemblyFactory.createComponentService());
+ Binding b = scaBindingFactory.createSCABinding();
+ ep.setBinding(b);
+ ep.setURI(uri);
+ return ep;
+ }
+
+ private void print(String prefix, Endpoint ep) {
+ System.out.println(prefix + ": "+ep);
+ }
+
+ public void endpointAdded(Endpoint endpoint) {
+ print("Added", endpoint);
+ }
+
+ public void endpointRemoved(Endpoint endpoint) {
+ print("Removed", endpoint);
+ }
+
+ public void endpointUpdated(Endpoint oldEndpoint, Endpoint newEndpoint) {
+ print("Updated", newEndpoint);
+ }
+
+}