diff --git a/.gitignore b/.gitignore
index ca2256e..9711ac3 100644
--- a/.gitignore
+++ b/.gitignore
@@ -11,4 +11,5 @@
 *.md.html
 .idea/
 generated/
+.vscode/
 
diff --git a/discovery/local/bnd.bnd b/discovery/local/bnd.bnd
index 203f6a1..d20f035 100644
--- a/discovery/local/bnd.bnd
+++ b/discovery/local/bnd.bnd
@@ -15,13 +15,12 @@
 #    KIND, either express or implied.  See the License for the
 #    specific language governing permissions and limitations
 #    under the License.
-Bundle-Activator: org.apache.aries.rsa.discovery.local.Activator
 Private-Package: org.apache.aries.rsa.discovery.local
 Export-Package: \
-	org.apache.aries.rsa.discovery.endpoint,\
 	org.osgi.xmlns.rsa.v1_0
 Provide-Capability: osgi.extender;osgi.extender="osgi.remoteserviceadmin";\
 	version:Version="1.1.0";\
 	uses:="org.osgi.service.remoteserviceadmin",\
 	osgi.remoteserviceadmin.discovery;\
 	protocols:List<String>="local"; version:Version=1.1.0
+-privatepackage: org.apache.aries.rsa.discovery.endpoint
\ No newline at end of file
diff --git a/discovery/local/src/main/java/org/apache/aries/rsa/discovery/endpoint/EndpointDescriptionParser.java b/discovery/local/src/main/java/org/apache/aries/rsa/discovery/endpoint/EndpointDescriptionParserImpl.java
similarity index 94%
rename from discovery/local/src/main/java/org/apache/aries/rsa/discovery/endpoint/EndpointDescriptionParser.java
rename to discovery/local/src/main/java/org/apache/aries/rsa/discovery/endpoint/EndpointDescriptionParserImpl.java
index 9fcb8b4..5acd583 100644
--- a/discovery/local/src/main/java/org/apache/aries/rsa/discovery/endpoint/EndpointDescriptionParser.java
+++ b/discovery/local/src/main/java/org/apache/aries/rsa/discovery/endpoint/EndpointDescriptionParserImpl.java
@@ -33,16 +33,19 @@
 import javax.xml.transform.Source;
 import javax.xml.transform.stream.StreamSource;
 
+import org.apache.aries.rsa.spi.EndpointDescriptionParser;
+import org.osgi.service.component.annotations.Component;
 import org.osgi.service.remoteserviceadmin.EndpointDescription;
 import org.osgi.xmlns.rsa.v1_0.EndpointDescriptionType;
 import org.osgi.xmlns.rsa.v1_0.EndpointDescriptionsType;
 import org.osgi.xmlns.rsa.v1_0.ObjectFactory;
 import org.osgi.xmlns.rsa.v1_0.PropertyType;
 
-public class EndpointDescriptionParser {
+@Component
+public class EndpointDescriptionParserImpl implements EndpointDescriptionParser {
     private JAXBContext jaxbContext;
 
-    public EndpointDescriptionParser() {
+    public EndpointDescriptionParserImpl() {
         try {
             jaxbContext = JAXBContext.newInstance(EndpointDescriptionsType.class);
         } catch (JAXBException e) {
diff --git a/discovery/local/src/main/java/org/apache/aries/rsa/discovery/local/Activator.java b/discovery/local/src/main/java/org/apache/aries/rsa/discovery/local/Activator.java
deleted file mode 100644
index 3e9a187..0000000
--- a/discovery/local/src/main/java/org/apache/aries/rsa/discovery/local/Activator.java
+++ /dev/null
@@ -1,78 +0,0 @@
-/**
- * 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.aries.rsa.discovery.local;
-
-import org.osgi.framework.BundleActivator;
-import org.osgi.framework.BundleContext;
-import org.osgi.framework.ServiceReference;
-import org.osgi.service.remoteserviceadmin.EndpointEventListener;
-import org.osgi.util.tracker.ServiceTracker;
-
-public class Activator implements BundleActivator {
-    private ServiceTracker<EndpointEventListener, EndpointEventListener> listenerTracker;
-    private LocalDiscovery localDiscovery;
-
-    public void start(BundleContext context) {
-        localDiscovery = new LocalDiscovery();
-        listenerTracker = new EPListenerTracker(context, localDiscovery);
-        listenerTracker.open();
-        localDiscovery.processExistingBundles(context.getBundles());
-        context.addBundleListener(localDiscovery);
-    }
-
-    public void stop(BundleContext context) {
-        listenerTracker.close();
-        context.removeBundleListener(localDiscovery);
-    }
-
-    private final class EPListenerTracker extends ServiceTracker<EndpointEventListener, EndpointEventListener> {
-        private final LocalDiscovery localDiscovery;
-    
-        private EPListenerTracker(BundleContext context, LocalDiscovery localDiscovery) {
-            super(context, EndpointEventListener.class, null);
-            this.localDiscovery = localDiscovery;
-        }
-    
-        @Override
-        public EndpointEventListener addingService(ServiceReference<EndpointEventListener> reference) {
-            EndpointEventListener service = super.addingService(reference);
-            localDiscovery.addListener(reference, service);
-            return service;
-        }
-    
-        @Override
-        public void modifiedService(ServiceReference<EndpointEventListener> reference, EndpointEventListener service) {
-            super.modifiedService(reference, service);
-            localDiscovery.removeListener(service);
-    
-            // This may cause duplicate registrations of remote services,
-            // but that's fine and should be filtered out on another level.
-            // See Remote Service Admin spec section 122.6.3
-            localDiscovery.addListener(reference, service);
-        }
-    
-        @Override
-        public void removedService(ServiceReference<EndpointEventListener> reference, EndpointEventListener service) {
-            super.removedService(reference, service);
-            localDiscovery.removeListener(service);
-        }
-    }
-
-    
-}
diff --git a/discovery/local/src/main/java/org/apache/aries/rsa/discovery/local/EndpointDescriptionBundleParser.java b/discovery/local/src/main/java/org/apache/aries/rsa/discovery/local/EndpointDescriptionBundleParser.java
index 1484dcb..1929b6c 100644
--- a/discovery/local/src/main/java/org/apache/aries/rsa/discovery/local/EndpointDescriptionBundleParser.java
+++ b/discovery/local/src/main/java/org/apache/aries/rsa/discovery/local/EndpointDescriptionBundleParser.java
@@ -25,7 +25,7 @@
 import java.util.Enumeration;
 import java.util.List;
 
-import org.apache.aries.rsa.discovery.endpoint.EndpointDescriptionParser;
+import org.apache.aries.rsa.discovery.endpoint.EndpointDescriptionParserImpl;
 import org.osgi.framework.Bundle;
 import org.osgi.service.remoteserviceadmin.EndpointDescription;
 import org.slf4j.Logger;
@@ -37,10 +37,10 @@
     private static final String REMOTE_SERVICES_HEADER_NAME = "Remote-Service";
     private static final String REMOTE_SERVICES_DIRECTORY = "OSGI-INF/remote-service/";
 
-    private EndpointDescriptionParser parser;
+    private EndpointDescriptionParserImpl parser;
 
     public EndpointDescriptionBundleParser() {
-        parser = new EndpointDescriptionParser();
+        parser = new EndpointDescriptionParserImpl();
     }
 
     public List<EndpointDescription> getAllEndpointDescriptions(Bundle b) {
@@ -57,7 +57,7 @@
         return elements;
     }
     
-    Enumeration<URL> getEndpointDescriptionURLs(Bundle b) {
+    private Enumeration<URL> getEndpointDescriptionURLs(Bundle b) {
         String origDir = getRemoteServicesDir(b);
         
         // Split origDir into dir and file pattern
diff --git a/discovery/local/src/main/java/org/apache/aries/rsa/discovery/local/LocalDiscovery.java b/discovery/local/src/main/java/org/apache/aries/rsa/discovery/local/LocalDiscovery.java
index adf98b4..22b1f2d 100644
--- a/discovery/local/src/main/java/org/apache/aries/rsa/discovery/local/LocalDiscovery.java
+++ b/discovery/local/src/main/java/org/apache/aries/rsa/discovery/local/LocalDiscovery.java
@@ -31,15 +31,21 @@
 
 import org.apache.aries.rsa.util.StringPlus;
 import org.osgi.framework.Bundle;
+import org.osgi.framework.BundleContext;
 import org.osgi.framework.BundleEvent;
 import org.osgi.framework.BundleListener;
 import org.osgi.framework.Filter;
 import org.osgi.framework.FrameworkUtil;
 import org.osgi.framework.ServiceReference;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Deactivate;
+import org.osgi.service.component.annotations.Reference;
 import org.osgi.service.remoteserviceadmin.EndpointDescription;
 import org.osgi.service.remoteserviceadmin.EndpointEvent;
 import org.osgi.service.remoteserviceadmin.EndpointEventListener;
 
+@Component(immediate = true)
 public class LocalDiscovery implements BundleListener {
 
     // this is effectively a set which allows for multiple service descriptions with the
@@ -53,6 +59,18 @@
     public LocalDiscovery() {
         this.bundleParser = new EndpointDescriptionBundleParser();
     }
+    
+    @Activate
+    public void activate(BundleContext context) {
+        Bundle[] bundles = context.getBundles();
+        processExistingBundles(bundles);
+        context.addBundleListener(this);
+    }
+    
+    @Deactivate
+    public void deactivate(BundleContext context) {
+        context.removeBundleListener(this);
+    }
 
     public void processExistingBundles(Bundle[] bundles) {
         if (bundles == null) {
@@ -66,7 +84,8 @@
         }
     }
 
-    void addListener(ServiceReference<EndpointEventListener> endpointListenerRef, EndpointEventListener endpointListener) {
+    @Reference
+    void bindListener(ServiceReference<EndpointEventListener> endpointListenerRef, EndpointEventListener endpointListener) {
         List<String> filters = StringPlus.normalize(endpointListenerRef.getProperty(EndpointEventListener.ENDPOINT_LISTENER_SCOPE));
         if (filters.isEmpty()) {
             return;
@@ -93,7 +112,7 @@
      * itself to clean up any orphans. See Remote Service Admin spec 122.6.3
      * @param endpointListener
      */
-    void removeListener(EndpointEventListener endpointListener) {
+    void unbindListener(EndpointEventListener endpointListener) {
         synchronized (listenerToFilters) {
             Collection<String> filters = listenerToFilters.remove(endpointListener);
             if (filters == null) {
diff --git a/discovery/local/src/test/java/org/apache/aries/rsa/discovery/endpoint/EndpointDescriptionParserTest.java b/discovery/local/src/test/java/org/apache/aries/rsa/discovery/endpoint/EndpointDescriptionParserTest.java
index 12ead94..9dcefc0 100644
--- a/discovery/local/src/test/java/org/apache/aries/rsa/discovery/endpoint/EndpointDescriptionParserTest.java
+++ b/discovery/local/src/test/java/org/apache/aries/rsa/discovery/endpoint/EndpointDescriptionParserTest.java
@@ -31,7 +31,7 @@
     @Test
     public void testEndpointDescriptionsFromURL() throws IOException {
         URL ed1URL = getClass().getResource("/ed1.xml");
-        List<EndpointDescription> edElements = new EndpointDescriptionParser().
+        List<EndpointDescription> edElements = new EndpointDescriptionParserImpl().
             readEndpoints(ed1URL.openStream());
         Assert.assertEquals(4, edElements.size());
     }
diff --git a/discovery/local/src/test/java/org/apache/aries/rsa/discovery/endpoint/PropertiesMapperTest.java b/discovery/local/src/test/java/org/apache/aries/rsa/discovery/endpoint/PropertiesMapperTest.java
index ffd1dbc..31fa1b7 100644
--- a/discovery/local/src/test/java/org/apache/aries/rsa/discovery/endpoint/PropertiesMapperTest.java
+++ b/discovery/local/src/test/java/org/apache/aries/rsa/discovery/endpoint/PropertiesMapperTest.java
@@ -80,7 +80,7 @@
 
         ByteArrayOutputStream bos = new ByteArrayOutputStream();
         EndpointDescription epd = new EndpointDescription(m);
-        new EndpointDescriptionParser().writeEndpoint(epd, bos);
+        new EndpointDescriptionParserImpl().writeEndpoint(epd, bos);
         byte[] epData = bos.toByteArray();
         System.out.println(new String(epData));
         URL edURL = getClass().getResource("/ed2-generated.xml");
diff --git a/discovery/local/src/test/java/org/apache/aries/rsa/discovery/local/LocalDiscoveryTest.java b/discovery/local/src/test/java/org/apache/aries/rsa/discovery/local/LocalDiscoveryTest.java
index 59b3c3f..20afdf6 100644
--- a/discovery/local/src/test/java/org/apache/aries/rsa/discovery/local/LocalDiscoveryTest.java
+++ b/discovery/local/src/test/java/org/apache/aries/rsa/discovery/local/LocalDiscoveryTest.java
@@ -103,7 +103,7 @@
         endpointListener.endpointChanged(EasyMock.anyObject(EndpointEvent.class), EasyMock.eq("(objectClass=*)"));
         EasyMock.expectLastCall();
         EasyMock.replay(endpointListener);
-        ld.addListener(sr, endpointListener);
+        ld.bindListener(sr, endpointListener);
 
         // Start the bundle
         BundleEvent be = new BundleEvent(BundleEvent.STARTED, bundle);
@@ -148,7 +148,7 @@
         // Add the EndpointListener Service
         assertEquals("Precondition failed", 0, ld.listenerToFilters.size());
         assertEquals("Precondition failed", 0, ld.filterToListeners.size());
-        ld.addListener(sr, el);
+        ld.bindListener(sr, el);
 
         assertEquals(1, ld.listenerToFilters.size());
         assertEquals(Collections.singletonList("(objectClass=org.example.ClassA)"), ld.listenerToFilters.get(el));
@@ -175,8 +175,8 @@
         }).times(2);
         EasyMock.replay(el);
 
-        ld.removeListener(el);
-        ld.addListener(sr2, el);
+        ld.unbindListener(el);
+        ld.bindListener(sr2, el);
         assertEquals(1, ld.listenerToFilters.size());
         assertEquals(Arrays.asList("(|(objectClass=org.example.ClassA)(objectClass=org.example.ClassB))"),
             ld.listenerToFilters.get(el));
@@ -189,7 +189,7 @@
         assertEquals(expectedEndpoints, actualEndpoints);
 
         // Remove the EndpointListener Service
-        ld.removeListener(el);
+        ld.unbindListener(el);
         assertEquals(0, ld.listenerToFilters.size());
         assertEquals(0, ld.filterToListeners.size());
     }
@@ -215,7 +215,7 @@
 
         assertEquals("Precondition failed", 0, ld.listenerToFilters.size());
         assertEquals("Precondition failed", 0, ld.filterToListeners.size());
-        ld.addListener(sr, endpointListener);
+        ld.bindListener(sr, endpointListener);
 
         assertEquals(1, ld.listenerToFilters.size());
         assertEquals(Collections.singletonList("(objectClass=Aaaa)"), ld.listenerToFilters.get(endpointListener));
@@ -235,7 +235,7 @@
 
         EndpointEventListener endpointListener2 = EasyMock.createMock(EndpointEventListener.class);
         EasyMock.replay(endpointListener2);
-        ld.addListener(sr2, endpointListener2);
+        ld.bindListener(sr2, endpointListener2);
 
         assertEquals(2, ld.listenerToFilters.size());
         assertEquals(Collections.singletonList("(objectClass=Aaaa)"), ld.listenerToFilters.get(endpointListener));
@@ -260,7 +260,7 @@
 
         EndpointEventListener endpointListener3 = EasyMock.createMock(EndpointEventListener.class);
         EasyMock.replay(endpointListener3);
-        ld.addListener(sr3, endpointListener3);
+        ld.bindListener(sr3, endpointListener3);
 
         assertEquals(3, ld.listenerToFilters.size());
         assertEquals(Collections.singletonList("(objectClass=Aaaa)"), ld.listenerToFilters.get(endpointListener));
@@ -287,11 +287,11 @@
         assertEquals(1, ld.listenerToFilters.size());
         assertEquals(2, ld.filterToListeners.size());
         assertEquals(1, ld.filterToListeners.values().iterator().next().size());
-        ld.removeListener(EasyMock.createMock(EndpointEventListener.class));
+        ld.unbindListener(EasyMock.createMock(EndpointEventListener.class));
         assertEquals(1, ld.listenerToFilters.size());
         assertEquals(2, ld.filterToListeners.size());
         assertEquals(1, ld.filterToListeners.values().iterator().next().size());
-        ld.removeListener(endpointListener);
+        ld.unbindListener(endpointListener);
         assertEquals(0, ld.listenerToFilters.size());
         assertEquals(0, ld.filterToListeners.size());
     }
diff --git a/discovery/zookeeper/bnd.bnd b/discovery/zookeeper/bnd.bnd
index 4a8cc29..4f06393 100644
--- a/discovery/zookeeper/bnd.bnd
+++ b/discovery/zookeeper/bnd.bnd
@@ -16,3 +16,4 @@
 #    under the License.
 Provide-Capability: osgi.remoteserviceadmin.discovery;\
 	protocols:List<String>="zookeeper"; version:Version=1.1.0
+Export-Package: org.apache.aries.rsa.discovery.zookeeper
\ No newline at end of file
diff --git a/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/ClientManager.java b/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/ClientManager.java
index 9ba3fd2..e6bcf6f 100644
--- a/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/ClientManager.java
+++ b/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/ClientManager.java
@@ -18,14 +18,15 @@
  */
 package org.apache.aries.rsa.discovery.zookeeper;
 
+import static java.util.concurrent.CompletableFuture.runAsync;
+
 import java.io.IOException;
 import java.util.Hashtable;
-import java.util.concurrent.CompletableFuture;
 
 import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.Watcher;
-import org.apache.zookeeper.ZooKeeper;
 import org.apache.zookeeper.Watcher.Event.KeeperState;
+import org.apache.zookeeper.ZooKeeper;
 import org.apache.zookeeper.server.ZooTrace;
 import org.osgi.framework.BundleContext;
 import org.osgi.framework.ServiceRegistration;
@@ -40,16 +41,16 @@
 @Component(//
         service = ClientManager.class,
         immediate = true,
-        configurationPid = "org.apache.aries.rsa.discovery.zookeeper", //
+        configurationPid = ClientManager.DISCOVERY_ZOOKEEPER_ID, //
         configurationPolicy = ConfigurationPolicy.REQUIRE //
 )
 public class ClientManager implements Watcher {
-
+    public static final String DISCOVERY_ZOOKEEPER_ID = "org.apache.aries.rsa.discovery.zookeeper";
     private static final Logger LOG = LoggerFactory.getLogger(ClientManager.class);
 
     private ZooKeeper zkClient;
     private DiscoveryConfig config;
-    private ServiceRegistration<ZooKeeper> reg;
+    private volatile ServiceRegistration<ZooKeeper> reg;
     private BundleContext context;
 
     @Activate
@@ -81,11 +82,7 @@
         if (reg != null) {
             reg.unregister();
         }
-        CompletableFuture.runAsync(new Runnable() {
-            public void run() {
-                closeClient();
-            }
-        });
+        runAsync(this::closeClient);
     }
 
     private void closeClient() {
diff --git a/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/subscribe/InterestManager.java b/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/InterestManager.java
similarity index 63%
rename from discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/subscribe/InterestManager.java
rename to discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/InterestManager.java
index 6e92c78..41bce86 100644
--- a/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/subscribe/InterestManager.java
+++ b/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/InterestManager.java
@@ -16,16 +16,19 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.aries.rsa.discovery.zookeeper.subscribe;
+package org.apache.aries.rsa.discovery.zookeeper;
 
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
-import org.apache.aries.rsa.discovery.zookeeper.ZooKeeperDiscovery;
-import org.apache.aries.rsa.discovery.zookeeper.repository.ZookeeperEndpointRepository;
 import org.apache.aries.rsa.util.StringPlus;
 import org.osgi.framework.ServiceReference;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Deactivate;
+import org.osgi.service.component.annotations.Reference;
+import org.osgi.service.component.annotations.ReferenceCardinality;
+import org.osgi.service.component.annotations.ReferencePolicy;
 import org.osgi.service.remoteserviceadmin.EndpointDescription;
 import org.osgi.service.remoteserviceadmin.EndpointEvent;
 import org.osgi.service.remoteserviceadmin.EndpointEventListener;
@@ -39,10 +42,12 @@
  * Events from repository are then forwarded to all interested EndpointEventListeners.
  */
 @SuppressWarnings({"deprecation", "rawtypes"})
-public class InterestManager implements EndpointEventListener {
+@Component(service = InterestManager.class)
+public class InterestManager {
     private static final Logger LOG = LoggerFactory.getLogger(InterestManager.class);
 
-    private final ZookeeperEndpointRepository repository;
+    private Map<String, EndpointDescription> nodes = new ConcurrentHashMap<>();
+    
     private final Map<ServiceReference, Interest> interests = new ConcurrentHashMap<>();
 
     protected static class Interest {
@@ -50,11 +55,51 @@
         Object epListener;
     }
 
-    public InterestManager(ZookeeperEndpointRepository repository) {
-        this.repository = repository;
+    @Reference(cardinality = ReferenceCardinality.MULTIPLE, policy = ReferencePolicy.DYNAMIC)
+    public void bindEndpointEventListener(ServiceReference<EndpointEventListener> sref, EndpointEventListener epListener) {
+        addInterest(sref, epListener);
+    }
+    
+    public void updatedEndpointEventListener(ServiceReference<EndpointEventListener> sref, EndpointEventListener epListener) {
+        addInterest(sref, epListener);
+    }
+    
+    public void unbindEndpointEventListener(ServiceReference<EndpointEventListener> sref) {
+        removeInterest(sref);
     }
 
-    public void addInterest(ServiceReference<?> sref, Object epListener) {
+    @Reference(cardinality = ReferenceCardinality.MULTIPLE, policy = ReferencePolicy.DYNAMIC)
+    public void bindEndpointListener(ServiceReference<EndpointListener> sref, EndpointListener epListener) {
+        addInterest(sref, epListener);
+    }
+    
+    public void updatedEndpointListener(ServiceReference<EndpointListener> sref, EndpointListener epListener) {
+        addInterest(sref, epListener);
+    }
+    
+    public void unbindEndpointListener(ServiceReference<EndpointListener> sref) {
+        removeInterest(sref);
+    }
+
+    private void removeInterest(ServiceReference<?> sref) {
+        if (interests.containsKey(sref)) {
+            List<String> scopes = getScopes(sref);
+            LOG.info("removing interests: {}", scopes);
+            interests.remove(sref);
+        }
+    }
+    
+    /**
+     * Read current endpoint stored at a znode
+     * 
+     * @param path
+     * @return
+     */
+    EndpointDescription read(String path) {
+        return nodes.get(path);
+    }
+
+    private void addInterest(ServiceReference<?> sref, Object epListener) {
         if (isOurOwnEndpointEventListener(sref)) {
             LOG.debug("Skipping our own EndpointEventListener");
             return;
@@ -71,28 +116,40 @@
             interest.scopes = scopes;
             interests.put(sref, interest);
             sendExistingEndpoints(scopes, epListener);
+        } else {
+            interest.scopes = scopes;
+            sendExistingEndpoints(scopes, epListener);
         }
     }
 
     private void sendExistingEndpoints(List<String> scopes, Object epListener) {
-        for (EndpointDescription endpoint : repository.getAll()) {
+        for (EndpointDescription endpoint : nodes.values()) {
             EndpointEvent event = new EndpointEvent(EndpointEvent.ADDED, endpoint);
             notifyListener(event, scopes, epListener);
         }
     }
 
-    private static boolean isOurOwnEndpointEventListener(ServiceReference<?> EndpointEventListener) {
+    private static boolean isOurOwnEndpointEventListener(ServiceReference<?> endpointEventListener) {
         return Boolean.parseBoolean(String.valueOf(
-                EndpointEventListener.getProperty(ZooKeeperDiscovery.DISCOVERY_ZOOKEEPER_ID)));
+                endpointEventListener.getProperty(ClientManager.DISCOVERY_ZOOKEEPER_ID)));
+    }
+    
+    public void handleRemoved(String path) {
+        EndpointDescription endpoint = nodes.remove(path);
+        if (endpoint != null) {
+            EndpointEvent event = new EndpointEvent(EndpointEvent.REMOVED, endpoint);
+            endpointChanged(event);
+        }
     }
 
-    public void removeInterest(ServiceReference<EndpointEventListener> epListenerRef) {
-        LOG.info("removing EndpointEventListener interests: {}", epListenerRef);
-        interests.remove(epListenerRef);
+    public void handleChanged(String path, EndpointDescription endpoint) {
+        EndpointDescription old = nodes.put(path, endpoint);
+        int type = old == null ? EndpointEvent.ADDED : EndpointEvent.MODIFIED;
+        EndpointEvent event = new EndpointEvent(type, endpoint);
+        endpointChanged(event);
     }
 
-    @Override
-    public void endpointChanged(EndpointEvent event, String filter) {
+    private void endpointChanged(EndpointEvent event) {
         for (Interest interest : interests.values()) {
             notifyListener(event, interest.scopes, interest.epListener);
         }
@@ -146,7 +203,9 @@
         }
     }
 
+    @Deactivate
     public synchronized void close() {
+        nodes.clear();
         interests.clear();
     }
 
diff --git a/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/PublishingEndpointListener.java b/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/PublishingEndpointListener.java
new file mode 100644
index 0000000..89e1556
--- /dev/null
+++ b/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/PublishingEndpointListener.java
@@ -0,0 +1,88 @@
+/**
+ * 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.aries.rsa.discovery.zookeeper;
+
+import java.util.Dictionary;
+import java.util.Hashtable;
+
+import org.osgi.framework.BundleContext;
+import org.osgi.framework.Constants;
+import org.osgi.framework.ServiceRegistration;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Deactivate;
+import org.osgi.service.component.annotations.Reference;
+import org.osgi.service.remoteserviceadmin.EndpointDescription;
+import org.osgi.service.remoteserviceadmin.EndpointEvent;
+import org.osgi.service.remoteserviceadmin.EndpointEventListener;
+import org.osgi.service.remoteserviceadmin.EndpointListener;
+import org.osgi.service.remoteserviceadmin.RemoteConstants;
+
+/**
+ * Listens for local EndpointEvents using old and new style listeners and publishes changes to 
+ * the ZooKeeperEndpointRepository
+ */
+@SuppressWarnings("deprecation")
+@Component(service = {}, immediate = true)
+public class PublishingEndpointListener implements EndpointEventListener, EndpointListener {
+
+    private ServiceRegistration<?> listenerReg;
+    
+    @Reference
+    private ZookeeperEndpointPublisher repository;
+
+    @Activate
+    public void start(BundleContext bctx) {
+        String uuid = bctx.getProperty(Constants.FRAMEWORK_UUID);
+        String[] ifAr = {EndpointEventListener.class.getName(), EndpointListener.class.getName()};
+        Dictionary<String, String> props = serviceProperties(uuid);
+        listenerReg = bctx.registerService(ifAr, this, props);
+    }
+
+    @Deactivate
+    public void stop() {
+        listenerReg.unregister();
+    }
+
+    @Override
+    public void endpointChanged(EndpointEvent event, String filter) {
+        repository.endpointChanged(event);
+    }
+    
+    @Override
+    public void endpointAdded(EndpointDescription endpoint, String matchedFilter) {
+        endpointChanged(new EndpointEvent(EndpointEvent.ADDED, endpoint), matchedFilter);
+    }
+
+    @Override
+    public void endpointRemoved(EndpointDescription endpoint, String matchedFilter) {
+        endpointChanged(new EndpointEvent(EndpointEvent.REMOVED, endpoint), matchedFilter);
+    }
+
+    private Dictionary<String, String> serviceProperties(String uuid) {
+        String scope = String.format("(&(%s=*)(%s=%s))", Constants.OBJECTCLASS, 
+                        RemoteConstants.ENDPOINT_FRAMEWORK_UUID, uuid);
+        Dictionary<String, String> props = new Hashtable<>();
+        props.put(EndpointEventListener.ENDPOINT_LISTENER_SCOPE, scope);
+        props.put(ClientManager.DISCOVERY_ZOOKEEPER_ID, "true");
+        return props;
+    }
+
+
+}
diff --git a/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/ZooKeeperDiscovery.java b/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/ZooKeeperDiscovery.java
deleted file mode 100644
index 1ca8516..0000000
--- a/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/ZooKeeperDiscovery.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/**
- * 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.aries.rsa.discovery.zookeeper;
-
-import org.apache.aries.rsa.discovery.zookeeper.publish.PublishingEndpointListener;
-import org.apache.aries.rsa.discovery.zookeeper.repository.ZookeeperEndpointRepository;
-import org.apache.aries.rsa.discovery.zookeeper.subscribe.EndpointListenerTracker;
-import org.apache.aries.rsa.discovery.zookeeper.subscribe.InterestManager;
-import org.apache.zookeeper.ZooKeeper;
-import org.osgi.framework.BundleContext;
-import org.osgi.service.component.annotations.Activate;
-import org.osgi.service.component.annotations.Component;
-import org.osgi.service.component.annotations.Deactivate;
-import org.osgi.service.component.annotations.Reference;
-import org.osgi.util.tracker.ServiceTracker;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-@Component
-public class ZooKeeperDiscovery {
-    public static final String DISCOVERY_ZOOKEEPER_ID = "org.apache.cxf.dosgi.discovery.zookeeper";
-
-    private static final Logger LOG = LoggerFactory.getLogger(ZooKeeperDiscovery.class);
-
-    @Reference
-    private ZooKeeper zkClient;
-    
-    private PublishingEndpointListener endpointListener;
-    private ServiceTracker<?, ?> endpointListenerTracker;
-    private InterestManager imManager;
-    private ZookeeperEndpointRepository repository;
-
-    @Activate
-    public void activate(BundleContext context) {
-        LOG.debug("Starting ZookeeperDiscovery");
-        repository = new ZookeeperEndpointRepository(zkClient);
-        endpointListener = new PublishingEndpointListener(repository);
-        endpointListener.start(context);
-        imManager = new InterestManager(repository);
-        repository.addListener(imManager);
-        endpointListenerTracker = new EndpointListenerTracker(context, imManager);
-        endpointListenerTracker.open();
-    }
-
-    @Deactivate
-    public void deactivate() {
-        LOG.debug("Stopping ZookeeperDiscovery");
-        endpointListener.stop();
-        endpointListenerTracker.close();
-        imManager.close();
-        repository.close();
-    }
-
-}
diff --git a/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/ZookeeperEndpointListener.java b/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/ZookeeperEndpointListener.java
new file mode 100644
index 0000000..37fdada
--- /dev/null
+++ b/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/ZookeeperEndpointListener.java
@@ -0,0 +1,120 @@
+/**
+ * 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.aries.rsa.discovery.zookeeper;
+
+import java.io.ByteArrayInputStream;
+import java.util.List;
+
+import org.apache.aries.rsa.spi.EndpointDescriptionParser;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.KeeperException.ConnectionLossException;
+import org.apache.zookeeper.KeeperException.NoNodeException;
+import org.apache.zookeeper.KeeperException.SessionExpiredException;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.data.Stat;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Reference;
+import org.osgi.service.remoteserviceadmin.EndpointDescription;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Component(immediate = true)
+public class ZookeeperEndpointListener {
+    private static final Logger LOG = LoggerFactory.getLogger(ZookeeperEndpointListener.class);
+    
+    @Reference
+    private ZooKeeper zk;
+    
+    @Reference
+    private EndpointDescriptionParser parser;
+    
+    @Reference
+    private InterestManager listener;
+    
+    @Reference
+    private ZookeeperEndpointPublisher publisher;
+    
+    public ZookeeperEndpointListener() {
+    }
+    
+    public ZookeeperEndpointListener(ZooKeeper zk, EndpointDescriptionParser parser, InterestManager listener) {
+        this.zk = zk;
+        this.parser = parser;
+        this.listener = listener;
+        activate();
+    }
+    
+    @Activate
+    public void activate() {
+        watchRecursive(ZookeeperEndpointPublisher.PATH_PREFIX);
+    }
+    
+    private void process(WatchedEvent event) {
+        String path = event.getPath();
+        LOG.info("Received event {}", event);
+        switch (event.getType()) {
+        case NodeCreated:
+        case NodeDataChanged:
+        case NodeChildrenChanged:
+            watchRecursive(path);
+            break;
+        case NodeDeleted:
+            listener.handleRemoved(path);
+            break;
+        default:
+            break;
+        }
+    }
+
+    private void watchRecursive(String path) {
+        LOG.info("Watching {}", path);
+        try {
+            EndpointDescription endpoint = read(path);
+            if (endpoint != null) {
+                listener.handleChanged(path, endpoint);
+            }
+            List<String> children = zk.getChildren(path, this::process);
+            if (children == null) {
+                return;
+            }
+            for (String child : children) {
+                String childPath = (path.endsWith("/") ? path : path + "/") + child;
+                watchRecursive(childPath);
+            }
+        } catch (NoNodeException | SessionExpiredException | ConnectionLossException e) {
+            // NoNodeException happens when a node was removed
+            LOG.debug(e.getMessage(), e);
+        } catch (Exception e) {
+            LOG.info(e.getMessage(), e);
+        }
+    }
+
+    EndpointDescription read(String path) throws KeeperException, InterruptedException {
+        Stat stat = new Stat();
+        byte[] data = zk.getData(path, this::process, stat);
+        if (data == null || data.length == 0) {
+            return null;
+        } else {
+            return parser.readEndpoint(new ByteArrayInputStream(data));
+        }
+    }
+
+}
diff --git a/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/ZookeeperEndpointPublisher.java b/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/ZookeeperEndpointPublisher.java
new file mode 100644
index 0000000..7757396
--- /dev/null
+++ b/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/ZookeeperEndpointPublisher.java
@@ -0,0 +1,187 @@
+/**
+ * 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.aries.rsa.discovery.zookeeper;
+
+import static java.util.Arrays.asList;
+import static java.util.stream.Collectors.toList;
+
+import java.io.ByteArrayOutputStream;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.aries.rsa.spi.EndpointDescriptionParser;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.KeeperException.NoNodeException;
+import org.apache.zookeeper.KeeperException.NodeExistsException;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.ZooKeeper;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Reference;
+import org.osgi.service.remoteserviceadmin.EndpointDescription;
+import org.osgi.service.remoteserviceadmin.EndpointEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Component(service = ZookeeperEndpointPublisher.class)
+public class ZookeeperEndpointPublisher {
+    public static final String PATH_PREFIX = "/osgi/service_registry";
+    private static final Logger LOG = LoggerFactory.getLogger(ZookeeperEndpointPublisher.class);
+    private final Map<Integer, String> typeNames = new HashMap<>();
+    
+    @Reference
+    private ZooKeeper zk;
+    
+    @Reference
+    private EndpointDescriptionParser parser;
+    
+    public ZookeeperEndpointPublisher() {
+        typeNames.put(EndpointEvent.ADDED, "added");
+        typeNames.put(EndpointEvent.MODIFIED, "modified");
+        typeNames.put(EndpointEvent.MODIFIED_ENDMATCH, "modified");
+        typeNames.put(EndpointEvent.REMOVED, "removed");
+    }
+    
+    public ZookeeperEndpointPublisher(ZooKeeper zk, EndpointDescriptionParser parser) {
+        this();
+        this.zk = zk;
+        this.parser = parser;
+    }
+
+    @Activate 
+    public void activate() {
+        try {
+            createPath(PATH_PREFIX);
+        } catch (Exception e) {
+            throw new IllegalStateException("Unable to create base path");
+        }
+    }
+
+    public void endpointChanged(EndpointEvent event) {
+        try {
+            EndpointDescription endpoint = event.getEndpoint();
+            switch (event.getType()) {
+            case EndpointEvent.ADDED:
+                add(endpoint);
+                break;
+            case EndpointEvent.MODIFIED:
+            case EndpointEvent.MODIFIED_ENDMATCH:
+                modify(endpoint);
+                break;
+            case EndpointEvent.REMOVED:
+                remove(endpoint);
+                break;
+            default:
+                break;
+            }
+        } catch (Exception e) {
+            logException(typeNames.get(event.getType()), event.getEndpoint(), e);
+        }
+    }
+    
+    private void logException(String operation, EndpointDescription endpoint, Exception ex) {
+        String msg = String.format("Exception during %s of endpoint %s", operation, endpoint.getId());
+        LOG.error(msg, ex);
+    }
+
+
+    private void add(EndpointDescription endpoint) throws KeeperException, InterruptedException  {
+        String path = getZooKeeperPath(endpoint.getId());
+        LOG.info("Exporting path: {}, Endpoint: {}", path, endpoint);
+        createBasePath();
+        byte[] data = getData(endpoint);
+        createEphemeralNode(path, data);
+    }
+
+    private void modify(EndpointDescription endpoint) throws KeeperException, InterruptedException {
+        String path = getZooKeeperPath(endpoint.getId());
+        LOG.info("Changing endpoint in zookeeper. Endpoint: {}, Path: {}", endpoint, path);
+        createBasePath();
+        zk.setData(path, getData(endpoint), -1);
+    }
+    
+    private void remove(EndpointDescription endpoint) throws KeeperException, InterruptedException {
+        String path = getZooKeeperPath(endpoint.getId());
+        LOG.info("Removing endpoint in zookeeper. Endpoint: {}, Path: {}", endpoint, path);
+        zk.delete(path, -1);
+    }
+    
+    private boolean notEmpty(String part) {
+        return part != null && !part.isEmpty();
+    }
+
+    static String getZooKeeperPath(String name) {
+        String escaped = name.replace('/', '#');
+        return name == null || name.isEmpty() ? PATH_PREFIX : PATH_PREFIX + '/' + escaped;
+    }
+
+    private void createBasePath() throws KeeperException, InterruptedException {
+        String path = ZookeeperEndpointPublisher.getZooKeeperPath("");
+        createPath(path);
+    }
+
+    private byte[] getData(EndpointDescription epd) {
+        ByteArrayOutputStream bos = new ByteArrayOutputStream();
+        parser.writeEndpoint(epd, bos);
+        return bos.toByteArray();
+    }
+    
+    private void createEphemeralNode(String fullPath, byte[] data) throws KeeperException, InterruptedException {
+        try {
+            zk.create(fullPath, data, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
+        } catch (NodeExistsException nee) {
+            // this sometimes happens after a ZooKeeper node dies and the ephemeral node
+            // that belonged to the old session was not yet deleted. We need to make our
+            // session the owner of the node so it won't get deleted automatically -
+            // we do this by deleting and recreating it ourselves.
+            LOG.info("node for endpoint already exists, recreating: {}", fullPath);
+            try {
+                zk.delete(fullPath, -1);
+            } catch (NoNodeException nne) {
+                // it's a race condition, but as long as it got deleted - it's ok
+            }
+            zk.create(fullPath, data, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
+        }
+    }
+    
+    private void createPath(String path) throws KeeperException, InterruptedException {
+        List<String> parts = asList(path.split("/")).stream()
+                .filter(this::notEmpty)
+                .collect(toList());
+        StringBuilder current = new StringBuilder();
+        for (String part : parts) {
+            current.append('/');
+            current.append(part);
+            createNode(current.toString());
+        }
+    }
+
+    private void createNode(String path) throws KeeperException, InterruptedException {
+        try {
+            if (zk.exists(path, false) == null) {
+                zk.create(path, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+            }
+        } catch (NodeExistsException nee) {
+            // it's not the first node with this path to ever exist - that's normal
+        }
+    }
+
+}
diff --git a/discovery/local/src/test/java/org/apache/aries/rsa/discovery/local/ActivatorTest.java b/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/package-info.java
similarity index 76%
rename from discovery/local/src/test/java/org/apache/aries/rsa/discovery/local/ActivatorTest.java
rename to discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/package-info.java
index 478e41b..1e835fe 100644
--- a/discovery/local/src/test/java/org/apache/aries/rsa/discovery/local/ActivatorTest.java
+++ b/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/package-info.java
@@ -16,12 +16,9 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.aries.rsa.discovery.local;
-
-import junit.framework.TestCase;
-
-public class ActivatorTest extends TestCase {
-
-    public void testActivator() throws Exception {
-    }
-}
+@org.osgi.annotation.bundle.Capability( //
+        namespace = "osgi.remoteserviceadmin.discovery", //
+        attribute = {"protocols:List<String>=zookeeper"}, //
+        version = "1.1.0"
+)
+package org.apache.aries.rsa.discovery.zookeeper;
diff --git a/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/publish/PublishingEndpointListener.java b/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/publish/PublishingEndpointListener.java
deleted file mode 100644
index 7f352a7..0000000
--- a/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/publish/PublishingEndpointListener.java
+++ /dev/null
@@ -1,118 +0,0 @@
-/**
- * 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.aries.rsa.discovery.zookeeper.publish;
-
-import java.util.Dictionary;
-import java.util.Hashtable;
-
-import org.apache.aries.rsa.discovery.zookeeper.ZooKeeperDiscovery;
-import org.apache.aries.rsa.discovery.zookeeper.repository.ZookeeperEndpointRepository;
-import org.osgi.framework.BundleContext;
-import org.osgi.framework.Constants;
-import org.osgi.framework.ServiceRegistration;
-import org.osgi.service.remoteserviceadmin.EndpointDescription;
-import org.osgi.service.remoteserviceadmin.EndpointEvent;
-import org.osgi.service.remoteserviceadmin.EndpointEventListener;
-import org.osgi.service.remoteserviceadmin.EndpointListener;
-import org.osgi.service.remoteserviceadmin.RemoteConstants;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Listens for local EndpointEvents using old and new style listeners and publishes changes to 
- * the ZooKeeperEndpointRepository
- */
-@SuppressWarnings("deprecation")
-public class PublishingEndpointListener implements EndpointEventListener, EndpointListener {
-
-    private static final Logger LOG = LoggerFactory.getLogger(PublishingEndpointListener.class);
-
-    private ServiceRegistration<?> listenerReg;
-    private ZookeeperEndpointRepository repository;
-
-    public PublishingEndpointListener(ZookeeperEndpointRepository repository) {
-        this.repository = repository;
-    }
-    
-    public void start(BundleContext bctx) {
-        Dictionary<String, String> props = new Hashtable<>();
-        String uuid = bctx.getProperty(Constants.FRAMEWORK_UUID);
-        props.put(EndpointEventListener.ENDPOINT_LISTENER_SCOPE, 
-                  String.format("(&(%s=*)(%s=%s))", Constants.OBJECTCLASS, 
-                                RemoteConstants.ENDPOINT_FRAMEWORK_UUID, uuid));
-        props.put(ZooKeeperDiscovery.DISCOVERY_ZOOKEEPER_ID, "true");
-        String[] ifAr = {EndpointEventListener.class.getName(), EndpointListener.class.getName()};
-        listenerReg = bctx.registerService(ifAr, this, props);
-    }
-    
-    public void stop() {
-        if (listenerReg != null) {
-            listenerReg.unregister();
-            listenerReg = null;
-        }
-    }
-
-    @Override
-    public void endpointChanged(EndpointEvent event, String filter) {
-        EndpointDescription endpoint = event.getEndpoint();
-        switch (event.getType()) {
-        case EndpointEvent.ADDED:
-            endpointAdded(endpoint, filter);
-            break;
-        case EndpointEvent.REMOVED:
-            endpointRemoved(endpoint, filter);
-            break;
-        case EndpointEvent.MODIFIED:
-            endpointModified(endpoint, filter);
-            break;
-        }
-    }
-    
-    private void endpointModified(EndpointDescription endpoint, String filter) {
-        try {
-            repository.modify(endpoint);
-        } catch (Exception ex) {
-            logException("modification", endpoint, ex);
-        }
-    }
-
-    @Override
-    public void endpointAdded(EndpointDescription endpoint, String matchedFilter) {
-        try {
-            repository.add(endpoint);
-        } catch (Exception ex) {
-            logException("adding", endpoint, ex);
-        }
-    }
-
-    @Override
-    public void endpointRemoved(EndpointDescription endpoint, String matchedFilter) {
-        try {
-            repository.remove(endpoint);
-        } catch (Exception ex) {
-            logException("removal", endpoint, ex);
-        }
-    }
-    
-    private void logException(String operation, EndpointDescription endpoint, Exception ex) {
-        String msg = String.format("Exception during %s of endpoint %s", operation, endpoint.getId());
-        LOG.error(msg, ex);
-    }
-
-}
diff --git a/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/repository/ZookeeperEndpointRepository.java b/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/repository/ZookeeperEndpointRepository.java
deleted file mode 100644
index 5b0d66b..0000000
--- a/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/repository/ZookeeperEndpointRepository.java
+++ /dev/null
@@ -1,260 +0,0 @@
-/**
- * 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.aries.rsa.discovery.zookeeper.repository;
-
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.Closeable;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
-import org.apache.aries.rsa.discovery.endpoint.EndpointDescriptionParser;
-import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.KeeperException.ConnectionLossException;
-import org.apache.zookeeper.KeeperException.NoNodeException;
-import org.apache.zookeeper.KeeperException.NodeExistsException;
-import org.apache.zookeeper.KeeperException.SessionExpiredException;
-import org.apache.zookeeper.WatchedEvent;
-import org.apache.zookeeper.Watcher;
-import org.apache.zookeeper.ZooDefs.Ids;
-import org.apache.zookeeper.ZooKeeper;
-import org.apache.zookeeper.data.Stat;
-import org.osgi.service.remoteserviceadmin.EndpointDescription;
-import org.osgi.service.remoteserviceadmin.EndpointEvent;
-import org.osgi.service.remoteserviceadmin.EndpointEventListener;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class ZookeeperEndpointRepository implements Closeable, Watcher {
-    private static final Logger LOG = LoggerFactory.getLogger(ZookeeperEndpointRepository.class);
-    private final ZooKeeper zk;
-    private final EndpointDescriptionParser parser;
-    private EndpointEventListener listener;
-    public static final String PATH_PREFIX = "/osgi/service_registry";
-    
-    private Map<String, EndpointDescription> nodes = new ConcurrentHashMap<>();
-    
-    public ZookeeperEndpointRepository(ZooKeeper zk) {
-        this.zk = zk;
-        this.parser = new EndpointDescriptionParser();
-        try {
-            createPath(PATH_PREFIX);
-        } catch (Exception e) {
-            throw new IllegalStateException("Unable to create base path");
-        }
-        this.registerWatcher();
-    }
-
-    public void addListener(EndpointEventListener listener) {
-        this.listener = listener;
-    }
-
-    /**
-     * Read current endpoint stored at a znode
-     * 
-     * @param path
-     * @return
-     */
-    public EndpointDescription read(String path) {
-        return nodes.get(path);
-    }
-
-    public void add(EndpointDescription endpoint) throws KeeperException, InterruptedException  {
-        String path = getZooKeeperPath(endpoint.getId());
-        LOG.info("Exporting endpoint to zookeeper. Endpoint: {}, Path: {}", endpoint, path);
-        createBasePath();
-        createEphemeralNode(path, getData(endpoint));
-    }
-
-    public void modify(EndpointDescription endpoint) throws KeeperException, InterruptedException {
-        String path = getZooKeeperPath(endpoint.getId());
-        LOG.info("Changing endpoint in zookeeper. Endpoint: {}, Path: {}", endpoint, path);
-        createBasePath();
-        zk.setData(path, getData(endpoint), -1);
-    }
-    
-    public void remove(EndpointDescription endpoint) throws InterruptedException, KeeperException {
-        String path = getZooKeeperPath(endpoint.getId());
-        LOG.info("Removing endpoint in zookeeper. Endpoint: {}, Path: {}", endpoint, path);
-        zk.delete(path, -1);
-    }
-    
-    public Collection<EndpointDescription> getAll() {
-        return nodes.values();
-    }
-
-    /**
-     * Removes nulls and empty strings from the given string array.
-     *
-     * @param strings an array of strings
-     * @return a new array containing the non-null and non-empty
-     *         elements of the original array in the same order
-     */
-    public static List<String> removeEmpty(List<String> strings) {
-        List<String> result = new ArrayList<>();
-        if (strings == null) {
-            return result;
-        }
-        for (String s : strings) {
-            if (s != null && !s.isEmpty()) {
-                result.add(s);
-            }
-        }
-        return result;
-    }
-
-    public static String getZooKeeperPath(String name) {
-        String escaped = name.replace('/', '#');
-        return name == null || name.isEmpty() ? PATH_PREFIX : PATH_PREFIX + '/' + escaped;
-    }
-
-    @Override
-    public void process(WatchedEvent event) {
-        LOG.info("Received event {}", event);
-        switch (event.getType()) {
-        case NodeCreated:
-        case NodeDataChanged:
-        case NodeChildrenChanged:
-            watchRecursive(event.getPath());
-            break;
-        case NodeDeleted:
-            handleRemoved(event.getPath());
-            break;
-        default:
-            break;
-        }
-    }
-
-    @Override
-    public void close() {
-        nodes.clear();
-    }
-
-    private void createBasePath() throws KeeperException, InterruptedException {
-        String path = ZookeeperEndpointRepository.getZooKeeperPath("");
-        createPath(path);
-    }
-
-    private void registerWatcher() {
-        try {
-            watchRecursive(ZookeeperEndpointRepository.PATH_PREFIX);
-        } catch (Exception e) {
-            LOG.info(e.getMessage(), e);
-        }
-    }
-
-    /**
-     * TODO Check if we handle connection losses correctly
-     * @param path
-     */
-    private void watchRecursive(String path) {
-        LOG.info("Watching {}", path);
-        try {
-            handleZNodeChanged(path);
-            List<String> children = zk.getChildren(path, this);
-            if (children == null) {
-                return;
-            }
-            for (String child : children) {
-                String childPath = (path.endsWith("/") ? path : path + "/") + child;
-                watchRecursive(childPath);
-            }
-        } catch (NoNodeException | SessionExpiredException | ConnectionLossException e) {
-            // NoNodeException happens when a node was removed
-            LOG.debug(e.getMessage(), e);
-        } catch (Exception e) {
-            LOG.info(e.getMessage(), e);
-        }
-    }
-
-    private byte[] getData(EndpointDescription epd) {
-        ByteArrayOutputStream bos = new ByteArrayOutputStream();
-        parser.writeEndpoint(epd, bos);
-        return bos.toByteArray();
-    }
-    
-    private void createEphemeralNode(String fullPath, byte[] data) throws KeeperException, InterruptedException {
-        try {
-            zk.create(fullPath, data, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
-        } catch (NodeExistsException nee) {
-            // this sometimes happens after a ZooKeeper node dies and the ephemeral node
-            // that belonged to the old session was not yet deleted. We need to make our
-            // session the owner of the node so it won't get deleted automatically -
-            // we do this by deleting and recreating it ourselves.
-            LOG.info("node for endpoint already exists, recreating: {}", fullPath);
-            try {
-                zk.delete(fullPath, -1);
-            } catch (NoNodeException nne) {
-                // it's a race condition, but as long as it got deleted - it's ok
-            }
-            zk.create(fullPath, data, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
-        }
-    }
-    
-    private void createPath(String path) throws KeeperException, InterruptedException {
-        StringBuilder current = new StringBuilder();
-        List<String> parts = ZookeeperEndpointRepository.removeEmpty(Arrays.asList(path.split("/")));
-        for (String part : parts) {
-            current.append('/');
-            current.append(part);
-            try {
-                if (zk.exists(current.toString(), false) == null) {
-                    zk.create(current.toString(), new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
-                }
-            } catch (NodeExistsException nee) {
-                // it's not the first node with this path to ever exist - that's normal
-            }
-        }
-    }
-
-    private void handleZNodeChanged(String path) throws KeeperException, InterruptedException {
-        Stat stat = new Stat();
-        byte[] data = zk.getData(path, this, stat);
-        if (data == null || data.length == 0) {
-            return;
-        }
-        EndpointDescription endpoint = parser.readEndpoint(new ByteArrayInputStream(data));
-        if (endpoint != null) {
-            handleChanged(path, endpoint);
-        }
-    }
-
-    private void handleRemoved(String path) {
-        EndpointDescription endpoint = nodes.remove(path);
-        EndpointEvent event = new EndpointEvent(EndpointEvent.REMOVED, endpoint);
-        if (listener != null) {
-            listener.endpointChanged(event, null);
-        }
-    }
-
-    private void handleChanged(String path, EndpointDescription endpoint) {
-        EndpointDescription old = nodes.put(path, endpoint);
-        int type = old == null ? EndpointEvent.ADDED : EndpointEvent.MODIFIED;
-        EndpointEvent event = new EndpointEvent(type, endpoint);
-        if (listener != null) {
-            listener.endpointChanged(event, null);
-        }
-    }
-
-}
diff --git a/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/subscribe/EndpointListenerTracker.java b/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/subscribe/EndpointListenerTracker.java
deleted file mode 100644
index 26910cd..0000000
--- a/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/subscribe/EndpointListenerTracker.java
+++ /dev/null
@@ -1,72 +0,0 @@
-/**
- * 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.aries.rsa.discovery.zookeeper.subscribe;
-
-import org.osgi.framework.BundleContext;
-import org.osgi.framework.Filter;
-import org.osgi.framework.FrameworkUtil;
-import org.osgi.framework.InvalidSyntaxException;
-import org.osgi.framework.ServiceReference;
-import org.osgi.service.remoteserviceadmin.EndpointEventListener;
-import org.osgi.service.remoteserviceadmin.EndpointListener;
-import org.osgi.util.tracker.ServiceTracker;
-
-/**
- * Tracks EndpointListeners and EndpointEventListeners. Delegates to InterestManager to handle them
- */
-@SuppressWarnings({ "rawtypes", "deprecation", "unchecked" })
-public class EndpointListenerTracker extends ServiceTracker {
-    private final InterestManager imManager;
-
-    public EndpointListenerTracker(BundleContext bctx, InterestManager imManager) {
-        super(bctx, getFilter(), null);
-        this.imManager = imManager;
-    }
-    
-    private static Filter getFilter() {
-        String filterSt = String.format("(|(objectClass=%s)(objectClass=%s))", EndpointEventListener.class.getName(), 
-                EndpointListener.class.getName());
-        try {
-            return FrameworkUtil.createFilter(filterSt);
-        } catch (InvalidSyntaxException e) {
-            throw new IllegalArgumentException(e.getMessage(), e);
-        }
-    }
-
-    @Override
-    public Object addingService(ServiceReference sref) {
-        Object epListener = super.addingService(sref);
-        imManager.addInterest(sref, epListener);
-        return epListener;
-    }
-
-    @Override
-    public void modifiedService(ServiceReference sref, Object epListener) {
-        // called when an EndpointListener updates its service properties,
-        // e.g. when its interest scope is expanded/reduced
-        imManager.addInterest(sref, epListener);
-    }
-
-    @Override
-    public void removedService(ServiceReference sref, Object epListener) {
-        imManager.removeInterest(sref);
-        super.removedService(sref, epListener);
-    }
-
-}
diff --git a/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/InterestManagerTest.java b/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/InterestManagerTest.java
new file mode 100644
index 0000000..985f1b0
--- /dev/null
+++ b/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/InterestManagerTest.java
@@ -0,0 +1,75 @@
+/**
+ * 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.aries.rsa.discovery.zookeeper;
+
+import static org.easymock.EasyMock.expect;
+import static org.junit.Assert.assertEquals;
+
+import org.easymock.EasyMock;
+import org.easymock.IMocksControl;
+import org.junit.Test;
+import org.osgi.framework.ServiceReference;
+import org.osgi.service.remoteserviceadmin.EndpointEventListener;
+
+public class InterestManagerTest {
+
+    @Test
+    public void testEndpointListenerTrackerCustomizer() {
+        IMocksControl c = EasyMock.createControl();
+        ServiceReference<EndpointEventListener> sref = createService(c, "(objectClass=mine)");
+        ServiceReference<EndpointEventListener> sref2 = createService(c, "(objectClass=mine)");
+        EndpointEventListener epListener1 = c.createMock(EndpointEventListener.class); 
+        EndpointEventListener epListener2 = c.createMock(EndpointEventListener.class); 
+
+        c.replay();
+
+        InterestManager im = new InterestManager();
+        // sref has no scope -> nothing should happen
+        assertEquals(0, im.getInterests().size());
+
+        im.bindEndpointEventListener(sref, epListener1);
+        assertEquals(1, im.getInterests().size());
+
+        im.bindEndpointEventListener(sref, epListener1);
+        assertEquals(1, im.getInterests().size());
+
+        im.bindEndpointEventListener(sref2, epListener2);
+        assertEquals(2, im.getInterests().size());
+
+        im.unbindEndpointEventListener(sref);
+        assertEquals(1, im.getInterests().size());
+
+        im.unbindEndpointEventListener(sref);
+        assertEquals(1, im.getInterests().size());
+
+        im.unbindEndpointEventListener(sref2);
+        assertEquals(0, im.getInterests().size());
+
+        c.verify();
+    }
+
+    @SuppressWarnings("unchecked")
+    private ServiceReference<EndpointEventListener> createService(IMocksControl c, String scope) {
+        ServiceReference<EndpointEventListener> sref = c.createMock(ServiceReference.class);
+        expect(sref.getProperty(EndpointEventListener.ENDPOINT_LISTENER_SCOPE)).andReturn(scope).atLeastOnce();
+        expect(sref.getProperty(ClientManager.DISCOVERY_ZOOKEEPER_ID)).andReturn(null).atLeastOnce();
+        return sref;
+    }
+
+}
diff --git a/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/PublishingEndpointListenerTest.java b/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/PublishingEndpointListenerTest.java
new file mode 100644
index 0000000..815c42a
--- /dev/null
+++ b/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/PublishingEndpointListenerTest.java
@@ -0,0 +1,64 @@
+/**
+ * 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.aries.rsa.discovery.zookeeper;
+
+import static org.mockito.Mockito.verify;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.zookeeper.KeeperException;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+import org.osgi.framework.Constants;
+import org.osgi.service.remoteserviceadmin.EndpointDescription;
+import org.osgi.service.remoteserviceadmin.EndpointEvent;
+import org.osgi.service.remoteserviceadmin.RemoteConstants;
+
+@RunWith(MockitoJUnitRunner.class)
+public class PublishingEndpointListenerTest {
+    @Mock
+    ZookeeperEndpointPublisher repository;
+
+    @InjectMocks
+    PublishingEndpointListener eli;
+
+    @Test
+    public void testEndpointRemovalAdding() throws KeeperException, InterruptedException {
+        EndpointDescription endpoint = createEndpoint();
+        EndpointEvent event1 = new EndpointEvent(EndpointEvent.ADDED, endpoint);
+        eli.endpointChanged(event1, null);
+        EndpointEvent event2 = new EndpointEvent(EndpointEvent.REMOVED, endpoint);
+        eli.endpointChanged(event2, null);
+        
+        verify(repository).endpointChanged(event1);
+        verify(repository).endpointChanged(event2);
+    }
+
+    private EndpointDescription createEndpoint() {
+        Map<String, Object> props = new HashMap<>();
+        props.put(Constants.OBJECTCLASS, new String[] {"myClass"});
+        props.put(RemoteConstants.ENDPOINT_ID, "http://google.de:80/test/sub");
+        props.put(RemoteConstants.SERVICE_IMPORTED_CONFIGS, "myConfig");
+        return new EndpointDescription(props);
+    }
+}
diff --git a/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/repository/ZookeeperEndpointRepositoryTest.java b/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/ZookeeperDiscoveryTest.java
similarity index 66%
rename from discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/repository/ZookeeperEndpointRepositoryTest.java
rename to discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/ZookeeperDiscoveryTest.java
index a65696b..02f6093 100644
--- a/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/repository/ZookeeperEndpointRepositoryTest.java
+++ b/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/ZookeeperDiscoveryTest.java
@@ -16,10 +16,10 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.aries.rsa.discovery.zookeeper.repository;
+package org.apache.aries.rsa.discovery.zookeeper;
 
+import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.hamcrest.Matchers.equalTo;
-import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThat;
 
@@ -35,9 +35,10 @@
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.aries.rsa.discovery.endpoint.EndpointDescriptionParserImpl;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.WatchedEvent;
-import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.Watcher.Event.KeeperState;
 import org.apache.zookeeper.ZooKeeper;
 import org.apache.zookeeper.server.NIOServerCnxnFactory;
 import org.apache.zookeeper.server.ServerCnxnFactory;
@@ -45,46 +46,35 @@
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.runners.MockitoJUnitRunner;
 import org.osgi.framework.Constants;
+import org.osgi.framework.ServiceReference;
 import org.osgi.service.remoteserviceadmin.EndpointDescription;
 import org.osgi.service.remoteserviceadmin.EndpointEvent;
 import org.osgi.service.remoteserviceadmin.EndpointEventListener;
 import org.osgi.service.remoteserviceadmin.RemoteConstants;
 
-public class ZookeeperEndpointRepositoryTest {
-    
+@RunWith(MockitoJUnitRunner.class)
+public class ZookeeperDiscoveryTest {
+    final Semaphore semConnected = new Semaphore(0);
+    final Semaphore sem = new Semaphore(0);
     private ZooKeeperServer server;
     private ZooKeeper zk;
     private ServerCnxnFactory factory;
     private List<EndpointEvent> events = new ArrayList<>();
+    @Mock
+    private ServiceReference<EndpointEventListener> sref;
 
     @Before
     public void before() throws IOException, InterruptedException, KeeperException {
-        File target = new File("target");
-        File zookeeperDir = new File(target, "zookeeper");
-        server = new ZooKeeperServer(zookeeperDir, zookeeperDir, 2000);
-        factory = new NIOServerCnxnFactory();
-        int clientPort = getClientPort();
-        factory.configure(new InetSocketAddress(clientPort), 10);
-        factory.startup(server);
-        Watcher watcher = new Watcher() {
-
-            @Override
-            public void process(WatchedEvent event) {
-                System.out.println(event);
-            }
-            
-        };
-        zk = new ZooKeeper("localhost:" + server.getClientPort(), 1000, watcher);
+        startZookeeperServer();
+        zk = new ZooKeeper("localhost:" + server.getClientPort(), 1000, this::process);
         printNodes("/");
     }
     
-    private int getClientPort() throws IOException {
-        try (ServerSocket serverSocket = new ServerSocket(0)) {
-            return serverSocket.getLocalPort();
-        }
-    }
-
     @After
     public void after() throws InterruptedException {
         //zk.close(); // Seems to cause SessionTimeout error 
@@ -93,44 +83,62 @@
 
     @Test
     public void test() throws IOException, URISyntaxException, KeeperException, InterruptedException {
-        final Semaphore sem = new Semaphore(0);
-        EndpointEventListener listener = new EndpointEventListener() {
-            
-            @Override
-            public void endpointChanged(EndpointEvent event, String filter) {
-                events.add(event);
-                sem.release();
-            }
-        };
-        ZookeeperEndpointRepository repository = new ZookeeperEndpointRepository(zk);
-        repository.addListener(listener);
+        EndpointDescriptionParserImpl parser = new EndpointDescriptionParserImpl();
+        ZookeeperEndpointPublisher repository = new ZookeeperEndpointPublisher(zk, parser);
+        repository.activate();
+        InterestManager im = new InterestManager();
+        
+        String scope = "("+ Constants.OBJECTCLASS +"=*)";
+        Mockito.when(sref.getProperty(Mockito.eq(EndpointEventListener.ENDPOINT_LISTENER_SCOPE))).thenReturn(scope);
+        im.bindEndpointEventListener(sref, this::onEndpointChanged);
+        ZookeeperEndpointListener zklistener = new ZookeeperEndpointListener(zk, parser, im);
+        
+        assertThat(semConnected.tryAcquire(1, SECONDS), equalTo(true));
         
         EndpointDescription endpoint = createEndpoint();
-        repository.add(endpoint);
+        repository.endpointChanged(new EndpointEvent(EndpointEvent.ADDED, endpoint));
         
-        assertThat(sem.tryAcquire(1000, TimeUnit.SECONDS), equalTo(true));
-
+        assertThat(sem.tryAcquire(100, SECONDS), equalTo(true));
+    
         String path = "/osgi/service_registry/http:##test.de#service1";
-        EndpointDescription ep2 = repository.read(path);
+        EndpointDescription ep2 = zklistener.read(path);
         assertNotNull(ep2);
 
-        repository.remove(endpoint);
-
+        repository.endpointChanged(new EndpointEvent(EndpointEvent.REMOVED, endpoint));
+    
         assertThat(sem.tryAcquire(1000, TimeUnit.SECONDS), equalTo(true));
         assertThat(events.get(0).getType(), equalTo(EndpointEvent.ADDED));
         assertThat(events.get(1).getType(), equalTo(EndpointEvent.REMOVED));
         assertThat(events.get(0).getEndpoint(), equalTo(endpoint));
         assertThat(events.get(1).getEndpoint(), equalTo(endpoint));
-        
-        repository.close();
+    }
+
+    private void process(WatchedEvent event) {
+        if (event.getState() == KeeperState.SyncConnected) {
+            semConnected.release();
+        }
+        System.out.println(event);
+    }
+
+    private void startZookeeperServer() throws IOException, InterruptedException {
+        File target = new File("target");
+        File zookeeperDir = new File(target, "zookeeper");
+        server = new ZooKeeperServer(zookeeperDir, zookeeperDir, 2000);
+        factory = new NIOServerCnxnFactory();
+        int clientPort = getClientPort();
+        factory.configure(new InetSocketAddress(clientPort), 10);
+        factory.startup(server);
     }
     
-    @Test
-    public void testGetZooKeeperPath() {
-        assertEquals(ZookeeperEndpointRepository.PATH_PREFIX + '/' + "http:##org.example.Test",
-            ZookeeperEndpointRepository.getZooKeeperPath("http://org.example.Test"));
+    private int getClientPort() throws IOException {
+        try (ServerSocket serverSocket = new ServerSocket(0)) {
+            return serverSocket.getLocalPort();
+        }
+    }
 
-        assertEquals(ZookeeperEndpointRepository.PATH_PREFIX, ZookeeperEndpointRepository.getZooKeeperPath(""));
+    private void onEndpointChanged(EndpointEvent event, String filter) {
+        events.add(event);
+        sem.release();
     }
 
     private EndpointDescription createEndpoint() {
@@ -143,7 +151,7 @@
         return endpoint;
     }
 
-    public void printNodes(String path) throws KeeperException, InterruptedException {
+    private void printNodes(String path) throws KeeperException, InterruptedException {
         List<String> children = zk.getChildren(path, false);
         for (String child : children) {
             String newPath = path.endsWith("/") ? path : path + "/";
diff --git a/discovery/local/src/test/java/org/apache/aries/rsa/discovery/local/ActivatorTest.java b/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/ZookeeperEndpointPublisherPathTest.java
similarity index 60%
copy from discovery/local/src/test/java/org/apache/aries/rsa/discovery/local/ActivatorTest.java
copy to discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/ZookeeperEndpointPublisherPathTest.java
index 478e41b..31441d3 100644
--- a/discovery/local/src/test/java/org/apache/aries/rsa/discovery/local/ActivatorTest.java
+++ b/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/ZookeeperEndpointPublisherPathTest.java
@@ -16,12 +16,20 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.aries.rsa.discovery.local;
+package org.apache.aries.rsa.discovery.zookeeper;
 
-import junit.framework.TestCase;
+import static org.junit.Assert.assertEquals;
 
-public class ActivatorTest extends TestCase {
+import org.junit.Test;
 
-    public void testActivator() throws Exception {
+public class ZookeeperEndpointPublisherPathTest {
+    
+    @Test
+    public void testGetZooKeeperPath() {
+        assertEquals(ZookeeperEndpointPublisher.PATH_PREFIX + '/' + "http:##org.example.Test",
+            ZookeeperEndpointPublisher.getZooKeeperPath("http://org.example.Test"));
+
+        assertEquals(ZookeeperEndpointPublisher.PATH_PREFIX, ZookeeperEndpointPublisher.getZooKeeperPath(""));
     }
+    
 }
diff --git a/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/publish/PublishingEndpointListenerTest.java b/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/publish/PublishingEndpointListenerTest.java
deleted file mode 100644
index 3884c95..0000000
--- a/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/publish/PublishingEndpointListenerTest.java
+++ /dev/null
@@ -1,86 +0,0 @@
-/**
- * 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.aries.rsa.discovery.zookeeper.publish;
-
-import static org.easymock.EasyMock.expect;
-
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.aries.rsa.discovery.zookeeper.repository.ZookeeperEndpointRepository;
-import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.ZooDefs.Ids;
-import org.apache.zookeeper.ZooKeeper;
-import org.easymock.EasyMock;
-import org.easymock.IMocksControl;
-import org.osgi.framework.Constants;
-import org.osgi.service.remoteserviceadmin.EndpointDescription;
-import org.osgi.service.remoteserviceadmin.EndpointEvent;
-import org.osgi.service.remoteserviceadmin.RemoteConstants;
-
-import junit.framework.TestCase;
-
-public class PublishingEndpointListenerTest extends TestCase {
-
-    private static final String ENDPOINT_PATH = "/osgi/service_registry/http:##google.de:80#test#sub";
-
-    public void testEndpointRemovalAdding() throws KeeperException, InterruptedException {
-        IMocksControl c = EasyMock.createNiceControl();
-
-        ZooKeeper zk = c.createMock(ZooKeeper.class);
-
-        String path = ENDPOINT_PATH;
-        expectCreated(zk, path);
-        expectDeleted(zk, path);
-
-        c.replay();
-
-        ZookeeperEndpointRepository repository = new ZookeeperEndpointRepository(zk);
-        PublishingEndpointListener eli = new PublishingEndpointListener(repository);
-        EndpointDescription endpoint = createEndpoint();
-        eli.endpointChanged(new EndpointEvent(EndpointEvent.ADDED, endpoint), null);
-        eli.endpointChanged(new EndpointEvent(EndpointEvent.ADDED, endpoint), null); // should do nothing
-        eli.endpointChanged(new EndpointEvent(EndpointEvent.REMOVED, endpoint), null);
-        eli.endpointChanged(new EndpointEvent(EndpointEvent.REMOVED, endpoint), null); // should do nothing
-
-        c.verify();
-    }
-
-    private void expectCreated(ZooKeeper zk, String path) throws KeeperException, InterruptedException {
-        expect(zk.create(EasyMock.eq(path), 
-                         (byte[])EasyMock.anyObject(), 
-                         EasyMock.eq(Ids.OPEN_ACL_UNSAFE),
-                         EasyMock.eq(CreateMode.EPHEMERAL)))
-            .andReturn("");
-    }
-
-    private void expectDeleted(ZooKeeper zk, String path) throws InterruptedException, KeeperException {
-        zk.delete(EasyMock.eq(path), EasyMock.eq(-1));
-        EasyMock.expectLastCall().once();
-    }
-
-    private EndpointDescription createEndpoint() {
-        Map<String, Object> props = new HashMap<>();
-        props.put(Constants.OBJECTCLASS, new String[] {"myClass"});
-        props.put(RemoteConstants.ENDPOINT_ID, "http://google.de:80/test/sub");
-        props.put(RemoteConstants.SERVICE_IMPORTED_CONFIGS, "myConfig");
-        return new EndpointDescription(props);
-    }
-}
diff --git a/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/subscribe/InterfaceMonitorManagerTest.java b/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/subscribe/InterfaceMonitorManagerTest.java
deleted file mode 100644
index 79b2d8d..0000000
--- a/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/subscribe/InterfaceMonitorManagerTest.java
+++ /dev/null
@@ -1,85 +0,0 @@
-/**
- * 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.aries.rsa.discovery.zookeeper.subscribe;
-
-import static org.easymock.EasyMock.expect;
-import static org.junit.Assert.assertEquals;
-
-import java.util.ArrayList;
-import java.util.List;
-
-import org.apache.aries.rsa.discovery.zookeeper.ZooKeeperDiscovery;
-import org.apache.aries.rsa.discovery.zookeeper.repository.ZookeeperEndpointRepository;
-import org.easymock.EasyMock;
-import org.easymock.IMocksControl;
-import org.junit.Test;
-import org.osgi.framework.ServiceReference;
-import org.osgi.service.remoteserviceadmin.EndpointDescription;
-import org.osgi.service.remoteserviceadmin.EndpointEventListener;
-
-public class InterfaceMonitorManagerTest {
-
-    @Test
-    public void testEndpointListenerTrackerCustomizer() {
-        IMocksControl c = EasyMock.createControl();
-        ServiceReference<EndpointEventListener> sref = createService(c, "(objectClass=mine)");
-        ServiceReference<EndpointEventListener> sref2 = createService(c, "(objectClass=mine)");
-        ZookeeperEndpointRepository repository = c.createMock(ZookeeperEndpointRepository.class);
-        List<EndpointDescription> endpoints = new ArrayList<>();
-        expect(repository.getAll()).andReturn(endpoints).atLeastOnce();
-        EndpointEventListener epListener1 = c.createMock(EndpointEventListener.class); 
-        EndpointEventListener epListener2 = c.createMock(EndpointEventListener.class); 
-
-        c.replay();
-
-        InterestManager eltc = new InterestManager(repository);
-        // sref has no scope -> nothing should happen
-        assertEquals(0, eltc.getInterests().size());
-
-
-        eltc.addInterest(sref, epListener1);
-        assertEquals(1, eltc.getInterests().size());
-
-        eltc.addInterest(sref, epListener1);
-        assertEquals(1, eltc.getInterests().size());
-
-        eltc.addInterest(sref2, epListener2);
-        assertEquals(2, eltc.getInterests().size());
-
-        eltc.removeInterest(sref);
-        assertEquals(1, eltc.getInterests().size());
-
-        eltc.removeInterest(sref);
-        assertEquals(1, eltc.getInterests().size());
-
-        eltc.removeInterest(sref2);
-        assertEquals(0, eltc.getInterests().size());
-
-        c.verify();
-    }
-
-    @SuppressWarnings("unchecked")
-    private ServiceReference<EndpointEventListener> createService(IMocksControl c, String scope) {
-        ServiceReference<EndpointEventListener> sref = c.createMock(ServiceReference.class);
-        expect(sref.getProperty(EndpointEventListener.ENDPOINT_LISTENER_SCOPE)).andReturn(scope).atLeastOnce();
-        expect(sref.getProperty(ZooKeeperDiscovery.DISCOVERY_ZOOKEEPER_ID)).andReturn(null).atLeastOnce();
-        return sref;
-    }
-
-}
diff --git a/discovery/zookeeper/src/test/resources/logback.xml b/discovery/zookeeper/src/test/resources/logback.xml
new file mode 100644
index 0000000..7f2cb48
--- /dev/null
+++ b/discovery/zookeeper/src/test/resources/logback.xml
@@ -0,0 +1,32 @@
+<?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.
+-->
+<configuration>
+  <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
+    <encoder>
+      <pattern>%date %level %logger:%line %msg%n</pattern>
+    </encoder>
+  </appender>
+
+  <logger name="org.apache.zookeeper" level="WARN" /> 
+
+  <root level="info">
+    <appender-ref ref="console"/>
+  </root>
+</configuration>
\ No newline at end of file
diff --git a/itests/felix/src/test/java/org/apache/aries/rsa/itests/felix/tcp/TestDiscoveryExport.java b/itests/felix/src/test/java/org/apache/aries/rsa/itests/felix/tcp/TestDiscoveryExport.java
index 0b6aadc..a24b140 100644
--- a/itests/felix/src/test/java/org/apache/aries/rsa/itests/felix/tcp/TestDiscoveryExport.java
+++ b/itests/felix/src/test/java/org/apache/aries/rsa/itests/felix/tcp/TestDiscoveryExport.java
@@ -18,20 +18,20 @@
  */
 package org.apache.aries.rsa.itests.felix.tcp;
 
+import static org.awaitility.Awaitility.await;
+
 import java.io.ByteArrayInputStream;
-import java.util.List;
 
 import javax.inject.Inject;
 
-import org.apache.aries.rsa.discovery.endpoint.EndpointDescriptionParser;
+import org.apache.aries.rsa.discovery.zookeeper.ZookeeperEndpointPublisher;
 import org.apache.aries.rsa.examples.echotcp.api.EchoService;
 import org.apache.aries.rsa.itests.felix.RsaTestBase;
 import org.apache.aries.rsa.spi.DistributionProvider;
+import org.apache.aries.rsa.spi.EndpointDescriptionParser;
 import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.WatchedEvent;
-import org.apache.zookeeper.Watcher;
 import org.apache.zookeeper.ZooKeeper;
-import org.apache.zookeeper.data.Stat;
+import org.hamcrest.Matchers;
 import org.junit.Assert;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -42,10 +42,14 @@
 
 @RunWith(PaxExam.class)
 public class TestDiscoveryExport extends RsaTestBase {
-    private static final String GREETER_ZOOKEEPER_NODE = "/osgi/service_registry";
-
     @Inject
     DistributionProvider tcpProvider;
+    
+    @Inject
+    EndpointDescriptionParser parser;
+    
+    @Inject
+    ZooKeeper zookeeper;
 
     @Configuration
     public static Option[] configure() throws Exception {
@@ -70,44 +74,21 @@
     }
 
     private EndpointDescription getEndpoint() throws Exception {
-        ZooKeeper zk = new ZooKeeper("localhost:" + ZK_PORT, 1000, new DummyWatcher());
-        assertNodeExists(zk, GREETER_ZOOKEEPER_NODE, 10000);
-        String endpointPath = getEndpointPath(zk, GREETER_ZOOKEEPER_NODE);
-        EndpointDescription epd = getEndpointDescription(zk, endpointPath);
-        zk.close();
-        return epd;
+        String endpointName = await("Node exists").until(this::getEndpointPath, Matchers.notNullValue());
+        return getEndpointDescription(zookeeper, ZookeeperEndpointPublisher.PATH_PREFIX + "/" + endpointName);
     }
 
     private EndpointDescription getEndpointDescription(ZooKeeper zk, String endpointPath)
         throws KeeperException, InterruptedException {
         byte[] data = zk.getData(endpointPath, false, null);
         ByteArrayInputStream is = new ByteArrayInputStream(data);
-        return new EndpointDescriptionParser().readEndpoint(is);
+        return parser.readEndpoint(is);
     }
 
-    private String getEndpointPath(ZooKeeper zk, String servicePath) throws KeeperException, InterruptedException {
-        List<String> children = zk.getChildren(servicePath, false);
-        return servicePath + "/" + children.iterator().next();
-    }
-
-    private void assertNodeExists(ZooKeeper zk, String zNode, int timeout) {
-        long endTime = System.currentTimeMillis() + timeout;
-        Stat stat = null;
-        while (stat == null && System.currentTimeMillis() < endTime) {
-            try {
-                stat = zk.exists(zNode, null);
-                Thread.sleep(200);
-            } catch (Exception e) {
-                // Ignore
-            }
-        }
-        Assert.assertNotNull("ZooKeeper node " + zNode + " was not found", stat);
-    }
-    
-    private final class DummyWatcher implements Watcher {
-        @Override
-        public void process(WatchedEvent event) {
-        }
+    private String getEndpointPath() throws KeeperException, InterruptedException {
+        return zookeeper.getChildren(ZookeeperEndpointPublisher.PATH_PREFIX, false).stream()
+                .findFirst()
+                .orElse(null);
     }
 
 }
diff --git a/itests/felix/src/test/java/org/apache/aries/rsa/itests/felix/tcp/TestDiscoveryImport.java b/itests/felix/src/test/java/org/apache/aries/rsa/itests/felix/tcp/TestDiscoveryImport.java
index e735f55..4aea83f 100644
--- a/itests/felix/src/test/java/org/apache/aries/rsa/itests/felix/tcp/TestDiscoveryImport.java
+++ b/itests/felix/src/test/java/org/apache/aries/rsa/itests/felix/tcp/TestDiscoveryImport.java
@@ -31,9 +31,8 @@
 
 import javax.inject.Inject;
 
-import org.apache.aries.rsa.discovery.zookeeper.repository.ZookeeperEndpointRepository;
+import org.apache.aries.rsa.discovery.zookeeper.ZookeeperEndpointPublisher;
 import org.apache.aries.rsa.itests.felix.RsaTestBase;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.ops4j.pax.exam.Configuration;
@@ -49,10 +48,14 @@
 @RunWith(PaxExam.class)
 public class TestDiscoveryImport extends RsaTestBase {
     @Inject
-    ZookeeperEndpointRepository repository;
+    ZookeeperEndpointPublisher publisher;
     
     @Inject
     BundleContext context;
+
+    private Semaphore sem = new Semaphore(0);;
+
+    private List<EndpointEvent> events = new ArrayList<>();
     
     @Configuration
     public static Option[] configure() throws Exception {
@@ -66,30 +69,32 @@
         };
     }
 
-    @Ignore
     @Test
     public void testDiscoveryImport() throws Exception {
-        final Semaphore sem = new Semaphore(0);
-        final List<EndpointEvent> events = new ArrayList<>();
-        EndpointEventListener listener = new EndpointEventListener() {
-            
-            @Override
-            public void endpointChanged(EndpointEvent event, String filter) {
-                events.add(event);
-                sem.release();
-            }
-        };
+        context.registerService(EndpointEventListener.class, this::endpointChanged, listenerProps());
+        EndpointDescription endpoint = createEndpoint();
+        publisher.endpointChanged(new EndpointEvent(EndpointEvent.ADDED, endpoint));
+        assertTrue(sem.tryAcquire(10, TimeUnit.SECONDS));
+        //assertThat(events.get(0), samePropertyValuesAs(new EndpointEvent(EndpointEvent.ADDED, endpoint)));
+    }
+
+    private Dictionary<String, Object> listenerProps() {
         Dictionary<String, Object> eprops = new Hashtable<>();
         eprops.put(EndpointEventListener.ENDPOINT_LISTENER_SCOPE, "(objectClass=*)");
-        context.registerService(EndpointEventListener.class, listener, eprops);
+        return eprops;
+    }
+
+    private EndpointDescription createEndpoint() {
         Map<String, Object> props = new HashMap<>();
         props.put(Constants.OBJECTCLASS, new String[]{"my"});
         props.put(RemoteConstants.ENDPOINT_ID, "myid");
         props.put(RemoteConstants.SERVICE_IMPORTED_CONFIGS, "myconfig");
         EndpointDescription endpoint = new EndpointDescription(props);
-        repository.add(endpoint);
-        assertTrue(sem.tryAcquire(10, TimeUnit.SECONDS));
-        //assertThat(events.get(0), samePropertyValuesAs(new EndpointEvent(EndpointEvent.ADDED, endpoint)));
+        return endpoint;
     }
 
+    public void endpointChanged(EndpointEvent event, String filter) {
+        events.add(event);
+        sem.release();
+    }
 }
diff --git a/itests/felix/src/test/java/org/apache/aries/rsa/itests/felix/tcp/TestFindHook.java b/itests/felix/src/test/java/org/apache/aries/rsa/itests/felix/tcp/TestFindHook.java
index 78598a9..09aa6fd 100644
--- a/itests/felix/src/test/java/org/apache/aries/rsa/itests/felix/tcp/TestFindHook.java
+++ b/itests/felix/src/test/java/org/apache/aries/rsa/itests/felix/tcp/TestFindHook.java
@@ -19,10 +19,10 @@
  */
 
 
+import static org.awaitility.Awaitility.await;
+
 import java.io.IOException;
 import java.util.Collection;
-import java.util.concurrent.Callable;
-import java.util.concurrent.TimeoutException;
 
 import javax.inject.Inject;
 
@@ -30,12 +30,13 @@
 import org.apache.aries.rsa.itests.felix.RsaTestBase;
 import org.apache.aries.rsa.itests.felix.ServerConfiguration;
 import org.apache.aries.rsa.itests.felix.TwoContainerPaxExam;
-import org.junit.Assert;
+import org.hamcrest.Matchers;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.ops4j.pax.exam.Configuration;
 import org.ops4j.pax.exam.Option;
 import org.osgi.framework.BundleContext;
+import org.osgi.framework.InvalidSyntaxException;
 import org.osgi.framework.ServiceReference;
 
 @RunWith(TwoContainerPaxExam.class)
@@ -66,48 +67,14 @@
                 configZKDiscovery()
         };
     }
-    
-    public <T> T tryTo(String message, Callable<T> func) throws TimeoutException {
-        return tryTo(message, func, 5000);
-    }
-    
-    public <T> T tryTo(String message, Callable<T> func, long timeout) throws TimeoutException {
-        Throwable lastException = null;
-        long startTime = System.currentTimeMillis();
-        while (System.currentTimeMillis() - startTime < timeout) {
-            try {
-                T result = func.call();
-                if (result != null) {
-                    return result;
-                }
-                lastException = null;
-            } catch (Throwable e) {
-                lastException = e;
-            }
-            try {
-                Thread.sleep(1000);
-            } catch (InterruptedException e) {
-                continue;
-            }
-        }
-        TimeoutException ex = new TimeoutException("Timeout while trying to " + message);
-        if (lastException != null) {
-            ex.addSuppressed(lastException);
-        }
-        throw ex;
-    }
 
     @Test
     public void testFind() throws Exception {
-        ServiceReference<EchoService> ref = tryTo("get EchoService", new Callable<ServiceReference<EchoService>>() {
+        await().until(() -> getEchoServices().size(), Matchers.equalTo(1));
+    }
 
-            @Override
-            public ServiceReference<EchoService> call() throws Exception {
-                Collection<ServiceReference<EchoService>> refs = context.getServiceReferences(EchoService.class, null);
-                return (refs.size() > 0)? refs.iterator().next() : null;
-            }
-        }, 10000);
-        Assert.assertNotNull(ref);
+    private Collection<ServiceReference<EchoService>> getEchoServices() throws InvalidSyntaxException {
+        return context.getServiceReferences(EchoService.class, null);
     }
 
 }
diff --git a/itests/felix/src/test/java/org/apache/aries/rsa/itests/felix/tcp/TestRoundTrip.java b/itests/felix/src/test/java/org/apache/aries/rsa/itests/felix/tcp/TestRoundTrip.java
index 2a038b8..7641219 100644
--- a/itests/felix/src/test/java/org/apache/aries/rsa/itests/felix/tcp/TestRoundTrip.java
+++ b/itests/felix/src/test/java/org/apache/aries/rsa/itests/felix/tcp/TestRoundTrip.java
@@ -58,7 +58,7 @@
                 rsaCore(),
                 rsaDiscoveryZookeeper(),
                 rsaProviderTcp(),
-                RsaTestBase.echoTcpConsumer(),
+                echoTcpConsumer(),
                 configZKDiscovery()
         };
     }
diff --git a/itests/felix/src/test/resources/logback.xml b/itests/felix/src/test/resources/logback.xml
index f4696e4..3b7921f 100644
--- a/itests/felix/src/test/resources/logback.xml
+++ b/itests/felix/src/test/resources/logback.xml
@@ -20,10 +20,12 @@
 <configuration>
   <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
     <encoder>
-      <pattern>%date %level %logger [%file : %line] %msg - %mdc %n</pattern>
+      <pattern>%date{HH:mm:ss} %level %logger{40}:%line - %msg%n</pattern>
     </encoder>
   </appender>
 
+  <logger name="org.apache.zookeeper" level="WARN" /> 
+
   <root level="info">
     <appender-ref ref="console"/>
   </root>
diff --git a/parent/pom.xml b/parent/pom.xml
index ab071f9..f65aac2 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -58,15 +58,20 @@
             <scope>provided</scope>
         </dependency>
         <dependency>
+            <groupId>org.osgi</groupId>
+            <artifactId>osgi.annotation</artifactId>
+            <version>7.0.0</version>
+        </dependency>
+        <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>
             <scope>provided</scope>
         </dependency>
 
         <dependency>
-            <groupId>org.slf4j</groupId>
-            <artifactId>slf4j-jdk14</artifactId>
-            <scope>test</scope>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+            <version>1.2.3</version>
         </dependency>
         <dependency>
             <groupId>junit</groupId>
@@ -234,7 +239,7 @@
                 <plugin>
                     <groupId>org.apache.felix</groupId>
                     <artifactId>maven-bundle-plugin</artifactId>
-                    <version>3.5.1</version>
+                    <version>4.2.0</version>
                     <extensions>true</extensions>
                 </plugin>
                 <plugin>
diff --git a/discovery/local/src/test/java/org/apache/aries/rsa/discovery/local/ActivatorTest.java b/spi/src/main/java/org/apache/aries/rsa/spi/EndpointDescriptionParser.java
similarity index 65%
copy from discovery/local/src/test/java/org/apache/aries/rsa/discovery/local/ActivatorTest.java
copy to spi/src/main/java/org/apache/aries/rsa/spi/EndpointDescriptionParser.java
index 478e41b..caa5d01 100644
--- a/discovery/local/src/test/java/org/apache/aries/rsa/discovery/local/ActivatorTest.java
+++ b/spi/src/main/java/org/apache/aries/rsa/spi/EndpointDescriptionParser.java
@@ -16,12 +16,19 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.aries.rsa.discovery.local;
+package org.apache.aries.rsa.spi;
 
-import junit.framework.TestCase;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.List;
 
-public class ActivatorTest extends TestCase {
+import org.osgi.service.remoteserviceadmin.EndpointDescription;
 
-    public void testActivator() throws Exception {
-    }
+public interface EndpointDescriptionParser {
+    
+    List<EndpointDescription> readEndpoints(InputStream is);
+    
+    EndpointDescription readEndpoint(InputStream is);
+    
+    void writeEndpoint(EndpointDescription epd, OutputStream os);
 }
diff --git a/spi/src/main/java/org/apache/aries/rsa/spi/packageinfo b/spi/src/main/java/org/apache/aries/rsa/spi/packageinfo
index c72722a..1da3be9 100644
--- a/spi/src/main/java/org/apache/aries/rsa/spi/packageinfo
+++ b/spi/src/main/java/org/apache/aries/rsa/spi/packageinfo
@@ -16,4 +16,4 @@
 # specific language governing permissions and limitations
 # under the License.
 #
-version 1.0.0
+version 1.1.0