From 9da25a6532fcbb9ae66294f1ff9ea903653c83e5 Mon Sep 17 00:00:00 2001 From: rfeng Date: Tue, 19 Jan 2010 05:37:38 +0000 Subject: Expose system definitions from the deployer Add the removal of entries when the member leaves the group Add listeners to the hazelcastInstance git-svn-id: http://svn.us.apache.org/repos/asf/tuscany@900661 13f79535-47bb-0310-9956-ffa450edef68 --- .../hazelcast/HazelcastEndpointRegistry.java | 133 +++++++++------ .../sca/endpoint/hazelcast/MultiRegTestCase.java | 183 +++++++++++++++++++++ 2 files changed, 265 insertions(+), 51 deletions(-) create mode 100644 sca-java-2.x/trunk/modules/endpoint-hazelcast/src/test/java/org/apache/tuscany/sca/endpoint/hazelcast/MultiRegTestCase.java (limited to 'sca-java-2.x/trunk/modules/endpoint-hazelcast/src') 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, MembershipListener { private final static Logger logger = Logger.getLogger(HazelcastEndpointRegistry.class.getName()); private List endpointreferences = new CopyOnWriteArrayList(); @@ -54,14 +59,14 @@ public class HazelcastEndpointRegistry implements EndpointRegistry, LifeCycleLis private ExtensionPointRegistry registry; private ConfigURI configURI; - private HazelcastInstance hazelcastInstance; - private Map map; + HazelcastInstance hazelcastInstance; + Map map; private List localEndpoints = new ArrayList();; public HazelcastEndpointRegistry(ExtensionPointRegistry registry, - Map attributes, - String domainRegistryURI, - String domainURI) { + Map 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(); } 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 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 event) { + // Should not happen + } + + public void entryRemoved(EntryEvent event) { + entryRemoved(event.getKey(), event.getValue()); + } + + public void entryUpdated(EntryEvent 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 attrs1 = new HashMap(); + // 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 attrs2 = new HashMap(); + // 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 attrs3 = new HashMap(); + // 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); + } + +} -- cgit v1.2.3