package com.github.reiseburo.rx.curator;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import rx.Observable;
import rx.Subscriber;
import rx.functions.Action0;

import java.io.IOException;

/**
 * PathChildren is an {@code Observable} which takes events from a {@code PathChildrenCache}
 * and emits them for subscription
 */
public class PathChildren {
    protected CuratorFramework curatorFramework;
    protected PathChildrenCache cache;

    protected PathChildren() {
    }

    public CuratorFramework getCuratorFramework() {
        return curatorFramework;
    }

    public void setCuratorFramework(CuratorFramework curatorFramework) {
        this.curatorFramework = curatorFramework;
    }


    /**
     * Create an instance of PathChildren configured with the provided
     * {@code CuratorFramework} instance. This assumes that {@code curatorFramework}
     * has already been started
     *
     * @param curatorFramework
     * @return
     */
    public static PathChildren with(CuratorFramework curatorFramework) {
        PathChildren instance = new PathChildren();
        instance.setCuratorFramework(curatorFramework);
        return instance;
    }


    public Observable<PathChildrenCacheEvent> watch(final String znodePath) {
        return Observable.create(new Observable.OnSubscribe<PathChildrenCacheEvent>() {
            @Override
            public void call(final Subscriber<? super PathChildrenCacheEvent> subscriber) {
                cache = new PathChildrenCache(curatorFramework, znodePath, true);
                try {
                    cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
                } catch (Exception ex) {
                    subscriber.onError(ex);
                }

                cache.getListenable().addListener(new PathChildrenCacheListener() {
                    @Override
                    public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                        subscriber.onNext(event);
                    }
                });
            }
        }).doOnUnsubscribe(new Action0() {
            @Override
            public void call() {
                /* properly close out our cache object when our subscriber leaves */
                if (cache instanceof PathChildrenCache) {
                    try {
                        cache.close();
                    } catch (IOException exception) {
                        /* Swallow, we don't need this exception */
                    }
                }

            }
        });
    }
}

@@ -1762,218 +1762,8 @@
 ue);
-%0A                try %7B%0A                    cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);%0A                %7D catch (Exception ex) %7B%0A                    subscriber.onError(ex);%0A                %7D
 %0A%0A  
@@ -2078,24 +2078,235 @@
          %7D);
+%0A%0A                try %7B%0A                    cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);%0A                %7D catch (Exception ex) %7B%0A                    subscriber.onError(ex);%0A                %7D
 %0A           

Reorder adding the listener