package com.github.lukaspili.reactivebilling.observable;

import android.content.ComponentName;
import android.content.Context;
import android.content.Intent;
import android.content.ServiceConnection;
import android.os.IBinder;

import com.android.vending.billing.IInAppBillingService;
import com.github.lukaspili.reactivebilling.BillingService;
import com.github.lukaspili.reactivebilling.ReactiveBillingLogger;

import java.util.concurrent.Semaphore;

import rx.Observable;
import rx.Observer;
import rx.Subscriber;
import rx.functions.Action0;
import rx.subscriptions.Subscriptions;


public abstract class BaseObservable<T> implements Observable.OnSubscribe<T> {

    protected final Context context;
    private final Semaphore semaphore = new Semaphore(0);

    private BillingService billingService;

    BaseObservable(Context context) {
        this.context = context;
    }

    @Override
    public void call(Subscriber<? super T> subscriber) {
        final Intent intent = new Intent("com.android.vending.billing.InAppBillingService.BIND");
        intent.setPackage("com.android.vending");

        final Connection connection = new Connection();

        ReactiveBillingLogger.log("Bind service (thread %s)", Thread.currentThread().getName());
        try {
            context.bindService(intent, connection, Context.BIND_AUTO_CREATE);
        } catch (SecurityException e) {
            ReactiveBillingLogger.log(e, "Bind service error");
            subscriber.onError(e);
        }

        subscriber.add(Subscriptions.create(new Action0() {
            @Override
            public void call() {
                ReactiveBillingLogger.log("Unbind service (thread %s)", Thread.currentThread().getName());
                context.unbindService(connection);
            }
        }));

        // freeze the current RX thread until service is connected
        // because bindService() will call the connection callback on the main thread
        // we want to get back on the current RX thread
        ReactiveBillingLogger.log("Acquire semaphore until service is ready");
        semaphore.acquireUninterruptibly();

        // once the semaphore is released
        // it means that the service is connected and available
        //TODO: what happens if the service is never connected?
        ReactiveBillingLogger.log("Billing service ready (thread %s)", Thread.currentThread().getName());
        onBillingServiceReady(billingService, subscriber);
    }

    protected abstract void onBillingServiceReady(BillingService billingService, Observer<? super T> observer);

    private class Connection implements ServiceConnection {

        /**
         * For some reason, that method is always called on the main thread
         * Regardless of the originating thread executing bindService()
         */
        @Override
        public void onServiceConnected(ComponentName name, IBinder service) {
            ReactiveBillingLogger.log("Service connected (thread %s)", Thread.currentThread().getName());

            IInAppBillingService inAppBillingService = IInAppBillingService.Stub.asInterface(service);
            billingService = new BillingService(context, inAppBillingService);

            // once the service is available, release the semaphore
            // that is blocking the originating thread
            semaphore.release();
        }

        @Override
        public void onServiceDisconnected(ComponentName name) {
            ReactiveBillingLogger.log("Service disconnected (thread %s)", Thread.currentThread().getName());
            billingService = null;
        }
    }
}

@@ -220,16 +220,42 @@
 IBinder;
+%0Aimport android.os.Looper;
 %0A%0Aimport
@@ -1120,24 +1120,191 @@
 vending%22);%0A%0A
+        // don't bother with semaphores in the originating thread is the main thread%0A        final boolean useSemaphore = Looper.myLooper() != Looper.getMainLooper();%0A
         fina
@@ -1344,16 +1344,40 @@
 nection(
+subscriber, useSemaphore
 );%0A%0A    
@@ -2009,24 +2009,56 @@
       %7D));%0A%0A
+        if (useSemaphore) %7B%0A    
         // f
@@ -2112,16 +2112,20 @@
 nnected%0A
+    
         
@@ -2210,16 +2210,20 @@
         
+ 
+   
 // we wa
@@ -2254,32 +2254,36 @@
 rrent RX thread%0A
+    
         Reactive
@@ -2341,20 +2341,70 @@
 is ready
-%22);%0A
+ (thread %25s)%22, Thread.currentThread().getName());%0A    
         
@@ -2440,32 +2440,36 @@
 bly();%0A%0A        
+    
 // once the sema
@@ -2494,16 +2494,20 @@
         
+   
+ 
 // it me
@@ -2562,16 +2562,20 @@
         
+    
 //TODO: 
@@ -2622,16 +2622,140 @@
 nected?%0A
+            deliverBillingService(subscriber);%0A        %7D%0A    %7D%0A%0A    private void deliverBillingService(Observer observer) %7B%0A
         
@@ -2898,24 +2898,22 @@
 ervice, 
-subscrib
+observ
 er);%0A   
@@ -3086,24 +3086,274 @@
 nnection %7B%0A%0A
+        private final Observer observer;%0A        private final boolean useSemaphore;%0A%0A        public Connection(Observer observer, boolean useSemaphore) %7B%0A            this.observer = observer;%0A            this.useSemaphore = useSemaphore;%0A        %7D%0A%0A
         /**%0A
@@ -3894,24 +3894,60 @@
 gService);%0A%0A
+            if (useSemaphore) %7B%0A    
             
@@ -4014,16 +4014,20 @@
         
+  
+  
 // that 
@@ -4077,28 +4077,226 @@
     
-semaphore.release();
+    ReactiveBillingLogger.log(%22Release semaphore (thread %25s)%22, Thread.currentThread().getName());%0A                semaphore.release();%0A            %7D else %7B%0A                deliverBillingService(observer);%0A            %7D
 %0A   

Fix deadlock when observable was called on main thread.