diff options
Diffstat (limited to '')
12 files changed, 547 insertions, 166 deletions
diff --git a/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/assembly/impl/RuntimeEndpointImpl.java b/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/assembly/impl/RuntimeEndpointImpl.java index 3dafaedf35..3e73960367 100644 --- a/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/assembly/impl/RuntimeEndpointImpl.java +++ b/sca-java-2.x/trunk/modules/core/src/main/java/org/apache/tuscany/sca/core/assembly/impl/RuntimeEndpointImpl.java @@ -508,8 +508,12 @@ public class RuntimeEndpointImpl extends EndpointImpl implements RuntimeEndpoint bind(compositeContext); } } - RuntimeEndpointImpl ep = (RuntimeEndpointImpl)serializer.readEndpoint(xml); - copyFrom(ep); + if (serializer != null) { + RuntimeEndpointImpl ep = (RuntimeEndpointImpl)serializer.readEndpoint(xml); + copyFrom(ep); + } else { + // FIXME: [rfeng] What should we do here? + } } super.resolve(); } diff --git a/sca-java-2.x/trunk/modules/deployment/src/main/java/org/apache/tuscany/sca/deployment/Deployer.java b/sca-java-2.x/trunk/modules/deployment/src/main/java/org/apache/tuscany/sca/deployment/Deployer.java index b0b08cd49e..6d62446ccb 100644 --- a/sca-java-2.x/trunk/modules/deployment/src/main/java/org/apache/tuscany/sca/deployment/Deployer.java +++ b/sca-java-2.x/trunk/modules/deployment/src/main/java/org/apache/tuscany/sca/deployment/Deployer.java @@ -42,6 +42,7 @@ import org.apache.tuscany.sca.contribution.processor.ContributionWriteException; import org.apache.tuscany.sca.contribution.processor.ProcessorContext; import org.apache.tuscany.sca.core.ExtensionPointRegistry; import org.apache.tuscany.sca.core.LifeCycleListener; +import org.apache.tuscany.sca.definitions.Definitions; import org.apache.tuscany.sca.monitor.Monitor; /** @@ -186,14 +187,8 @@ public interface Deployer extends LifeCycleListener { */ ExtensionPointRegistry getExtensionPointRegistry(); - /* - * @see org.apache.tuscany.sca.core.LifeCycleListener#start() - */ - void start(); - - /* - * @see org.apache.tuscany.sca.core.LifeCycleListener#stop() + /** + * Get the system definitions */ - void stop(); - + Definitions getSystemDefinitions(); } diff --git a/sca-java-2.x/trunk/modules/deployment/src/main/java/org/apache/tuscany/sca/deployment/impl/DeployerImpl.java b/sca-java-2.x/trunk/modules/deployment/src/main/java/org/apache/tuscany/sca/deployment/impl/DeployerImpl.java index b012fd795d..e4340102fc 100644 --- a/sca-java-2.x/trunk/modules/deployment/src/main/java/org/apache/tuscany/sca/deployment/impl/DeployerImpl.java +++ b/sca-java-2.x/trunk/modules/deployment/src/main/java/org/apache/tuscany/sca/deployment/impl/DeployerImpl.java @@ -50,8 +50,8 @@ import org.apache.tuscany.sca.contribution.ContributionFactory; import org.apache.tuscany.sca.contribution.DefaultImport; import org.apache.tuscany.sca.contribution.Export; import org.apache.tuscany.sca.contribution.Import; -import org.apache.tuscany.sca.contribution.namespace.NamespaceImport; import org.apache.tuscany.sca.contribution.java.JavaImport; +import org.apache.tuscany.sca.contribution.namespace.NamespaceImport; import org.apache.tuscany.sca.contribution.processor.ContributionReadException; import org.apache.tuscany.sca.contribution.processor.ContributionResolveException; import org.apache.tuscany.sca.contribution.processor.ContributionWriteException; @@ -568,4 +568,9 @@ public class DeployerImpl implements Deployer { this.schemaValidationEnabled = schemaValidationEnabled; } + public Definitions getSystemDefinitions() { + init(); + return systemDefinitions; + } + } 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); + } + +} diff --git a/sca-java-2.x/trunk/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/AbstractReplicatedMap.java b/sca-java-2.x/trunk/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/AbstractReplicatedMap.java index 47b3450594..fe683af025 100644 --- a/sca-java-2.x/trunk/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/AbstractReplicatedMap.java +++ b/sca-java-2.x/trunk/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/AbstractReplicatedMap.java @@ -708,7 +708,7 @@ public abstract class AbstractReplicatedMap extends MapStore implements RpcCallb // [rfeng] Change the behavior to replicate to all nodes if (entry.isPrimary() && self.equals(entry.getPrimary())) { try { - Member[] backup = publishEntryInfo(entry.getKey(), entry.getValue(), entry.getBackupNodes()); + Member[] backup = publishEntryInfo(entry.getKey(), entry.getValue()); entry.setBackupNodes(backup); entry.setPrimary(self); } catch (ChannelException x) { @@ -768,7 +768,7 @@ public abstract class AbstractReplicatedMap extends MapStore implements RpcCallb if (log.isDebugEnabled()) log.debug("[1] Primary choosing a new backup"); try { - Member[] backup = publishEntryInfo(entry.getKey(), entry.getValue(), entry.getBackupNodes()); + Member[] backup = publishEntryInfo(entry.getKey(), entry.getValue()); entry.setBackupNodes(backup); entry.setPrimary(channel.getLocalMember(false)); } catch (ChannelException x) { @@ -798,7 +798,7 @@ public abstract class AbstractReplicatedMap extends MapStore implements RpcCallb entry.setPrimary(channel.getLocalMember(false)); entry.setBackup(false); entry.setProxy(false); - Member[] backup = publishEntryInfo(entry.getKey(), entry.getValue(), entry.getBackupNodes()); + Member[] backup = publishEntryInfo(entry.getKey(), entry.getValue()); entry.setBackupNodes(backup); if (mapOwner != null) mapOwner.objectMadePrimay(entry.getKey(), entry.getValue()); @@ -833,7 +833,7 @@ public abstract class AbstractReplicatedMap extends MapStore implements RpcCallb return members[node]; } - protected abstract Member[] publishEntryInfo(Object key, Object value, Member[] backupNodes) throws ChannelException; + protected abstract Member[] publishEntryInfo(Object key, Object value) throws ChannelException; public void heartbeat() { try { @@ -916,7 +916,7 @@ public abstract class AbstractReplicatedMap extends MapStore implements RpcCallb } if (entry.isBackup()) { //select a new backup node - backup = publishEntryInfo(key, entry.getValue(), entry.getBackupNodes()); + backup = publishEntryInfo(key, entry.getValue()); } else if (entry.isProxy()) { //invalidate the previous primary msg = @@ -997,7 +997,7 @@ public abstract class AbstractReplicatedMap extends MapStore implements RpcCallb old = remove(key); try { if (notify) { - Member[] backup = publishEntryInfo(key, value, entry.getBackupNodes()); + Member[] backup = publishEntryInfo(key, value); entry.setBackupNodes(backup); } } catch (ChannelException x) { diff --git a/sca-java-2.x/trunk/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/ReplicatedEndpointRegistry.java b/sca-java-2.x/trunk/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/ReplicatedEndpointRegistry.java index 0c2144ea49..42d2eda6a3 100644 --- a/sca-java-2.x/trunk/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/ReplicatedEndpointRegistry.java +++ b/sca-java-2.x/trunk/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/ReplicatedEndpointRegistry.java @@ -38,11 +38,13 @@ import java.util.logging.Logger; import org.apache.catalina.tribes.Channel; import org.apache.catalina.tribes.ChannelException; +import org.apache.catalina.tribes.ChannelReceiver; import org.apache.catalina.tribes.Member; import org.apache.catalina.tribes.group.GroupChannel; import org.apache.catalina.tribes.group.interceptors.StaticMembershipInterceptor; import org.apache.catalina.tribes.membership.McastService; import org.apache.catalina.tribes.membership.StaticMember; +import org.apache.catalina.tribes.transport.ReceiverBase; import org.apache.tuscany.sca.assembly.Endpoint; import org.apache.tuscany.sca.assembly.EndpointReference; import org.apache.tuscany.sca.core.ExtensionPointRegistry; @@ -67,6 +69,9 @@ public class ReplicatedEndpointRegistry implements EndpointRegistry, LifeCycleLi private String address = MULTICAST_ADDRESS; private String bind = null; private int timeout = 50; + private int receiverPort = 4000; + private int receiverAutoBind = 100; + private List<URI> staticRoutes; private final static String DEFAULT_DOMAIN_URI = "http://tuscany.apache.org/sca/1.1/domains/default"; private String domainURI = DEFAULT_DOMAIN_URI; @@ -75,7 +80,6 @@ public class ReplicatedEndpointRegistry implements EndpointRegistry, LifeCycleLi private ExtensionPointRegistry registry; private ReplicatedMap map; - private static List<URI> staticRoutes; private String id; private boolean noMultiCast; @@ -175,6 +179,14 @@ public class ReplicatedEndpointRegistry implements EndpointRegistry, LifeCycleLi if (mcast != null) { noMultiCast = Boolean.valueOf(mcast); } + String recvPort = attributes.get("receiverPort"); + if (recvPort != null) { + receiverPort = Integer.parseInt(recvPort); + } + String recvAutoBind = attributes.get("receiverAutoBind"); + if (recvAutoBind != null) { + receiverAutoBind = Integer.parseInt(recvAutoBind); + } } public void start() { @@ -190,9 +202,24 @@ public class ReplicatedEndpointRegistry implements EndpointRegistry, LifeCycleLi if (noMultiCast) { map.getChannel().addInterceptor(new DisableMcastInterceptor()); } - - // http://www.mail-archive.com/users@tomcat.apache.org/msg24873.html - int port = channel.getChannelReceiver().getPort(); + + // Configure the receiver ports + ChannelReceiver receiver = channel.getChannelReceiver(); + if (receiver instanceof ReceiverBase) { + ((ReceiverBase)receiver).setAutoBind(receiverAutoBind); + ((ReceiverBase)receiver).setPort(receiverPort); + } + + /* + Object sender = channel.getChannelSender(); + if (sender instanceof ReplicationTransmitter) { + sender = ((ReplicationTransmitter)sender).getTransport(); + } + if (sender instanceof AbstractSender) { + ((AbstractSender)sender).setKeepAliveCount(0); + ((AbstractSender)sender).setMaxRetryAttempts(5); + } + */ if (staticRoutes != null) { StaticMembershipInterceptor smi = new StaticMembershipInterceptor(); @@ -200,12 +227,12 @@ public class ReplicatedEndpointRegistry implements EndpointRegistry, LifeCycleLi Member member; try { // The port has to match the receiver port - member = new StaticMember(staticRoute.getHost(), port, 5000); + member = new StaticMember(staticRoute.getHost(), staticRoute.getPort(), 5000); } catch (IOException e) { throw new RuntimeException(e); } smi.addStaticMember(member); - logger.info("Added static route: " + staticRoute.getHost() + ":" + port); + logger.info("Added static route: " + staticRoute.getHost() + ":" + staticRoute.getPort()); } smi.setLocalMember(map.getChannel().getLocalMember(false)); map.getChannel().addInterceptor(smi); @@ -410,6 +437,7 @@ public class ReplicatedEndpointRegistry implements EndpointRegistry, LifeCycleLi logger.info(id + " Remote endpoint removed: " + entry.getValue()); } Endpoint oldEp = (Endpoint)entry.getValue(); + ((RuntimeEndpoint) oldEp).bind(registry, this); for (EndpointListener listener : listeners) { listener.endpointRemoved(oldEp); } diff --git a/sca-java-2.x/trunk/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/ReplicatedMap.java b/sca-java-2.x/trunk/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/ReplicatedMap.java index 762407604d..669ad82192 100644 --- a/sca-java-2.x/trunk/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/ReplicatedMap.java +++ b/sca-java-2.x/trunk/modules/endpoint-tribes/src/main/java/org/apache/tuscany/sca/endpoint/tribes/ReplicatedMap.java @@ -17,6 +17,8 @@ package org.apache.tuscany.sca.endpoint.tribes; import java.io.Serializable; +import java.util.Iterator; +import java.util.Map; import org.apache.catalina.tribes.Channel; import org.apache.catalina.tribes.ChannelException; @@ -97,29 +99,65 @@ public class ReplicatedMap extends AbstractReplicatedMap implements RpcCallback, * @return Member - the backup node * @throws ChannelException */ - protected Member[] publishEntryInfo(Object key, Object value, Member[] backupNodes) throws ChannelException { + protected Member[] publishEntryInfo(Object key, Object value) throws ChannelException { if (!(key instanceof Serializable && value instanceof Serializable)) return new Member[0]; //select a backup node - Member[] backup = getMapMembers(); + Member[] members = getMapMembers(); - if (backup == null || backup.length == 0) - return null; - - // Set the receivers to these members that are not in the backup nodes yet - Member[] members = backup; - if (backupNodes != null) { - members = getMapMembersExcl(backupNodes); + if (members == null || members.length == 0) { + return new Member[0]; } //publish the data out to all nodes MapMessage msg = new MapMessage(getMapContextName(), MapMessage.MSG_COPY, false, (Serializable)key, (Serializable)value, - null, channel.getLocalMember(false), backup); + null, channel.getLocalMember(false), members); getChannel().send(members, msg, getChannelSendOptions()); - return backup; + return members; + } + + /** + * Override the base method to look up existing entries only + */ + public Object get(Object key) { + MapEntry entry = super.getInternal(key); + if (log.isTraceEnabled()) + log.trace("Requesting id:" + key + " entry:" + entry); + if (entry == null) { + return null; + } + return entry.getValue(); } + /** + * Override the base method to remove all entries owned by the member that disappeared + */ + public void memberDisappeared(Member member) { + boolean removed = false; + synchronized (mapMembers) { + removed = (mapMembers.remove(member) != null); + if (!removed) { + if (log.isDebugEnabled()) + log.debug("Member[" + member + "] disappeared, but was not present in the map."); + return; //the member was not part of our map. + } + } + + Iterator<Map.Entry<Object, Object>> i = super.entrySetFull().iterator(); + while (i.hasNext()) { + Map.Entry<Object, Object> e = i.next(); + MapEntry entry = (MapEntry)super.getInternal(e.getKey()); + if (entry == null) { + continue; + } + if (member.equals(entry.getPrimary())) { + if (log.isDebugEnabled()) + log.debug("[2] Primary disappeared"); + i.remove(); + } //end if + } //while + } } diff --git a/sca-java-2.x/trunk/modules/endpoint-tribes/src/test/java/org/apache/tuscany/sca/endpoint/tribes/MultiRegTestCase.java b/sca-java-2.x/trunk/modules/endpoint-tribes/src/test/java/org/apache/tuscany/sca/endpoint/tribes/MultiRegTestCase.java index d329ebd066..a470c47ba0 100644 --- a/sca-java-2.x/trunk/modules/endpoint-tribes/src/test/java/org/apache/tuscany/sca/endpoint/tribes/MultiRegTestCase.java +++ b/sca-java-2.x/trunk/modules/endpoint-tribes/src/test/java/org/apache/tuscany/sca/endpoint/tribes/MultiRegTestCase.java @@ -19,6 +19,7 @@ package org.apache.tuscany.sca.endpoint.tribes;
+import java.net.InetAddress;
import java.util.HashMap;
import java.util.Map;
@@ -28,109 +29,141 @@ 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.Ignore;
+import org.junit.BeforeClass;
import org.junit.Test;
// Ignore so its not run in the build yet till its working
-@Ignore
-public class MultiRegTestCase {
-
-// @Test
-// public void testTwoNodesMultiCast() throws InterruptedException {
-// DefaultExtensionPointRegistry extensionPoints = new DefaultExtensionPointRegistry();
-// FactoryExtensionPoint factories = extensionPoints.getExtensionPoint(FactoryExtensionPoint.class);
-// AssemblyFactory assemblyFactory = factories.getFactory(AssemblyFactory.class);
-//
-// ReplicatedEndpointRegistry reg1 = new ReplicatedEndpointRegistry(extensionPoints, null, "foo", "bar");
-// reg1.start();
-//
-// Endpoint ep1 = assemblyFactory.createEndpoint();
-// Component comp = assemblyFactory.createComponent();
-// ep1.setComponent(comp);
-// ep1.setService(assemblyFactory.createComponentService());
-// Binding b = new SCABindingFactoryImpl().createSCABinding();
-// ep1.setBinding(b);
-// ep1.setURI("ep1uri");
-// reg1.addEndpoint(ep1);
-//
-// Endpoint ep1p = reg1.getEndpoint("ep1uri");
-// Assert.assertNotNull(ep1p);
-// Assert.assertEquals("ep1uri", ep1p.getURI());
-//
-// ReplicatedEndpointRegistry reg2 = new ReplicatedEndpointRegistry(extensionPoints, null, "foo", "bar");
-// reg2.start();
-// Thread.sleep(5000);
-//
-// Endpoint ep1p2 = reg2.getEndpoint("ep1uri");
-// Assert.assertNotNull(ep1p2);
-// Assert.assertEquals("ep1uri", ep1p2.getURI());
-//
-// reg1.stop();
-// reg2.stop();
-// }
+public class MultiRegTestCase implements EndpointListener {
+ private static ExtensionPointRegistry extensionPoints;
+ private static AssemblyFactory assemblyFactory;
+ private static SCABindingFactory scaBindingFactory;
- @Test
- public void testTwoNodesStaticNoMultiCast() throws InterruptedException {
- DefaultExtensionPointRegistry extensionPoints = new DefaultExtensionPointRegistry();
+ @BeforeClass
+ public static void init() {
+ extensionPoints = new DefaultExtensionPointRegistry();
FactoryExtensionPoint factories = extensionPoints.getExtensionPoint(FactoryExtensionPoint.class);
- AssemblyFactory assemblyFactory = factories.getFactory(AssemblyFactory.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("routes", "9.167.197.91:4001 9.167.197.91:4002");
+ // attrs1.put("nomcast", "true");
+ attrs1.put("bind", bind);
+ attrs1.put("receiverPort", port1);
+ attrs1.put("receiverAutoBind", range);
+ // attrs1.put("routes", host + ":" + port2 + " " + host + ":" + port3);
ReplicatedEndpointRegistry reg1 = new ReplicatedEndpointRegistry(extensionPoints, attrs1, "foo", "bar");
+ reg1.addListener(this);
reg1.start();
- Endpoint ep1 = assemblyFactory.createEndpoint();
- Component comp = assemblyFactory.createComponent();
- ep1.setComponent(comp);
- ep1.setService(assemblyFactory.createComponentService());
- Binding b = factories.getFactory(SCABindingFactory.class).createSCABinding();
- ep1.setBinding(b);
- ep1.setURI("ep1uri");
- reg1.addEndpoint(ep1);
-
- Endpoint ep1p = reg1.getEndpoint("ep1uri");
- Assert.assertNotNull(ep1p);
- Assert.assertEquals("ep1uri", ep1p.getURI());
-
Map<String, String> attrs2 = new HashMap<String, String>();
- attrs2.put("nomcast", "true");
- attrs2.put("routes", "9.167.197.91:4000");
+ // attrs2.put("nomcast", "true");
+ attrs1.put("bind", bind);
+ attrs2.put("receiverPort", port2);
+ attrs2.put("receiverAutoBind", range);
+ // attrs2.put("routes", host + ":"+port1);
ReplicatedEndpointRegistry reg2 = new ReplicatedEndpointRegistry(extensionPoints, attrs2, "foo", "bar");
+ reg2.addListener(this);
reg2.start();
-
- System.out.println("wait");
- Thread.sleep(10000);
- System.out.println("run");
-
- Endpoint ep1p2 = reg2.getEndpoint("ep1uri");
- Assert.assertNotNull(ep1p2);
- Assert.assertEquals("ep1uri", ep1p2.getURI());
Map<String, String> attrs3 = new HashMap<String, String>();
- attrs3.put("nomcast", "true");
- attrs3.put("routes", "9.167.197.91:4000");
+ // attrs3.put("nomcast", "true");
+ attrs1.put("bind", bind);
+ attrs3.put("receiverPort", port3);
+ attrs3.put("receiverAutoBind", range);
+ // attrs3.put("routes", host + ":"+port1);
ReplicatedEndpointRegistry reg3 = new ReplicatedEndpointRegistry(extensionPoints, attrs3, "foo", "bar");
+ reg3.addListener(this);
reg3.start();
-
- System.out.println("wait");
- Thread.sleep(5000);
- System.out.println("run");
- Endpoint ep1p3 = reg3.getEndpoint("ep1uri");
- Assert.assertNotNull(ep1p3);
- Assert.assertEquals("ep1uri", ep1p3.getURI());
+ 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");
+ reg1.stop();
+ Thread.sleep(6000);
+ Assert.assertNull(reg2.getEndpoint("ep1uri"));
+ Assert.assertNull(reg3.getEndpoint("ep1uri"));
+ assertExists(reg2, "ep2uri");
+ assertExists(reg3, "ep2uri");
+
+ reg1.start();
+ ep1.bind(extensionPoints, reg1);
+ reg1.addEndpoint(ep1);
+ assertExists(reg1, "ep1uri");
+ assertExists(reg2, "ep1uri");
+ assertExists(reg3, "ep1uri");
- System.out.println("wait2");
- Thread.sleep(5000);
- System.out.println("end");
reg1.stop();
reg2.stop();
reg3.stop();
+ System.out.println(); // closed
+ }
+
+ private Endpoint assertExists(ReplicatedEndpointRegistry reg, String uri) throws InterruptedException {
+ Endpoint ep = null;
+ int count = 0;
+ while (ep == null && count < 15) {
+ ep = reg.getEndpoint(uri);
+ Thread.sleep(1000);
+ count++;
+ System.out.println(reg + ": tries=" + 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);
}
}
diff --git a/sca-java-2.x/trunk/modules/node-impl-osgi/src/main/java/org/apache/tuscany/sca/osgi/remoteserviceadmin/impl/EndpointHelper.java b/sca-java-2.x/trunk/modules/node-impl-osgi/src/main/java/org/apache/tuscany/sca/osgi/remoteserviceadmin/impl/EndpointHelper.java index 949d2d8af7..8ecc5f7ea8 100644 --- a/sca-java-2.x/trunk/modules/node-impl-osgi/src/main/java/org/apache/tuscany/sca/osgi/remoteserviceadmin/impl/EndpointHelper.java +++ b/sca-java-2.x/trunk/modules/node-impl-osgi/src/main/java/org/apache/tuscany/sca/osgi/remoteserviceadmin/impl/EndpointHelper.java @@ -69,6 +69,30 @@ public class EndpointHelper { if (serviceID != null) { props.put(RemoteConstants.ENDPOINT_SERVICE_ID, Long.parseLong(serviceID)); } + + // FIXME: [rfeng] We need to calculate the intents supported by this endpoint + /* + QName bindingTypeName = endpoint.getBinding().getType(); + Definitions definitions = null; + if(definitions!=null) { + for(BindingType bindingType: definitions.getBindingTypes()) { + if(bindingType.getType().equals(bindingTypeName)) { + bindingType.getAlwaysProvidedIntents(); + } + } + */ + + String intents = (String)props.get(RemoteConstants.SERVICE_EXPORTED_INTENTS); + String extraIntents = (String)props.get(RemoteConstants.SERVICE_EXPORTED_INTENTS_EXTRA); + if (intents == null) { + intents = ""; + } + if (extraIntents != null) { + intents = intents + " " + extraIntents; + } + + props.put(RemoteConstants.SERVICE_INTENTS, intents.trim()); + props.put(RemoteConstants.ENDPOINT_ID, endpoint.getURI()); // FIXME: [rfeng] How to pass in the remote service id from the endpoint XML props.put(RemoteConstants.SERVICE_IMPORTED_CONFIGS, new String[] {"org.osgi.sca"}); diff --git a/sca-java-2.x/trunk/modules/node-impl-osgi/src/main/java/org/apache/tuscany/sca/osgi/remoteserviceadmin/impl/RemoteServiceAdminImpl.java b/sca-java-2.x/trunk/modules/node-impl-osgi/src/main/java/org/apache/tuscany/sca/osgi/remoteserviceadmin/impl/RemoteServiceAdminImpl.java index 608c74bcfc..f4521cfbfc 100644 --- a/sca-java-2.x/trunk/modules/node-impl-osgi/src/main/java/org/apache/tuscany/sca/osgi/remoteserviceadmin/impl/RemoteServiceAdminImpl.java +++ b/sca-java-2.x/trunk/modules/node-impl-osgi/src/main/java/org/apache/tuscany/sca/osgi/remoteserviceadmin/impl/RemoteServiceAdminImpl.java @@ -27,6 +27,12 @@ import java.util.Hashtable; import java.util.List; import java.util.Map; +import org.apache.tuscany.sca.core.ExtensionPointRegistry; +import org.apache.tuscany.sca.core.UtilityExtensionPoint; +import org.apache.tuscany.sca.definitions.Definitions; +import org.apache.tuscany.sca.deployment.Deployer; +import org.apache.tuscany.sca.policy.BindingType; +import org.apache.tuscany.sca.policy.Intent; import org.osgi.framework.Bundle; import org.osgi.framework.BundleContext; import org.osgi.framework.Constants; @@ -73,10 +79,28 @@ public class RemoteServiceAdminImpl implements RemoteServiceAdmin, ManagedServic importer.start(); Hashtable<String, Object> props = new Hashtable<String, Object>(); props.put(RemoteConstants.REMOTE_CONFIGS_SUPPORTED, new String[] {"org.osgi.sca"}); + + ExtensionPointRegistry registry = exporter.getExtensionPointRegistry(); + UtilityExtensionPoint utilities = registry.getExtensionPoint(UtilityExtensionPoint.class); + Deployer deployer = utilities.getUtility(Deployer.class); + Definitions definitions = deployer.getSystemDefinitions(); + + String[] intents = new String[definitions.getIntents().size()]; + int i = 0; + for (Intent intent : definitions.getIntents()) { + intents[i++] = intent.toString(); + } + + String[] bindingTypes = new String[definitions.getBindingTypes().size()]; + i = 0; + for (BindingType bindingType : definitions.getBindingTypes()) { + bindingTypes[i++] = bindingType.getType().toString(); + } + // FIXME: We should ask SCA domain for the supported intents - props.put(RemoteConstants.REMOTE_INTENTS_SUPPORTED, new String[] {}); + props.put(RemoteConstants.REMOTE_INTENTS_SUPPORTED, intents); // FIXME: We should ask SCA domain for the supported binding types - props.put("org.osgi.sca.binding.types", new String[] {}); + props.put("org.osgi.sca.binding.types", bindingTypes); registration = context.registerService(RemoteServiceAdmin.class.getName(), this, props); props = new Hashtable<String, Object>(); diff --git a/sca-java-2.x/trunk/modules/node-impl-osgi/src/main/java/org/apache/tuscany/sca/osgi/service/discovery/impl/DomainDiscoveryService.java b/sca-java-2.x/trunk/modules/node-impl-osgi/src/main/java/org/apache/tuscany/sca/osgi/service/discovery/impl/DomainDiscoveryService.java index f6e9855556..a7c6d04ee9 100644 --- a/sca-java-2.x/trunk/modules/node-impl-osgi/src/main/java/org/apache/tuscany/sca/osgi/service/discovery/impl/DomainDiscoveryService.java +++ b/sca-java-2.x/trunk/modules/node-impl-osgi/src/main/java/org/apache/tuscany/sca/osgi/service/discovery/impl/DomainDiscoveryService.java @@ -53,6 +53,19 @@ public class DomainDiscoveryService extends AbstractDiscoveryService implements this.domainRegistryFactory = registry.getExtensionPoint(UtilityExtensionPoint.class).getUtility(DomainRegistryFactory.class); domainRegistryFactory.addListener(this); + + // [rfeng] Starting of the endpoint registry takes a long time and it leaves the bundle + // state to be starting. When the registry is started, remote endpoints come in and that + // triggers the classloading from this bundle. + Thread thread = new Thread() { + public void run() { + startEndpointRegistry(); + } + }; + thread.start(); + } + + private void startEndpointRegistry() { // The following code forced the start() of the domain registry in absense of services String domainRegistry = context.getProperty("org.osgi.sca.domain.registry"); if (domainRegistry == null) { @@ -103,6 +116,8 @@ public class DomainDiscoveryService extends AbstractDiscoveryService implements { // Notify the endpoint listeners EndpointDescription description = createEndpointDescription(bundleContext, endpoint); + // Set the owning bundle to runtime bundle to avoid NPE + servicesInfo.put(description, context.getBundle()); endpointChanged(description, ADDED); } } @@ -115,6 +130,7 @@ public class DomainDiscoveryService extends AbstractDiscoveryService implements */ { EndpointDescription description = createEndpointDescription(context, endpoint); + servicesInfo.remove(description); endpointChanged(description, REMOVED); } } |