Hotfix compilation error and missing generics

akarnokd · akarnokd · commit bf1cfcb5fee9 · 2016-08-08T16:02:11.000+02:00
diff --git a/src/main/java/io/reactivex/Observable.java b/src/main/java/io/reactivex/Observable.java
@@ -1050,7 +1050,7 @@ public final Observable<Boolean> any(Predicate<? super T> predicate) {
 
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Observable<T> asObservable() {
-        return new ObservableWrapper<T>(this);
+        return new ObservableFromSource<T>(this);
     }
 
     @SchedulerSupport(SchedulerSupport.NONE)
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableFromPublisher.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableFromPublisher.java
@@ -36,6 +36,9 @@ static final class PublisherSubscriber<T>
     extends AtomicBoolean
     implements Subscriber<T>, Disposable {
 
+        /** */
+        private static final long serialVersionUID = -7306579371159152354L;
+        
         private final Observer<? super T> o;
         private Subscription inner;
 
diff --git a/src/main/java/io/reactivex/observers/single/SingleSerializedObserver.java b/src/main/java/io/reactivex/observers/single/SingleSerializedObserver.java
@@ -17,13 +17,13 @@
 import io.reactivex.disposables.Disposable;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-public final class SingleSerializedObserver implements SingleObserver {
+public final class SingleSerializedObserver<T> implements SingleObserver<T> {
 
-    final SingleObserver actual;
+    final SingleObserver<? super T> actual;
 
     final AtomicBoolean once = new AtomicBoolean();
 
-    public SingleSerializedObserver(SingleObserver actual) {
+    public SingleSerializedObserver(SingleObserver<? super T> actual) {
         this.actual = actual;
     }
     
@@ -33,7 +33,7 @@ public void onSubscribe(Disposable d) {
     }
 
     @Override
-    public void onSuccess(Object value) {
+    public void onSuccess(T value) {
         if (once.compareAndSet(false, true)) {
             actual.onSuccess(value);
         }

Original file line number	Diff line number	Diff line change
`@@ -1050,7 +1050,7 @@ public final Observable<Boolean> any(Predicate<? super T> predicate) {`
`1050`	`1050`
`1051`	`1051`	`@SchedulerSupport(SchedulerSupport.NONE)`
`1052`	`1052`	`public final Observable<T> asObservable() {`
`1053`		`- return new ObservableWrapper<T>(this);`
	`1053`	`+ return new ObservableFromSource<T>(this);`
`1054`	`1054`	`}`
`1055`	`1055`
`1056`	`1056`	`@SchedulerSupport(SchedulerSupport.NONE)`
Original file line number	Diff line number	Diff line change
`@@ -17,13 +17,13 @@`
`17`	`17`	`import io.reactivex.disposables.Disposable;`
`18`	`18`	`import java.util.concurrent.atomic.AtomicBoolean;`
`19`	`19`
`20`		`-public final class SingleSerializedObserver implements SingleObserver {`
	`20`	`+public final class SingleSerializedObserver<T> implements SingleObserver<T> {`
`21`	`21`
`22`		`- final SingleObserver actual;`
	`22`	`+ final SingleObserver<? super T> actual;`
`23`	`23`
`24`	`24`	`final AtomicBoolean once = new AtomicBoolean();`
`25`	`25`
`26`		`- public SingleSerializedObserver(SingleObserver actual) {`
	`26`	`+ public SingleSerializedObserver(SingleObserver<? super T> actual) {`
`27`	`27`	`this.actual = actual;`
`28`	`28`	`}`
`29`	`29`
`@@ -33,7 +33,7 @@ public void onSubscribe(Disposable d) {`
`33`	`33`	`}`
`34`	`34`
`35`	`35`	`@Override`
`36`		`- public void onSuccess(Object value) {`
	`36`	`+ public void onSuccess(T value) {`
`37`	`37`	`if (once.compareAndSet(false, true)) {`
`38`	`38`	`actual.onSuccess(value);`
`39`	`39`	`}`