/*
 *
 * 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.airavata.db.event.manager.utils;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.ZKPaths;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by goshenoy on 3/21/17.
 */
public class DbEventManagerZkUtils {

    public static void createDBEventMgrZkNode(CuratorFramework curatorClient, String publisherNode, String subscriberNode) throws Exception {
        // construct ZK paths for pub,sub
        String publisherZkPath = ZKPaths.makePath(Constants.DB_EVENT_MGR_ZK_PATH, publisherNode);
        String subscriberZkPath = ZKPaths.makePath(publisherZkPath, subscriberNode);

        // construct byte-data(s) for pub, sub
        byte[] publisherZkData = subscriberNode.getBytes();
        byte[] subscriberZkData = subscriberNode.getBytes();

        // create zkNode: "/db-event-mgr/pubnodename/subnodename"
        ZKPaths.mkdirs(curatorClient.getZookeeperClient().getZooKeeper(), subscriberZkPath);

        // set zkNode data for pub,sub
        curatorClient.setData().withVersion(-1 ).forPath(publisherZkPath, publisherZkData);
        curatorClient.setData().withVersion(-1 ).forPath(subscriberZkPath, subscriberZkData);
    }

    public static Map<String, List<String>> getSubscribersForPublisher(CuratorFramework curatorClient, String publisherNode) throws Exception {
        Map<String, List<String>> subscriberMap = new HashMap<>();

        // construct ZK path for pub
        String publisherZkPath = ZKPaths.makePath(Constants.DB_EVENT_MGR_ZK_PATH, publisherNode);

        // get children-list for pub
        List<String> subscriberList = curatorClient.getChildren().forPath(publisherZkPath);

        subscriberMap.put(publisherNode, subscriberList);
        return subscriberMap;
    }

//    public static void main(String[] args) {
//        String connectionString = "localhost:2181";
//        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);
//
//        CuratorFramework curatorClient = CuratorFrameworkFactory.newClient(connectionString, retryPolicy);
//        curatorClient.start();
//        try {
//            DbEventManagerZkUtils.createDBEventMgrZkNode(curatorClient, "userprofile", "sharing");
//            DbEventManagerZkUtils.createDBEventMgrZkNode(curatorClient, "userprofile", "appcatalog");
//            System.out.println(DbEventManagerZkUtils.getSubscribersForPublisher(curatorClient, "userprofile"));
//        } catch (Exception ex) {
//            ex.printStackTrace();
//        }
//    }
}

@@ -855,24 +855,154 @@
 ger.utils;%0A%0A
+import org.apache.airavata.common.exception.ApplicationSettingsException;%0Aimport org.apache.airavata.common.utils.ServerSettings;%0A
 import org.a
@@ -1028,16 +1028,16 @@
 Policy;%0A
-
 import o
@@ -1240,16 +1240,73 @@
 ZKPaths;
+%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;
 %0A%0Aimport
@@ -1452,16 +1452,1027 @@
 tils %7B%0A%0A
+    private static final Logger logger = LoggerFactory.getLogger(DbEventManagerZkUtils.class);%0A    private static CuratorFramework curatorClient;%0A%0A    /**%0A     *  Get curatorFramework instance%0A     * @return%0A     * @throws ApplicationSettingsException%0A     */%0A    public static CuratorFramework getCuratorClient() throws ApplicationSettingsException %7B%0A        if (curatorClient == null) %7B%0A            synchronized (DbEventManagerZkUtils.class) %7B%0A                if (curatorClient == null) %7B%0A                    String connectionSting = ServerSettings.getZookeeperConnection();%0A                    RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);%0A                    curatorClient = CuratorFrameworkFactory.newClient(connectionSting, retryPolicy);%0A                %7D%0A            %7D%0A        %7D%0A%0A        return curatorClient;%0A    %7D%0A%0A    /**%0A     * Create Zk node for db event manager%0A     * @param curatorClient%0A     * @param publisherNode%0A     * @param subscriberNode%0A     * @throws Exception%0A     */%0A
     publ
@@ -3066,16 +3066,96 @@
 dename%22%0A
+        logger.debug(%22Creating Zk node for db-event-mgr: %22 + subscriberZkPath);%0A
         
@@ -3468,24 +3468,187 @@
 ta);%0A    %7D%0A%0A
+    /**%0A     * Get list of subscribers for given publisher%0A     * @param curatorClient%0A     * @param publisherNode%0A     * @return%0A     * @throws Exception%0A     */%0A
     public s

create curatorclient factory