package com.github.davidmoten.rx.operators;

import java.io.IOException;
import java.nio.file.ClosedWatchServiceException;
import java.nio.file.WatchEvent;
import java.nio.file.WatchKey;
import java.nio.file.WatchService;
import java.util.concurrent.atomic.AtomicBoolean;

import rx.Observable.Operator;
import rx.Observer;
import rx.Subscriber;
import rx.Subscription;
import rx.observers.Subscribers;

public class OperatorWatchServiceEvents implements Operator<WatchEvent<?>, WatchService> {

    @Override
    public Subscriber<? super WatchService> call(final Subscriber<? super WatchEvent<?>> subscriber) {
        Subscriber<WatchService> result = Subscribers.from(new Observer<WatchService>() {

            @Override
            public void onCompleted() {
                subscriber.onCompleted();
            }

            @Override
            public void onError(Throwable e) {
                subscriber.onError(e);
            }

            @Override
            public void onNext(WatchService watchService) {
                AtomicBoolean subscribed = new AtomicBoolean(true);
                if (!subscribed.get()) {
                    subscriber.onError(new RuntimeException(
                            "WatchService closed. You can only subscribe once to a WatchService."));
                    return;
                }
                subscriber.add(createSubscriptionToCloseWatchService(watchService, subscribed, subscriber));
                emitEvents(watchService, subscriber, subscribed);
            }
        });
        subscriber.add(result);
        return result;
    }

    private static void emitEvents(WatchService watchService, Subscriber<? super WatchEvent<?>> subscriber,
            AtomicBoolean subscribed) {
        // get the first event before looping
        WatchKey key = nextKey(watchService, subscriber, subscribed);

        while (key != null) {
            if (subscriber.isUnsubscribed())
                return;
            // we have a polled event, now we traverse it and
            // receive all the states from it
            for (WatchEvent<?> event : key.pollEvents()) {
                if (subscriber.isUnsubscribed())
                    return;
                else
                    subscriber.onNext(event);
            }

            boolean valid = key.reset();
            if (!valid && subscribed.get()) {
                subscriber.onCompleted();
                return;
            } else if (!valid)
                return;

            key = nextKey(watchService, subscriber, subscribed);
        }
    }

    private static WatchKey nextKey(WatchService watchService, Subscriber<? super WatchEvent<?>> subscriber,
            AtomicBoolean subscribed) {
        try {
            // this command blocks but unsubscribe close the watch
            // service and interrupt it
            return watchService.take();
        } catch (ClosedWatchServiceException e) {
            // must have unsubscribed
            if (subscribed.get())
                subscriber.onCompleted();
            return null;
        } catch (InterruptedException e) {
            // this case is problematic because unsubscribe may call
            // Thread.interrupt() before calling the unsubscribe method of
            // the Subscription. Thus at this point we don't know if a
            // deliberate interrupt was called in which case I would call
            // onComplete or if unsubscribe was called in which case I
            // should not call anything. For the moment I choose to not call
            // anything partly because a deliberate stop of the
            // watchService.take ignorant of the Observable should ideally
            // happen via a call to the WatchService.close() method rather
            // than Thread.interrupt().
            // TODO raise the issue with RxJava team in particular
            // Subscriptions.from(Future) calling FutureTask.cancel(true)
            try {
                watchService.close();
            } catch (IOException e1) {
                // do nothing
            }
            return null;
        }
    }

    private final static Subscription createSubscriptionToCloseWatchService(final WatchService watchService,
            final AtomicBoolean subscribed, final Subscriber<? super WatchEvent<?>> subscriber) {
        return new Subscription() {

            @Override
            public void unsubscribe() {
                try {
                    watchService.close();
                } catch (ClosedWatchServiceException e) {
                    // do nothing
                } catch (IOException e) {
                    // do nothing
                } finally {
                    subscribed.set(false);
                }
            }

            @Override
            public boolean isUnsubscribed() {
                return !subscribed.get();
            }
        };
    }

}

@@ -397,16 +397,90 @@
 ibers;%0A%0A
+/**%0A * Emits %7B@link WatchEvent%7Ds for each input %7B@link WatchService%7D.%0A */%0A
 public c

add javadoc