package com.pushtorefresh.storio.test;

import android.support.annotation.NonNull;

import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicReference;

import rx.Subscription;

public abstract class AbstractEmissionChecker<T> {

    @NonNull
    private final Queue<T> expectedValues;

    @NonNull
    private final Queue<T> obtainedValues;

    @NonNull
    private final AtomicReference<Throwable> onNextObtainedThrowable = new AtomicReference<Throwable>(null);

    public AbstractEmissionChecker(@NonNull Queue<T> expectedValues) {
        this.expectedValues = new ConcurrentLinkedQueue<T>(expectedValues);
        this.obtainedValues = new ConcurrentLinkedQueue<T>();
    }

    /**
     * Returns timeout of assertion for {@link #awaitNextExpectedValue()}.
     *
     * @return timeout in millis.
     */
    protected long timeoutMillis() {
        return 60000; // 60 seconds
    }

    /**
     * Asserts that next expected value was received.
     */
    public void awaitNextExpectedValue() {
        final long startTime = System.currentTimeMillis(); // We can not use SystemClock here :( Not in class path
        final long timeoutMillis = timeoutMillis();
        final T expected = expectedValues.remove();

        boolean expectedValueWasReceived = false;

        while (!expectedValueWasReceived
                && onNextObtainedThrowable.get() == null
                && !(System.currentTimeMillis() - startTime > timeoutMillis)) {

            if (obtainedValues.size() > 0) {
                final T obtained = obtainedValues.remove();

                if (expected.equals(obtained)) {
                    expectedValueWasReceived = true;
                } else {
                    throw new AssertionError("Obtained item not equals to expected: obtained = "
                            + obtained + ", expected = " + expected);
                }
            }

            Thread.yield(); // let other threads work
        }

        if (onNextObtainedThrowable.get() != null) {
           throw new AssertionError("Throwable occurred while waiting: " + onNextObtainedThrowable.get());
        } else if (!expectedValueWasReceived) {
            throw new AssertionError("Expected value = " + expected + " was not received, " +
                    "timeout = " + timeoutMillis + "ms, expectedValues.size = " + expectedValues.size() + ", obtainedValues = " + obtainedValues);
        }
    }

    /**
     * Asserts that all expected values were received.
     */
    public void assertThatNoExpectedValuesLeft() {
        if (expectedValues.size() != 0) {
            throw new AssertionError("Not all expected values were received: queue = " + expectedValues);
        }
    }

    /**
     * Checks that new received value is expected.
     *
     * @param obtained new value.
     */
    protected void onNextObtained(@NonNull T obtained) {
        try {
            obtainedValues.add(obtained);
        } catch (Throwable throwable) {
            // Catch everything, it's not a bug, it's a feature
            // Really, we don't want to break contract of Emission Checker if something goes wrong
            // Because problem can be handled via rx.Observable's Subscriber
            // And if so -> it'll break behavior of Emission Checker
            onNextObtainedThrowable.set(throwable);
        }
    }

    @NonNull
    public abstract Subscription subscribe();
}

@@ -94,21 +94,26 @@
 va.util.
-Queue
+LinkedList
 ;%0Aimport
@@ -127,35 +127,8 @@
 til.
-concurrent.ConcurrentLinked
 Queu
@@ -493,24 +493,84 @@
 le%3E(null);%0A%0A
+    @NonNull%0A    private final Object lock = new Object();%0A%0A
     public A
@@ -662,37 +662,26 @@
 s = new 
-Concurrent
 Linked
-Queue
+List
 %3CT%3E(expe
@@ -731,29 +731,18 @@
 new 
-Concurrent
 Linked
-Queue
+List
 %3CT%3E(
@@ -1259,16 +1259,69 @@
 final T 
+expected;%0A%0A        synchronized (lock) %7B%0A            
 expected
@@ -1339,32 +1339,42 @@
 Values.remove();
+%0A        %7D
 %0A%0A        boolea
@@ -1585,24 +1585,62 @@
 Millis)) %7B%0A%0A
+            synchronized (lock) %7B%0A    
             
@@ -1668,24 +1668,28 @@
 ze() %3E 0) %7B%0A
+    
             
@@ -1745,32 +1745,36 @@
                 
+    
 if (expected.equ
@@ -1786,24 +1786,28 @@
 btained)) %7B%0A
+    
             
@@ -1859,24 +1859,28 @@
             
+   
+ 
 %7D else %7B%0A   
@@ -1876,16 +1876,20 @@
  else %7B%0A
+    
         
@@ -2005,16 +2005,20 @@
         
+    
 + obtain
@@ -2047,24 +2047,46 @@
  expected);%0A
+                    %7D%0A
             
@@ -2227,32 +2227,33 @@
 l) %7B%0A           
+ 
 throw new Assert
@@ -2756,24 +2756,58 @@
 uesLeft() %7B%0A
+        synchronized (lock) %7B%0A    
         if (
@@ -2832,24 +2832,28 @@
 e() != 0) %7B%0A
+    
             
@@ -2938,32 +2938,46 @@
 xpectedValues);%0A
+            %7D%0A
         %7D%0A    %7D%0A
@@ -3138,24 +3138,58 @@
 obtained) %7B%0A
+        synchronized (lock) %7B%0A    
         try 
@@ -3202,16 +3202,20 @@
         
+  
+  
 obtained
@@ -3232,24 +3232,28 @@
 (obtained);%0A
+    
         %7D ca
@@ -3292,16 +3292,20 @@
         
+  
+  
 // Catch
@@ -3348,16 +3348,20 @@
 feature%0A
+    
         
@@ -3463,16 +3463,20 @@
         
+  
+  
 // Becau
@@ -3544,16 +3544,20 @@
         
+    
 // And i
@@ -3617,16 +3617,20 @@
         
+  
+  
 onNextOb
@@ -3661,16 +3661,30 @@
 wable);%0A
+            %7D%0A
         

Use synchronization instead of concurrent queues