2.x: scalar flatMap optimization enabled

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

@@ -596,7 +596,7 @@ public static Observable<Long> intervalRange(long start, long count, long initia
     @SchedulerSupport(SchedulerKind.NONE)
     public static <T> Observable<T> just(T value) {
         Objects.requireNonNull(value);
-        return create(new PublisherScalarSource<>(value));
+        return new ObservableScalarSource<>(value);
     }
 
     @BackpressureSupport(BackpressureKind.FULL)
@@ -1648,9 +1648,9 @@ public final <R> Observable<R> flatMap(Function<? super T, ? extends Publisher<?
             throw new IllegalArgumentException("maxConcurrency > 0 required but it was " + maxConcurrency);
         }
         validateBufferSize(bufferSize);
-        if (onSubscribe instanceof PublisherScalarSource) {
-            PublisherScalarSource<T> scalar = (PublisherScalarSource<T>) onSubscribe;
-            return create(scalar.flatMap(mapper));
+        if (this instanceof ObservableScalarSource) {
+            ObservableScalarSource<T> scalar = (ObservableScalarSource<T>) this;
+            return create(scalar.scalarFlatMap(mapper));
         }
         return lift(new OperatorFlatMap<>(mapper, delayErrors, maxConcurrency, bufferSize));
     }

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

@@ -0,0 +1,58 @@
+/**
+ * Copyright 2015 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.internal.operators;
+
+import java.util.function.Function;
+
+import org.reactivestreams.*;
+
+import io.reactivex.Observable;
+import io.reactivex.internal.subscriptions.*;
+
+/**
+ * Represents a constant scalar value.
+ */
+public final class ObservableScalarSource<T> extends Observable<T> {
+    private final T value;
+    public ObservableScalarSource(T value) {
+        super(new Publisher<T>() {
+            @Override
+            public void subscribe(Subscriber<? super T> s) {
+                s.onSubscribe(new ScalarSubscription<>(s, value));
+            }
+        });
+        this.value = value;
+    }
+
+    public T value() {
+        return value;
+    }
+
+    public <U> Publisher<U> scalarFlatMap(Function<? super T, ? extends Publisher<? extends U>> mapper) {
+        return s -> {
+            Publisher<? extends U> other;
+            try {
+                other = mapper.apply(value);
+            } catch (Throwable e) {
+                EmptySubscription.error(e, s);
+                return;
+            }
+            if (other == null) {
+                EmptySubscription.error(new NullPointerException("The publisher returned by the function is null"), s);
+                return;
+            }
+            other.subscribe(s);
+        };
+    }
+}

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

@@ -1,17 +1,14 @@
-/*
- * Copyright 2011-2015 David Karnok
+/**
+ * Copyright 2015 Netflix, Inc.
  * 
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
  * 
- *     http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
  * 
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
  */
 
 package io.reactivex.internal.operators;
@@ -134,8 +131,8 @@ public void onNext(T t) {
                 onError(e);
                 return;
             }
-            if (p instanceof PublisherScalarSource) {
-                tryEmitScalar(((PublisherScalarSource<? extends U>)p).value());
+            if (p instanceof ObservableScalarSource) {
+                tryEmitScalar(((ObservableScalarSource<? extends U>)p).value());
             } else {
                 InnerSubscriber<T, U> inner = new InnerSubscriber<>(this, uniqueId++);
                 addInner(inner);
@@ -220,7 +217,11 @@ void tryEmitScalar(U value) {
                         s.request(1);
                     }
                 } else {
-
+                    Queue<U> q = getMainQueue();
+                    if (!q.offer(value)) {
+                        onError(new IllegalStateException("Scalar queue full?!"));
+                        return;
+                    }
                 }
                 if (decrementAndGet() == 0) {
                     return;

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

@@ -1,19 +1,17 @@
-/*
- * Copyright 2011-2015 David Karnok
+/**
+ * Copyright 2015 Netflix, Inc.
  * 
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
  * 
- *     http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
  * 
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
  */
 
+
 package io.reactivex.internal.operators;
 
 import java.util.function.Function;

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

@@ -1,17 +1,14 @@
-/*
- * Copyright 2011-2015 David Karnok
+/**
+ * Copyright 2015 Netflix, Inc.
  * 
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
  * 
- *     http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
  * 
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
  */
 
 package io.reactivex.internal.operators;

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

@@ -1,18 +1,16 @@
-/*
- * Copyright 2011-2015 David Karnok
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+/**
+ * Copyright 2015 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
  */
+
 package io.reactivex.internal.operators;
 
 import java.util.concurrent.CompletableFuture;

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

@@ -1,18 +1,16 @@
-/*
- * Copyright 2011-2015 David Karnok
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+/**
+ * Copyright 2015 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
  */
+
 package io.reactivex.internal.operators;
 
 import java.util.function.Supplier;

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

@@ -1,17 +1,14 @@
-/*
- * Copyright 2011-2015 David Karnok
+/**
+ * Copyright 2015 Netflix, Inc.
  * 
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
  * 
- *     http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
  * 
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
  */
 
 package io.reactivex.internal.operators;

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

@@ -1,17 +1,14 @@
-/*
- * Copyright 2011-2015 David Karnok
+/**
+ * Copyright 2015 Netflix, Inc.
  * 
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
  * 
- *     http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
  * 
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
  */
 
 package io.reactivex.internal.operators;

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

@@ -1,18 +1,16 @@
-/*
- * Copyright 2011-2015 David Karnok
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+/**
+ * Copyright 2015 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
  */
+
 package io.reactivex.internal.operators;
 
 import java.util.Iterator;

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

@@ -1,17 +1,14 @@
-/*
- * Copyright 2011-2015 David Karnok
+/**
+ * Copyright 2015 Netflix, Inc.
  * 
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
  * 
- *     http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
  * 
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
  */
 
 package io.reactivex.internal.operators;

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

@@ -1,18 +1,16 @@
-/*
- * Copyright 2011-2015 David Karnok
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+/**
+ * Copyright 2015 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
  */
+
 package io.reactivex.internal.operators;
 
 import java.util.concurrent.Callable;