initial version of endpoint dht register

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

5 files changed, 422 insertions, 0 deletions

diff --git a/java/sca/modules/binding-rmi-runtime/META-INF/MANIFEST.MF b/java/sca/modules/binding-rmi-runtime/META-INF/MANIFEST.MF
index 21b3d0f6e1..687f5e68ad 100644
--- a/java/sca/modules/binding-rmi-runtime/META-INF/MANIFEST.MF
+++ b/java/sca/modules/binding-rmi-runtime/META-INF/MANIFEST.MF
@@ -30,3 +30,4 @@ Import-Package: net.sf.cglib.core,
 Bundle-SymbolicName: org.apache.tuscany.sca.binding.rmi.runtime

 Bundle-DocURL: http://www.apache.org/

 Bundle-RequiredExecutionEnvironment: J2SE-1.5,JavaSE-1.6
+Require-Bundle: org.objectweb.asm

diff --git a/java/sca/modules/endpoint-dht/META-INF/MANIFEST.MF b/java/sca/modules/endpoint-dht/META-INF/MANIFEST.MF
new file mode 100644
index 0000000000..82533cda1b
--- /dev/null
+++ b/java/sca/modules/endpoint-dht/META-INF/MANIFEST.MF
@@ -0,0 +1,19 @@
+Manifest-Version: 1.0

+Private-Package: org.apache.tuscany.sca.xsd.impl;version="2.0.0"

+Tool: Bnd-0.0.255

+Bundle-Name: Apache Tuscany SCA Tomcat Tribes Based EndPoint Registry

+Created-By: 1.6.0_07 (Sun Microsystems Inc.)

+Bundle-Vendor: The Apache Software Foundation

+Bundle-Version: 2.0.0

+Bnd-LastModified: 1225397174343

+Bundle-ManifestVersion: 2

+Bundle-License: http://www.apache.org/licenses/LICENSE-2.0.txt

+Bundle-Description: Apache Tuscany SCA XSD Model

+Bundle-SymbolicName: org.apache.tuscany.sca.endpoint.dht

+Bundle-DocURL: http://www.apache.org/

+Bundle-RequiredExecutionEnvironment: J2SE-1.5,JavaSE-1.6
+Import-Package: org.apache.juli.logging;resolution:=optional,

+ org.apache.tuscany.sca.assembly;version="2.0.0",

+ org.apache.tuscany.sca.core;version="2.0.0",

+ org.apache.tuscany.sca.policy;version="2.0.0",

+ org.apache.tuscany.sca.runtime;version="2.0.0"

diff --git a/java/sca/modules/endpoint-dht/pom.xml b/java/sca/modules/endpoint-dht/pom.xml
new file mode 100644
index 0000000000..8c62c5dde6
--- /dev/null
+++ b/java/sca/modules/endpoint-dht/pom.xml
@@ -0,0 +1,47 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ * 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.
+-->
+<project>
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.tuscany.sca</groupId>
+        <artifactId>tuscany-modules</artifactId>
+        <version>2.0-SNAPSHOT</version>
+        <relativePath>../pom.xml</relativePath>
+    </parent>
+    <artifactId>tuscany-endpoint-dht</artifactId>
+    <name>Apache Tuscany SCA OverlayWeaver DHT Based EndPoint Registry</name>
+
+    <dependencies>
+    <!--    <dependency>
+            <groupId>org.apache.tomcat</groupId>
+            <artifactId>overlay</artifactId>
+            <version>6.0.18</version>
+            <scope>compile</scope>
+        </dependency>
+	-->
+        <dependency>
+            <groupId>org.apache.tuscany.sca</groupId>
+            <artifactId>tuscany-core-spi</artifactId>
+            <version>2.0-SNAPSHOT</version>
+            <scope>compile</scope>
+        </dependency>
+    </dependencies>
+
+</project>
diff --git a/java/sca/modules/endpoint-dht/src/main/java/org/apache/tuscany/sca/endpoint/dht/OverlayEndpointRegistry.java b/java/sca/modules/endpoint-dht/src/main/java/org/apache/tuscany/sca/endpoint/dht/OverlayEndpointRegistry.java
new file mode 100644
index 0000000000..1bbf042379
--- /dev/null
+++ b/java/sca/modules/endpoint-dht/src/main/java/org/apache/tuscany/sca/endpoint/dht/OverlayEndpointRegistry.java
@@ -0,0 +1,338 @@
+/*
+ * 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.dht;
+
+import java.net.InetAddress;
+import java.net.NetworkInterface;
+import java.util.ArrayList;
+import java.util.Enumeration;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import ow.id.ID;
+import ow.routing.RoutingException;
+import ow.dht.DHT;
+import ow.dht.DHTConfiguration;
+
+import org.apache.tuscany.sca.assembly.Endpoint;
+import org.apache.tuscany.sca.assembly.EndpointReference;
+import org.apache.tuscany.sca.core.ExtensionPointRegistry;
+import org.apache.tuscany.sca.core.LifeCycleListener;
+import org.apache.tuscany.sca.runtime.EndpointListener;
+import org.apache.tuscany.sca.runtime.EndpointRegistry;
+
+/**
+ * A EndpointRegistry based on Overlay Weaver DHT
+ */
+public class OverlayEndpointRegistry implements EndpointRegistry, LifeCycleListener {
+    private final static Logger logger = Logger.getLogger(OverlayEndpointRegistry.class.getName());
+    private final static int DEFAULT_PORT = 3997;
+    private final static int DEFAULT_TTL = 600 * 1000;
+    private final static String DEFAULT_DOMAIN_URI = "http://tuscany.apache.org/sca/1.1/domains/default";
+    private String domainURI = DEFAULT_DOMAIN_URI;
+    private List<EndpointReference> endpointreferences = new CopyOnWriteArrayList<EndpointReference>();
+    private List<EndpointListener> listeners = new CopyOnWriteArrayList<EndpointListener>();
+    private DHT<Endpoint> map = null;
+    private ConcurrentHashMap<String, Endpoint> publishedEndpoints = new ConcurrentHashMap<String,Endpoint>();
+    private ExtensionPointRegistry registry;
+    private  String  joinPort= null;
+    private  String address  = null;
+
+
+    public OverlayEndpointRegistry(ExtensionPointRegistry registry, Map<String, String> attributes) {
+        this.registry = registry;
+        String portStr = attributes.get("port");
+        if (portStr != null) {
+            port = DEFAULT_PORT;
+        }
+	/* This is the address that you need for join a DHT */
+        String address = attributes.get("address");
+        if (address == null) {
+            address = getBindAddress();
+        }
+    }
+
+    public OverlayEndpointRegistry(String domainURI) {
+        this.domainURI = domainURI;
+        // start();
+    }
+
+    public void start() {
+        if (map != null) {
+            throw new IllegalStateException("The registry has already been started");
+        }
+	/* here you have to join the DHT */
+       DHTConfiguration config = DHTFactory.getDefaultConfiguration();
+		/* the DHT behaviour should be configurable */
+		config.setRoutingStyle("Iterative");
+		config.setRoutingAlgorithm("Pastry");
+		config.setSelfPort(DEFAULT_PORT);
+		try {
+			map = DHTFactory.<Endpoint>getDHT(config);
+		}
+		catch (Exception e) {
+			throw new IllegalStateException(e);
+			}
+		try {
+			map.joinOverlay(address, joinPort);
+		}
+		catch (IOException e) {
+			throw new IllegalStateException(e);
+		}
+
+    }
+
+    public void stop() {
+        if (map != null) {
+            map.stop();
+            map = null;
+        }
+    }
+
+    public void addEndpoint(Endpoint endpoint) {
+	int idSize = map.getRoutingAlgorithmConfiguration().getIDSizeInByte();
+	ID key = ID.getHashcodeBasedID(endpoint.getURI(), idSize);	
+        map.put(key, endpoint);
+	publishedEndpoints.put(endpoint.getURI, endpoint);
+        logger.info("Add endpoint - " + endpoint);
+    }
+
+    public void addEndpointReference(EndpointReference endpointReference) {
+        endpointreferences.add(endpointReference);
+        logger.info("Add endpoint reference - " + endpointReference);
+    }
+
+    public void addListener(EndpointListener listener) {
+        listeners.add(listener);
+    }
+
+    /**
+     * Parse the component/service/binding URI into an array of parts (componentURI, serviceName, bindingName)
+     * @param uri
+     * @return
+     */
+    private String[] parse(String uri) {
+        String[] names = new String[3];
+        int index = uri.lastIndexOf('#');
+        if (index == -1) {
+            names[0] = uri;
+        } else {
+            names[0] = uri.substring(0, index);
+            String str = uri.substring(index + 1);
+            if (str.startsWith("service-binding(") && str.endsWith(")")) {
+                str = str.substring("service-binding(".length(), str.length() - 1);
+                String[] parts = str.split("/");
+                if (parts.length != 2) {
+                    throw new IllegalArgumentException("Invalid service-binding URI: " + uri);
+                }
+                names[1] = parts[0];
+                names[2] = parts[1];
+            } else if (str.startsWith("service(") && str.endsWith(")")) {
+                str = str.substring("service(".length(), str.length() - 1);
+                names[1] = str;
+            } else {
+                throw new IllegalArgumentException("Invalid component/service/binding URI: " + uri);
+            }
+        }
+        return names;
+    }
+
+    private boolean matches(String target, String uri) {
+        String[] parts1 = parse(target);
+        String[] parts2 = parse(uri);
+        for (int i = 0; i < parts1.length; i++) {
+            if (parts1[i] == null || parts1[i].equals(parts2[i])) {
+                continue;
+            } else {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    public List<Endpoint> findEndpoint(EndpointReference endpointReference) {
+        List<Endpoint> foundEndpoints = new ArrayList<Endpoint>();
+
+        logger.info("Find endpoint for reference - " + endpointReference);
+
+        if (endpointReference.getReference() != null) {
+            Endpoint targetEndpoint = endpointReference.getTargetEndpoint();
+	    int idSize = map.getRoutingAlgorithmConfiguration().getIDSizeInByte();
+	    ID key = ID.getHashcodeBasedID(targetEndpoint.getURI(), idSize);	
+	    logger.info("Matching against - " + endpoint);
+	    Set<ValueInfo<Endpoint>> values = map.get(key);
+	   for (ValueInfo<Endpoint>v : values)
+	   {
+		   Endpoint endpoint = v.getValue();
+		   if (matches(targetEndpoint.getURI(), endpoint.getURI())) {
+                    
+                    if (!isLocal(v)) {
+                        endpoint.setRemote(true);
+                    }
+                   
+                    endpoint.setExtensionPointRegistry(registry);
+                  
+                    foundEndpoints.add(endpoint);
+                    logger.info("Found endpoint with matching service  - " + endpoint);
+                }
+     
+	   }
+	
+		
+      }
+        return foundEndpoints;
+    }
+
+    private boolean isLocal(Endpoint entry) {
+	Endpoint local;
+	local = publishedEndpoint.get(entry.getURI());
+	if (local != null)
+	    return true;
+	return false;
+    }
+
+    public List<EndpointReference> findEndpointReference(Endpoint endpoint) {
+        return endpointreferences;
+    }
+
+    public Endpoint getEndpoint(String uri) {
+	/* if is local there no need to go on the net*/
+	Endpoint local;
+	local = publishedEndpoint.get(uri);
+	if (local != null)
+	    return local;
+	/* otherwise we should check on the net */
+	int idSize = map.getRoutingAlgorithmConfiguration().getIDSizeInByte();
+	ID key = ID.getHashcodeBasedID(uri, idSize);	
+	Set<ValueInfo<Endpoint>> values = map.get(key);
+        return (Endpoint)map.get(uri);
+    }
+
+    public List<EndpointReference> getEndpointRefereneces() {
+        return endpointreferences;
+    }
+
+    public List<Endpoint> getEndpoints() {
+	        /*TODO*/
+	}
+
+    public List<EndpointListener> getListeners() {
+        return listeners;
+    }
+
+    public void removeEndpoint(Endpoint endpoint) {
+	    /*TODO*/
+        publishedEndpoint.remove(endpoint.getURI());
+        logger.info("Remove endpoint - " + endpoint);
+    }
+
+    public void removeEndpointReference(EndpointReference endpointReference) {
+        endpointreferences.remove(endpointReference);
+        logger.info("Remove endpoint reference - " + endpointReference);
+    }
+
+    public void removeListener(EndpointListener listener) {
+        listeners.remove(listener);
+    }
+
+    public void updateEndpoint(String uri, Endpoint endpoint) {
+	 /* TODO*/
+    }
+
+    public void entryAdded(Object key, Object value) {
+         /* TODO*/
+    }
+
+    public void entryRemoved(Object key, Object value) {
+	     /* TODO*/
+     }
+
+    public void entryUpdated(Object key, Object oldValue, Object newValue) {
+      /* TODO*/   
+    }
+
+    public static void main(String[] args) throws Exception {
+		DHTConfiguration config = DHTFactory.getDefaultConfiguration();
+		/* the DHT behaviour should be configurable */
+		config.setRoutingStyle("Iterative");
+		config.setRoutingAlgorithm("Pastry");
+		config.setSelfPort(DEFAULT_PORT);
+		try {
+			map = DHTFactory.<Endpoint>getDHT(config);
+		}
+		catch (Exception e) {
+			throw new IllegalStateException(e);
+			}
+		
+		try {
+			map.joinOverlay(address, joinPort);
+		}
+		catch (IOException e) {
+			throw new IllegalStateException(e);
+		}
+		map.put(UUID.randomUUID().toString(), localhost.getHostAddress());
+        for (int i = 0; i < 4; i++) {
+            Thread.sleep(3000);
+            System.out.println(localhost + ": " + map.keySet());
+        }
+        for (Object e : map.entrySetFull()) {
+            Map.Entry en = (Map.Entry)e;
+            AbstractReplicatedMap.MapEntry entry = (AbstractReplicatedMap.MapEntry)en.getValue();
+            System.out.println(entry);
+        }
+        map.breakdown();
+        channel.stop(Channel.DEFAULT);
+    }
+
+    private static String getBindAddress() {
+        try {
+            Enumeration<NetworkInterface> nis = NetworkInterface.getNetworkInterfaces();
+            while (nis.hasMoreElements()) {
+                NetworkInterface ni = nis.nextElement();
+                // The following APIs require JDK 1.6
+                /*
+                if (ni.isLoopback() || !ni.isUp() || !ni.supportsMulticast()) {
+                    continue;
+                }
+                */
+                Enumeration<InetAddress> ips = ni.getInetAddresses();
+                if (!ips.hasMoreElements()) {
+                    continue;
+                }
+                while (ips.hasMoreElements()) {
+                    InetAddress addr = ips.nextElement();
+                    if (addr.isLoopbackAddress()) {
+                        continue;
+                    }
+                    return addr.getHostAddress();
+                }
+            }
+            return InetAddress.getLocalHost().getHostAddress();
+        } catch (Exception e) {
+            logger.log(Level.SEVERE, e.getMessage(), e);
+            return null;
+        }
+    }
+
+}
diff --git a/java/sca/modules/endpoint-dht/src/main/resources/META-INF/services/org.apache.tuscany.sca.runtime.EndpointRegistry b/java/sca/modules/endpoint-dht/src/main/resources/META-INF/services/org.apache.tuscany.sca.runtime.EndpointRegistry
new file mode 100644
index 0000000000..e262d28708
--- /dev/null
+++ b/java/sca/modules/endpoint-dht/src/main/resources/META-INF/services/org.apache.tuscany.sca.runtime.EndpointRegistry
@@ -0,0 +1,17 @@
+# 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.

+org.apache.tuscany.sca.endpoint.tribes.ReplicatedEndpointRegistry;ranking=150,address=planetlab2.fem.tu-ilmenau.de,port=3998,timeout=50