Merge pull request #3232 from akarnokd/OperatorFirstLast2x

Operators first, last, multi-offer for SpscLinkedArrayQueue

src/main/java/io/reactivex/Observable.java

@@ -1497,4 +1497,20 @@ public final Observable<List<T>> takeLastBuffer(long time, TimeUnit unit) {
     public final Observable<List<T>> takeLastBuffer(long time, TimeUnit unit, Scheduler scheduler) {
         return takeLast(time, unit, scheduler).toList();
     }
+
+    public final Observable<T> first() {
+        return take(1).single();
+    }
+
+    public final Observable<T> first(T defaultValue) {
+        return take(1).single(defaultValue);
+    }
+
+    public final Observable<T> last() {
+        return takeLast(1).single();
+    }
+
+    public final Observable<T> last(T defaultValue) {
+        return takeLast(1).single(defaultValue);
+    }
 }

src/main/java/io/reactivex/internal/operators/OperatorSkipLastTimed.java

@@ -13,17 +13,16 @@
 
 package io.reactivex.internal.operators;
 
-import java.util.Queue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.*;
 
 import org.reactivestreams.*;
 
 import io.reactivex.Observable.Operator;
+import io.reactivex.Scheduler;
 import io.reactivex.internal.queue.SpscLinkedArrayQueue;
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
 import io.reactivex.internal.util.BackpressureHelper;
-import io.reactivex.Scheduler;
 
 public final class OperatorSkipLastTimed<T> implements Operator<T, T> {
     final long time;
@@ -88,12 +87,11 @@ public void onSubscribe(Subscription s) {
 
         @Override
         public void onNext(T t) {
-            final Queue<Object> q = queue;
+            final SpscLinkedArrayQueue<Object> q = queue;
 
             long now = scheduler.now(unit);
 
-            q.offer(now);
-            q.offer(t);
+            q.offer(now, t);
 
             drain();
         }

src/main/java/io/reactivex/internal/queue/SpscLinkedArrayQueue.java

@@ -311,4 +311,39 @@ public T remove() {
     public T element() {
         throw new UnsupportedOperationException();
     }
+
+    /**
+     * Offer two elements at the same time.
+     * <p>Don't use the regular offer() with this at all!
+     * @param first
+     * @param second
+     * @return
+     */
+    public boolean offer(T first, T second) {
+        final AtomicReferenceArray<Object> buffer = producerBuffer;
+        final long p = producerIndex;
+        final int m = producerMask;
+
+        int pi = calcWrappedOffset(p + 1, m);
+
+        if (null == lvElement(buffer, pi)) {
+            soElement(buffer, pi, second);
+            soProducerIndex(p + 2);
+            soElement(buffer, pi - 1, first);
+        } else {
+            final int capacity = buffer.length();
+            final AtomicReferenceArray<Object> newBuffer = new AtomicReferenceArray<>(capacity);
+            producerBuffer = newBuffer;
+
+            soElement(newBuffer, pi, second);// StoreStore
+            soElement(newBuffer, pi - 1, first);
+            soNext(buffer, newBuffer);
+
+            soProducerIndex(p + 2);// this ensures correctness on 32bit platforms
+
+            soElement(buffer, pi - 1, HAS_NEXT); // new buffer is visible after element is
+        }
+
+        return true;
+    }
 }