package rx.curator

import org.apache.curator.RetryPolicy
import org.apache.curator.framework.CuratorFramework
import org.apache.curator.framework.CuratorFrameworkFactory
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 org.apache.curator.retry.RetryNTimes

import rx.Observable
import rx.Observer
import rx.Subscriber
import rx.observables.BlockingObservable

class PathCacheObservable {
    static void main(String[] arguments) {
        println "Starting PatchCacheObservable experiment"

        RetryPolicy retryPolicy = new RetryNTimes(3, 1000)
        CuratorFramework curator = CuratorFrameworkFactory.newClient('localhost:2181', retryPolicy)
        curator.start()
        PathChildrenCache cache = new PathChildrenCache(curator, '/brokers', true)
        println "Starting dataset: ${cache.currentData}"

        Observable observable = Observable.create { Subscriber subscriber ->
            cache.listenable.addListener({ CuratorFramework cf, PathChildrenCacheEvent ev ->
                println "heard ${ev} from ${cf}"

                /* If we've got our terminate event, properly close the cache and
                 * tell our subscriber, if it exists, to complete
                 */
                if (ev.type == PathChildrenCacheEvent.Type.INITIALIZED) {
                    cache.close()
                    if (!subscriber.isUnsubscribed()) {
                        subscriber.onCompleted()
                    }
                }
                else {
                    /* emit each event that isn't our terminator */
                    subscriber.onNext(ev)
                }
            } as PathChildrenCacheListener)

            /* Starting with a post-initialized event so we know when to terminate */
            cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT)
        }

        boolean waitFor = true

        observable.subscribe([
                onError: { Throwable e -> println "error: ${e}"; e.printStackTrace() },
                onCompleted: { waitFor = false; println "complete"; println "Final data: ${cache.currentData}" },
                onNext: { println "received: ${it}" }
        ] as Observer<PathChildrenCacheEvent>)

        /* since this is evented, we need something to sit around and wait for us */
        synchronized (this) {
            while (waitFor) { wait(500) }
        }
    }
}


Add a simple PathChildrenCache experiment trying to tie Curator into RxJava
Add a simple PathChildrenCache experiment trying to tie Curator into RxJava

Running this against a running Kafka results in:

    9:14:42 AM: Executing external task 'runExperiment -t -PexperimentClass=rx.curator.PathCacheObservable'...
    :compileJava UP-TO-DATE
    :compileGroovy UP-TO-DATE
    :processResources UP-TO-DATE
    :classes UP-TO-DATE
    :compileExperimentsJava UP-TO-DATE
    :compileExperimentsGroovy
    :processExperimentsResources UP-TO-DATE
    :experimentsClasses
    :compileTestJava UP-TO-DATE
    :compileTestGroovy UP-TO-DATE
    :processTestResources UP-TO-DATE
    :testClasses UP-TO-DATE
    :test UP-TO-DATE
    :runExperiment
    Starting PatchCacheObservable experiment
    log4j:WARN No appenders could be found for logger (org.apache.curator.framework.imps.CuratorFrameworkImpl).
    log4j:WARN Please initialize the log4j system properly.
    log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
    Starting dataset: []
    heard PathChildrenCacheEvent{type=CHILD_ADDED, data=ChildData{path='/brokers/topics', stat=4294967331,4294967331,1412043135590,1412043135590,0,27,0,0,0,27,824641831943
    , data=null}} from org.apache.curator.framework.imps.CuratorFrameworkImpl@5eb30746
    received: PathChildrenCacheEvent{type=CHILD_ADDED, data=ChildData{path='/brokers/topics', stat=4294967331,4294967331,1412043135590,1412043135590,0,27,0,0,0,27,824641831943
    , data=null}}
    heard PathChildrenCacheEvent{type=CHILD_ADDED, data=ChildData{path='/brokers/ids', stat=4294967330,4294967330,1412043135586,1412043135586,0,135,0,0,0,3,824659713847
    , data=null}} from org.apache.curator.framework.imps.CuratorFrameworkImpl@5eb30746
    received: PathChildrenCacheEvent{type=CHILD_ADDED, data=ChildData{path='/brokers/ids', stat=4294967330,4294967330,1412043135586,1412043135586,0,135,0,0,0,3,824659713847
    , data=null}}
    heard PathChildrenCacheEvent{type=INITIALIZED, data=null} from org.apache.curator.framework.imps.CuratorFrameworkImpl@5eb30746
    complete
    Final data: [ChildData{path='/brokers/ids', stat=4294967330,4294967330,1412043135586,1412043135586,0,135,0,0,0,3,824659713847
    , data=null}, ChildData{path='/brokers/topics', stat=4294967331,4294967331,1412043135590,1412043135590,0,27,0,0,0,27,824641831943
    , data=null}]

    BUILD SUCCESSFUL

    Total time: 9.821 secs

This doesn't really result in much interesting other than the events being sent
along to the subscriber properly, which I suppose is neat
