Added platform check.

Author: akarnokd

src/main/java/rx/internal/operators/OperatorObserveOn.java

@@ -15,14 +15,15 @@
  */
 package rx.internal.operators;
 
+import java.util.Queue;
 import java.util.concurrent.atomic.*;
 
 import rx.Observable.Operator;
 import rx.*;
 import rx.exceptions.MissingBackpressureException;
 import rx.functions.Action0;
-import rx.internal.util.RxRingBuffer;
-import rx.internal.util.unsafe.SpscArrayQueue;
+import rx.internal.util.*;
+import rx.internal.util.unsafe.*;
 import rx.schedulers.*;
 
 /**
@@ -64,7 +65,7 @@ private static final class ObserveOnSubscriber<T> extends Subscriber<T> {
         final ScheduledUnsubscribe scheduledUnsubscribe;
         final NotificationLite<T> on = NotificationLite.instance();
 
-        final SpscArrayQueue<Object> queue = new SpscArrayQueue<Object>(RxRingBuffer.SIZE);
+        final Queue<Object> queue;
         volatile boolean completed = false;
         volatile boolean failure = false;
 
@@ -84,6 +85,11 @@ private static final class ObserveOnSubscriber<T> extends Subscriber<T> {
         public ObserveOnSubscriber(Scheduler scheduler, Subscriber<? super T> child) {
             this.child = child;
             this.recursiveScheduler = scheduler.createWorker();
+            if (UnsafeAccess.isUnsafeAvailable()) {
+                queue = new SpscArrayQueue<Object>(RxRingBuffer.SIZE);
+            } else {
+                queue = new SynchronizedQueue<Object>(RxRingBuffer.SIZE);
+            }
             this.scheduledUnsubscribe = new ScheduledUnsubscribe(recursiveScheduler);
             child.add(scheduledUnsubscribe);
             child.setProducer(new Producer() {