Merge branch 'master' into idiomaticscala

Author: samuelgruetter

CHANGES.md

@@ -1,5 +1,15 @@
 # RxJava Releases #
 
+### Version 0.13.4 ([Maven Central](http://search.maven.org/#search%7Cga%7C1%7Cg%3A%22com.netflix.rxjava%22%20AND%20v%3A%220.13.4%22)) ###
+
+* [Pull 393](https://github.com/Netflix/RxJava/pull/393) Parallel Operator & ObserveOn/ScheduledObserver Fixes
+* [Pull 394](https://github.com/Netflix/RxJava/pull/394) Change Interval and Sample default Scheduler
+* [Pull 391](https://github.com/Netflix/RxJava/pull/391) Fix OSGI support for rxjava-scala
+
+### Version 0.13.3
+
+* Upload to Sonatype failed so version skipped
+
 ### Version 0.13.2 ([Maven Central](http://search.maven.org/#search%7Cga%7C1%7Cg%3A%22com.netflix.rxjava%22%20AND%20v%3A%220.13.2%22)) ###
 
 * [Pull 389](https://github.com/Netflix/RxJava/pull/389) Scala Adaptor Improvements

gradle.properties

@@ -1 +1 @@
-version=0.13.3-SNAPSHOT
+version=0.13.5-SNAPSHOT

language-adaptors/rxjava-groovy/src/test/groovy/rx/lang/groovy/TestParallel.groovy

@@ -0,0 +1,21 @@
+package rx.lang.groovy
+
+import org.junit.Test
+
+import rx.Observable
+import rx.Scheduler
+import rx.concurrency.Schedulers
+import rx.util.functions.Func1
+
+class TestParallel {
+
+    @Test
+    public void testParallelOperator() {
+        Observable.range(0, 100)
+                .parallel({
+                    it.map({ return it; })
+                })
+                .toBlockingObservable()
+                .forEach({ println("T: " + it + " Thread: " + Thread.currentThread());  });
+    }
+}

language-adaptors/rxjava-scala/build.gradle

@@ -49,7 +49,7 @@ jar {
         name = 'rxjava-scala'
         instruction 'Bundle-Vendor', 'Netflix'
         instruction 'Bundle-DocURL', 'https://github.com/Netflix/RxJava'
-        instruction 'Import-Package', '!org.junit,!junit.framework,!org.mockito.*,*'
+        instruction 'Import-Package', '!org.junit,!junit.framework,!org.mockito.*,!org.scalatest.*,*'
         instruction 'Fragment-Host', 'com.netflix.rxjava.core'
     }
 }

language-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/Observable.scala

@@ -176,7 +176,7 @@ class Observable[+T](val asJava: rx.Observable[_ <: T])
    *         Observable, and that synchronously notifies its {@link Observer}s
    */
   def synchronize: Observable[T] = {
-    Observable[T](JObservable.synchronize(asJava))
+    Observable[T](asJava.synchronize)
   }
 
   /**

rxjava-core/src/main/java/rx/Observable.java

@@ -212,12 +212,23 @@ public Subscription call(Scheduler scheduler, Void state) {
     }
 
     /**
-     * Returns the scheduler's notion of current absolute time in milliseconds.
+     * @return the scheduler's notion of current absolute time in milliseconds.
      */
     public long now() {
         return System.currentTimeMillis();
     }
 
+    /**
+     * Parallelism available to a Scheduler.
+     * <p>
+     * This defaults to {@code Runtime.getRuntime().availableProcessors()} but can be overridden for use cases such as scheduling work on a computer cluster.
+     * 
+     * @return the scheduler's available degree of parallelism.
+     */
+    public int degreeOfParallelism() {
+        return Runtime.getRuntime().availableProcessors();
+    }
+
     public static class UnitTest {
         @SuppressWarnings("unchecked") // mocking is unchecked, unfortunately
         @Test

rxjava-core/src/main/java/rx/Scheduler.java

@@ -15,12 +15,15 @@
  */
 package rx.concurrency;
 
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 
 import rx.Scheduler;
 import rx.Subscription;
-import rx.operators.SafeObservableSubscription;
 import rx.subscriptions.CompositeSubscription;
 import rx.subscriptions.Subscriptions;
 import rx.util.functions.Func2;
@@ -29,27 +32,74 @@
  * Schedules work on a new thread.
  */
 public class NewThreadScheduler extends Scheduler {
-    private static final NewThreadScheduler INSTANCE = new NewThreadScheduler();
+
+    private final static NewThreadScheduler INSTANCE = new NewThreadScheduler();
+    private final static AtomicLong count = new AtomicLong();
 
     public static NewThreadScheduler getInstance() {
         return INSTANCE;
     }
 
-    @Override
-    public <T> Subscription schedule(final T state, final Func2<? super Scheduler, ? super T, ? extends Subscription> action) {
-        final SafeObservableSubscription subscription = new SafeObservableSubscription();
-        final Scheduler _scheduler = this;
+    private NewThreadScheduler() {
 
-        Thread t = new Thread(new Runnable() {
-            @Override
-            public void run() {
-                subscription.wrap(action.call(_scheduler, state));
-            }
-        }, "RxNewThreadScheduler");
+    }
 
-        t.start();
+    private static class EventLoopScheduler extends Scheduler {
+        private final ExecutorService executor;
 
-        return subscription;
+        private EventLoopScheduler() {
+            executor = Executors.newFixedThreadPool(1, new ThreadFactory() {
+
+                @Override
+                public Thread newThread(Runnable r) {
+                    return new Thread(r, "RxNewThreadScheduler-" + count.incrementAndGet());
+                }
+            });
+        }
+
+        @Override
+        public <T> Subscription schedule(final T state, final Func2<? super Scheduler, ? super T, ? extends Subscription> action) {
+            final Scheduler _scheduler = this;
+            return Subscriptions.from(executor.submit(new Runnable() {
+
+                @Override
+                public void run() {
+                    action.call(_scheduler, state);
+                }
+            }));
+        }
+
+        @Override
+        public <T> Subscription schedule(final T state, final Func2<? super Scheduler, ? super T, ? extends Subscription> action, final long delayTime, final TimeUnit unit) {
+            // we will use the system scheduler since it doesn't make sense to launch a new Thread and then sleep
+            // we will instead schedule the event then launch the thread after the delay has passed
+            final Scheduler _scheduler = this;
+            final CompositeSubscription subscription = new CompositeSubscription();
+            ScheduledFuture<?> f = GenericScheduledExecutorService.getInstance().schedule(new Runnable() {
+
+                @Override
+                public void run() {
+                    if (!subscription.isUnsubscribed()) {
+                        // when the delay has passed we now do the work on the actual scheduler
+                        Subscription s = _scheduler.schedule(state, action);
+                        // add the subscription to the CompositeSubscription so it is unsubscribed
+                        subscription.add(s);
+                    }
+                }
+            }, delayTime, unit);
+
+            // add the ScheduledFuture as a subscription so we can cancel the scheduled action if an unsubscribe happens
+            subscription.add(Subscriptions.create(f));
+
+            return subscription;
+        }
+
+    }
+
+    @Override
+    public <T> Subscription schedule(final T state, final Func2<? super Scheduler, ? super T, ? extends Subscription> action) {
+        EventLoopScheduler s = new EventLoopScheduler();
+        return s.schedule(state, action);
     }
 
     @Override

rxjava-core/src/main/java/rx/concurrency/NewThreadScheduler.java

@@ -46,7 +46,7 @@ public final class OperationInterval {
      * Creates an event each time interval.
      */
     public static OnSubscribeFunc<Long> interval(long interval, TimeUnit unit) {
-        return interval(interval, unit, Schedulers.executor(Executors.newSingleThreadScheduledExecutor()));
+        return interval(interval, unit, Schedulers.threadPoolForComputation());
     }
 
     /**

rxjava-core/src/main/java/rx/operators/OperationInterval.java

@@ -20,6 +20,7 @@
 
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.Test;
 import org.mockito.InOrder;
@@ -33,6 +34,8 @@
 import rx.Subscription;
 import rx.concurrency.ImmediateScheduler;
 import rx.concurrency.Schedulers;
+import rx.subscriptions.CompositeSubscription;
+import rx.util.functions.Func2;
 
 /**
  * Asynchronously notify Observers on the specified Scheduler.
@@ -60,7 +63,9 @@ public Subscription onSubscribe(final Observer<? super T> observer) {
                 // do nothing if we request ImmediateScheduler so we don't invoke overhead
                 return source.subscribe(observer);
             } else {
-                return source.subscribe(new ScheduledObserver<T>(observer, scheduler));
+                CompositeSubscription s = new CompositeSubscription();
+                s.add(source.subscribe(new ScheduledObserver<T>(s, observer, scheduler)));
+                return s;
             }
         }
     }