package com.github.davidmoten.rx2;

import java.util.concurrent.atomic.AtomicReference;

import com.github.davidmoten.guavamini.Preconditions;

public final class MpscQueue<T> {

    private final AtomicReference<Node<T>> head = new AtomicReference<Node<T>>();
    private final AtomicReference<Boolean> tailSet = new AtomicReference<Boolean>(false);

    // mutable
    private Node<T> tail;

    private static final class Node<T> {

        // mutable

        T value;
        Node<T> next;

        Node(T value) {
            this.value = value;
        }
    }

    public void push(T value) {
        // performs one volatile read and two CAS operations per call to this method
        // (under contention can be higher)

        Preconditions.checkNotNull(value, "cannot push a null value");
        Node<T> node = new Node<T>(value);
        while (true) {
            Node<T> h = head.get();
            if (head.compareAndSet(h, node)) {
                h.next = node;
                if (tailSet.compareAndSet(false, true)) {
                    // don't mind that this is not instantly visible to poll() because
                    // in reactive operators the push is always followed by a drain that
                    // forces a poll which will be ordered after this assignment (so this
                    // assignment will be visible)
                    tail = h;
                }
                break;
            }
        }
    }

    public T poll() {
        // performs one volatile write only when reading the end of the queue

        if (tail == null) {
            return null;
        } else {
            Node<T> temp = tail;
            Node<T> next = temp.next;
            if (next == null) {
                tail = null;
                tailSet.set(false);
            } else {
                tail = next;
            }
            T v = temp.value;
            // help gc including prevent nepotism
            temp.value = null;
            return v;
        }
    }

}

@@ -86,63 +86,8 @@
 e;%0A%0A
-import com.github.davidmoten.guavamini.Preconditions;%0A%0A
 publ
@@ -673,79 +673,8 @@
 r)%0A%0A
-        Preconditions.checkNotNull(value, %22cannot push a null value%22);%0A
     
@@ -877,17 +877,21 @@
 tailSet.
-c
+weakC
 ompareAn
@@ -1380,22 +1380,16 @@
 one 
-volatile write
+lazy set
  onl
@@ -1651,17 +1651,21 @@
 tailSet.
-s
+lazyS
 et(false
@@ -1811,16 +1811,16 @@
 epotism%0A
-
         
@@ -1834,32 +1834,62 @@
 p.value = null;%0A
+            temp.next = null;%0A
             retu

test dev