some additions and fixes:

- added WiringSender.toString() method for better debugging
- changed ttl and update intervals for etcd discovery
- use threadsafe collections
- added some logging

Author: Marc Sluiter

org.inaetics.wiring/src/org/inaetics/wiring/admin/http/WiringSenderImpl.java

@@ -25,8 +25,12 @@ public WiringSenderImpl(HttpClientEndpointFactory endpointFactory, HttpAdminConf
 
 	@Override
 	public String sendMessage(String message) throws Exception {
-
 		return m_endpointFactory.sendMessage(m_endpoint.getId(), message);
 	}
 
+	@Override
+	public String toString() {
+		return "WiringSenderImpl [endpointId=" + m_endpoint.getId() + "]";
+	}
+	
 }

org.inaetics.wiring/src/org/inaetics/wiring/discovery/AbstractDiscovery.java

@@ -152,8 +152,7 @@ public void run() {
     }
 
     private boolean isLocalEndpoint(WiringEndpointDescription endpointDescription) {
-//    	return endpointDescription.getZone().equals(m_configuration.getZone())
-//    			&& endpointDescription.getNode().equals(m_configuration.getNode());
+    	//todo check framework uuid?!
     	return false;
     }
 

org.inaetics.wiring/src/org/inaetics/wiring/discovery/etcd/EtcdNodeDiscovery.java

@@ -314,13 +314,13 @@ private String getEndpointPath(WiringEndpointDescription endpoint) {
 
     private class EtcdRegistrationUpdater implements Runnable {
 
-        private static final int ETCD_REGISTRATION_TTL = 60;
+        private static final int ETCD_REGISTRATION_TTL = 30;
 
         private final ScheduledFuture<?> m_future;
 
         public EtcdRegistrationUpdater() throws Exception {
             m_future =
-                m_executor.scheduleAtFixedRate(this, 0, ETCD_REGISTRATION_TTL - 5,
+                m_executor.scheduleAtFixedRate(this, 0, ETCD_REGISTRATION_TTL - 10,
                     TimeUnit.SECONDS);
         }
 

org.inaetics.wiring/src/org/inaetics/wiring/topology/promiscuous/PromiscuousTopologyManager.java

@@ -3,16 +3,14 @@
  */
 package org.inaetics.wiring.topology.promiscuous;
 
-import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Dictionary;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.Hashtable;
-import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 
 import org.apache.felix.dm.Component;
 import org.apache.felix.dm.DependencyManager;
@@ -32,7 +30,6 @@
 import org.inaetics.wiring.endpoint.WiringTopologyManager;
 import org.osgi.framework.BundleContext;
 import org.osgi.framework.ServiceReference;
-import org.osgi.framework.ServiceRegistration;
 import org.osgi.service.cm.ConfigurationException;
 import org.osgi.service.cm.ManagedService;
 
@@ -47,17 +44,18 @@ public final class PromiscuousTopologyManager extends AbstractWiringEndpointPubl
 
     public final static String SERVICE_PID = "org.amdatu.remote.topology.promiscuous";
 
-    private final Set<WiringReceiver> m_exportableReceivers = Collections.newSetFromMap(new HashMap<WiringReceiver, Boolean>());
+    private final Set<WiringReceiver> m_exportableReceivers = Collections.newSetFromMap(new ConcurrentHashMap<WiringReceiver, Boolean>());
     private final Map<WiringReceiver, Map<WiringAdmin, ExportRegistration>> m_exportedReceivers =
-            new HashMap<WiringReceiver, Map<WiringAdmin, ExportRegistration>>();
+            new ConcurrentHashMap<WiringReceiver, Map<WiringAdmin, ExportRegistration>>();
 
-    private final Set<WiringEndpointDescription> m_importableEndpoints = new HashSet<WiringEndpointDescription>();
+    private final Set<WiringEndpointDescription> m_importableEndpoints = Collections.newSetFromMap(new ConcurrentHashMap<WiringEndpointDescription, Boolean>());
     private final Map<WiringEndpointDescription, Map<WiringAdmin, ImportRegistration>> m_importedEndpoints =
-        new HashMap<WiringEndpointDescription, Map<WiringAdmin, ImportRegistration>>();
+        new ConcurrentHashMap<WiringEndpointDescription, Map<WiringAdmin, ImportRegistration>>();
+    
     private final Map<ImportRegistration, Component> m_registeredSenders =
-            new HashMap<ImportRegistration, Component>();
+            new ConcurrentHashMap<ImportRegistration, Component>();
 
-    private final List<WiringAdmin> m_wiringAdmins = new ArrayList<WiringAdmin>();
+    private final Set<WiringAdmin> m_wiringAdmins = Collections.newSetFromMap(new ConcurrentHashMap<WiringAdmin, Boolean>());
 
 	private volatile BundleContext m_context;
 
@@ -267,6 +265,7 @@ private void unImportEndpoints(WiringAdmin admin) {
 	}
 
 	private void unImportEndpoint(WiringEndpointDescription endpointDescription) {
+		logInfo("unimport wiring endpoint %s", endpointDescription.getId());
 		m_importableEndpoints.remove(endpointDescription);
 		Map<WiringAdmin, ImportRegistration> adminMap = m_importedEndpoints.remove(endpointDescription);
 		Collection<ImportRegistration> registrations = adminMap.values();
@@ -276,12 +275,14 @@ private void unImportEndpoint(WiringEndpointDescription endpointDescription) {
 	}
 
 	private void unImportEndpoint(ImportRegistration registration) {
+		logInfo("unimport registration %s", registration.getImportReference().getEndpointDescription().getId());
 		unregisterService(registration);
 		registration.close();
 	}
 
 	private void unregisterService(ImportRegistration registration) {
 		Component component = m_registeredSenders.get(registration);
+		logInfo("unregistering WiringsSender %s", component.getService());
 		m_manager.remove(component);
 		m_registeredSenders.remove(registration);
 	}