package com.github.bckfnn.reactstreams.vertx;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.vertx.java.core.AsyncResult;
import org.vertx.java.core.Handler;
import org.vertx.java.core.buffer.Buffer;
import org.vertx.java.core.file.AsyncFile;

import com.github.bckfnn.reactstreams.BaseSubscription;
import com.github.bckfnn.reactstreams.Builder;
import com.github.bckfnn.reactstreams.Operations;

public class RsAsyncFile {
    private AsyncFile asyncFile;

    public RsAsyncFile(AsyncFile asyncFile) {
        this.asyncFile = asyncFile;
    }

    public Operations<Buffer> read() {
        return Builder.as(new Publisher<Buffer>() {
            @Override
            public void subscribe(Subscriber<Buffer> subscriber) {
                subscriber.onSubscribe(new BaseSubscription<Buffer>(subscriber) {
                    boolean started = false;
                    @Override
                    public void request(int elements) {
                        super.request(elements);
                        if (getPending() <= 0) {
                            asyncFile.pause();
                        } else {
                            asyncFile.resume();
                        }

                        if (!started) {
                            started = true;

                            asyncFile.dataHandler(new Handler<Buffer>() {
                                @Override
                                public void handle(Buffer event) {
                                    sendNext(event);
                                }
                            });
                            asyncFile.endHandler(new Handler<Void>() {
                                @Override
                                public void handle(Void event) {
                                    sendComplete();
                                }
                            });
                            asyncFile.exceptionHandler(new Handler<Throwable>() {
                                @Override
                                public void handle(Throwable event) {
                                    sendError(event);
                                }
                            });
                        }
                    }
                });
            }
        });
    }

    public Operations<Void> close() {
        return Builder.as(new Publisher<Void>() {
            @Override
            public void subscribe(Subscriber<Void> subscriber) {
                subscriber.onSubscribe(new BaseSubscription<Void>(subscriber) {
                    boolean started = false;
                    @Override
                    public void request(int elements) {
                        if (started) {
                            return;
                        }
                        started = true;
                        asyncFile.close(new Handler<AsyncResult<Void>>() {
                            @Override
                            public void handle(AsyncResult<Void> event) {
                                if (event.succeeded()) {
                                    sendComplete();
                                } else {
                                    sendError(event.cause());
                                }
                            }
                        });
                    }
                });
            }
        });
    }
}

@@ -284,24 +284,78 @@
 yncFile;%0D%0A%0D%0A
+import com.github.bckfnn.reactstreams.BaseProcessor;%0D%0A
 import com.g
@@ -497,16 +497,66 @@
 ations;%0D
+%0Aimport com.github.bckfnn.reactstreams.Processor;%0D
 %0A%0D%0Apubli
@@ -1032,32 +1032,270 @@
                 
+%0D%0A                    @Override%0D%0A%09%09%09%09%09public void cancel() %7B%0D%0A%09%09%09%09%09%09super.cancel();%0D%0A%09%09%09%09%09%09asyncFile.pause();%0D%0A%09%09%09%09%09%09asyncFile.dataHandler(null);%0D%0A%09%09%09%09%09%09asyncFile.exceptionHandler(null);%0D%0A%09%09%09%09%09%09asyncFile.endHandler(null);%0D%0A%09%09%09%09%09%7D%0D%0A%0D%0A%09%09%09%09%09
 @Override%0D%0A     
@@ -2721,32 +2721,463 @@
      %7D);%0D%0A    %7D%0D
+%0A    %0D%0A    public Processor%3CBuffer, Void%3E write() %7B%0D%0A%09%09return new BaseProcessor%3CBuffer, Void%3E() %7B%0D%0A%09%09%09@Override%0D%0A%09%09%09public void doNext(Buffer value) %7B%0D%0A%09%09%09%09asyncFile.write(value);%0D%0A%09%09%09%09if (!asyncFile.writeQueueFull()) %7B%0D%0A%09%09%09%09%09sendRequest();%0D%0A%09%09%09%09%7D else %7B%0D%0A%09%09%09%09%09asyncFile.drainHandler(new Handler%3CVoid%3E() %7B%0D%0A%09%09%09%09%09%09@Override%0D%0A%09%09%09%09%09%09public void handle(Void event) %7B%0D%0A%09%09%09%09%09%09%09sendRequest();%0D%0A%09%09%09%09%09%09%7D%0D%0A%09%09%09%09%09%7D);%0D%0A%09%09%09%09%7D%0D%0A%09%09%09%7D%0D%0A%09%09%7D;%0D%0A    %7D%0D
 %0A%0D%0A    public Op

support writing.