1.x: just() construction to call the onCreate execution hook

src/main/java/rx/internal/util/ScalarSynchronousObservable.java

@@ -23,6 +23,7 @@
 import rx.internal.producers.SingleProducer;
 import rx.internal.schedulers.EventLoopsScheduler;
 import rx.observers.Subscribers;
+import rx.plugins.*;
 
 /**
  * An Observable that emits a single constant scalar value to Subscribers.
@@ -33,6 +34,8 @@
  * @param <T> the value type
  */
 public final class ScalarSynchronousObservable<T> extends Observable<T> {
+    /** The execution hook instance. */
+    static RxJavaObservableExecutionHook hook = RxJavaPlugins.getInstance().getObservableExecutionHook();
     /**
      * Indicates that the Producer used by this Observable should be fully
      * threadsafe. It is possible, but unlikely that multiple concurrent
@@ -72,14 +75,7 @@ public static <T> ScalarSynchronousObservable<T> create(T t) {
     final T t;
 
     protected ScalarSynchronousObservable(final T t) {
-        super(new OnSubscribe<T>() {
-
-            @Override
-            public void call(Subscriber<? super T> s) {
-                s.setProducer(createProducer(s, t));
-            }
-
-        });
+        super(hook.onCreate(new JustOnSubscribe<T>(t)));
         this.t = t;
     }
 
@@ -131,6 +127,20 @@ public void call() {
         return create(new ScalarAsyncOnSubscribe<T>(t, onSchedule));
     }
 
+    /** The OnSubscribe callback for the Observable constructor. */
+    static final class JustOnSubscribe<T> implements OnSubscribe<T> {
+        private final T t;
+
+        private JustOnSubscribe(T t) {
+            this.t = t;
+        }
+
+        @Override
+        public void call(Subscriber<? super T> s) {
+            s.setProducer(createProducer(s, t));
+        }
+    }
+
     /**
      * The OnSubscribe implementation that creates the ScalarAsyncProducer for each
      * incoming subscriber.

src/test/java/rx/internal/util/ScalarSynchronousObservableTest.java

@@ -16,12 +16,16 @@
 
 package rx.internal.util;
 
-import org.junit.Test;
+import java.util.concurrent.atomic.*;
+
+import org.junit.*;
 
 import rx.Observable;
+import rx.Observable.OnSubscribe;
 import rx.exceptions.TestException;
 import rx.functions.Func1;
 import rx.observers.TestSubscriber;
+import rx.plugins.*;
 import rx.schedulers.Schedulers;
 
 public class ScalarSynchronousObservableTest {
@@ -230,4 +234,30 @@ public void onNext(Integer t) {
         ts.assertError(TestException.class);
         ts.assertNotCompleted();
     }
+
+    @Test
+    public void hookCalled() {
+        RxJavaObservableExecutionHook save = ScalarSynchronousObservable.hook;
+        try {
+            final AtomicInteger c = new AtomicInteger();
+
+            ScalarSynchronousObservable.hook = new RxJavaObservableExecutionHook() {
+                @Override
+                public <T> OnSubscribe<T> onCreate(OnSubscribe<T> f) {
+                    c.getAndIncrement();
+                    return f;
+                }
+            };
+
+            int n = 10;
+
+            for (int i = 0; i < n; i++) {
+                Observable.just(1).subscribe();
+            }
+
+            Assert.assertEquals(n, c.get());
+        } finally {
+            ScalarSynchronousObservable.hook = save;
+        }
+    }
 }