package ethereumjava.net.provider;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.TimeUnit;

import ethereumjava.exception.EthereumJavaException;
import ethereumjava.net.Request;
import ethereumjava.net.Response;
import rx.Observable;
import rx.Subscriber;
import rx.functions.Action1;
import rx.functions.Func1;

/**
 * Created by gunicolas on 31/08/16.
 */
public abstract class IpcAbstractProvider extends AbstractProvider {

    public final String ipcFilePath;

    Thread listeningThread;
    boolean listen;

    public IpcAbstractProvider(String _ipcFilePath) throws EthereumJavaException {
        super();
        this.ipcFilePath = _ipcFilePath;
    }

    @Override
    public void init() throws EthereumJavaException {
        startListening();
    }

    @Override
    protected Observable send(byte[] stringRequest, final Request request){
        try {
            out.write(stringRequest);
        } catch (IOException e) {
            throw new EthereumJavaException(e);
        }

        requestQueue.put(request.getId(),request);

        Observable ret = Observable.create(new Observable.OnSubscribe<Object>() {
            @Override
            public void call(Subscriber<? super Object> subscriber) {
                request.addSubscriber(subscriber);
            }
        });

        return ret;
    }

    private void startListening() throws EthereumJavaException {

        Observable.interval(500, TimeUnit.MILLISECONDS) //Thread computation
                .map(readFile())
                .filter(removeNullLines())
                .subscribe(manageResponse());

    }

    private Subscriber<String> manageResponse() {
        return new Subscriber<String>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {
                //TODO on error on ipc file
            }

            @Override
            public void onNext(String line) {
                Response response = gson.fromJson(line, Response.class);

                if (response.request != null) {

                    List<Subscriber> subscribers = response.request.getSubscribers();
                    if (response.isError()) {
                        for (Subscriber subscriber : subscribers) {
                            subscriber.onError(new EthereumJavaException(response.error.message));
                        }
                    } else {
                        for (Subscriber subscriber : subscribers) {
                            subscriber.onNext(response.result);
                            subscriber.onCompleted();
                        }
                    }
                    requestQueue.remove(response.id);
                }
            }
        };
    }
    private Func1<String, Boolean> removeNullLines() {
        return new Func1<String, Boolean>() {
            @Override
            public Boolean call(String s) {
                return s == null;
            }
        };
    }
    private Func1<Long,String> readFile(){
        return new Func1<Long, String>() {
            @Override
            public String call(Long aLong) {
                try {
                    if( ! in.ready() ){
                        return null;
                    } else {
                        String line = in.readLine();
                        return line;
                    }
                } catch (IOException e) {
                    throw new EthereumJavaException(e);
                }
            }
        };
    }

    @Override
    public void stop() throws EthereumJavaException {
        if( this.listeningThread != null ){
            this.listen = false;
        }
        try {
            this.inputStream.close();
            this.outputStream.close();
        }catch(IOException e){
            throw new EthereumJavaException(e);
        }
    }
}

@@ -1622,17 +1622,16 @@
 nse());%0A
-%0A
     %7D%0A%0A 
@@ -1641,26 +1641,23 @@
 private 
-Subscriber
+Action1
 %3CString%3E
@@ -1695,26 +1695,23 @@
 urn new 
-Subscriber
+Action1
 %3CString%3E
@@ -1707,33 +1707,32 @@
 on1%3CString%3E() %7B%0A
- 
            @Over
@@ -1739,33 +1739,32 @@
 ride%0A           
- 
 public void onCo
@@ -1763,220 +1763,12 @@
 oid 
-onCompleted() %7B%0A%0A            %7D%0A%0A            @Override%0A            public void onError(Throwable e) %7B%0A                //TODO on error on ipc file%0A            %7D%0A%0A            @Override%0A            public void onNext
+call
 (Str
@@ -1775,25 +1775,24 @@
 ing line) %7B%0A
- 
             
@@ -1848,25 +1848,24 @@
 se.class);%0A%0A
- 
             
@@ -1919,17 +1919,16 @@
         
- 
 List%3CSub
@@ -1981,25 +1981,24 @@
 scribers();%0A
- 
             
@@ -2045,33 +2045,32 @@
                 
- 
 for (Subscriber 
@@ -2089,33 +2089,32 @@
  subscribers) %7B%0A
- 
                 
@@ -2210,36 +2210,34 @@
                 
- 
 %7D%0A
- 
                 
@@ -2240,33 +2240,32 @@
        %7D else %7B%0A
- 
                 
@@ -2307,33 +2307,32 @@
  subscribers) %7B%0A
- 
                 
@@ -2397,33 +2397,32 @@
                 
- 
 subscriber.onCom
@@ -2446,36 +2446,34 @@
                 
- 
 %7D%0A
- 
                 
@@ -2469,33 +2469,32 @@
               %7D%0A
- 
                 
@@ -2537,36 +2537,34 @@
 %0A               
- 
 %7D%0A
- 
            %7D%0A   
@@ -2552,33 +2552,32 @@
  %7D%0A           %7D%0A
- 
        %7D;%0A    %7D%0A

Replace Subscriber by Action because only onNext Action was implemented