Fix StyxObservable type parameters

components/api/src/main/java/com/hotels/styx/api/StyxCoreObservable.java

@@ -25,7 +25,7 @@
 /**
  * An observable that underpins the StyxObservable interface.
  *
- * @param <T>
+ * @param <T> type of object published
  */
 class StyxCoreObservable<T> implements StyxObservable<T> {
     private final Observable<T> delegate;
@@ -38,18 +38,6 @@ public StyxCoreObservable(CompletionStage<T> future) {
         this.delegate = toObservable(future);
     }
 
-    private static <T> Observable<T> toObservable(CompletionStage<T> future) {
-        return Observable.create(subscriber ->
-                future.whenComplete((result, error) -> {
-                    if (error != null) {
-                        subscriber.onError(error);
-                    } else {
-                        subscriber.onNext(result);
-                        subscriber.onCompleted();
-                    }
-                }));
-    }
-
     public static <T> StyxObservable<T> empty() {
         return new StyxCoreObservable<>(Observable.empty());
     }
@@ -62,31 +50,23 @@ public static <T> StyxObservable<T> error(Throwable cause) {
         return new StyxCoreObservable<T>(Observable.error(cause));
     }
 
-    public <U> StyxObservable<U> map(Function<T, U> transformation) {
+    public <R> StyxObservable<R> map(Function<? super T, ? extends R> transformation) {
         return new StyxCoreObservable<>(delegate.map(transformation::apply));
     }
 
-    public <U> StyxObservable<U> flatMap(Function<T, StyxObservable<U>> transformation) {
+    public <R> StyxObservable<R> flatMap(Function<? super T, ? extends StyxObservable<? extends R>> transformation) {
         return new StyxCoreObservable<>(delegate.flatMap(response ->
                 toObservable(transformation.apply(response))));
     }
 
-    private static <U> Observable<? extends U> toObservable(StyxObservable<U> styxObservable) {
-        return styxObservable instanceof StyxCoreObservable
-                ? ((StyxCoreObservable<U>) styxObservable).delegate
-                : toObservable(styxObservable.asCompletableFuture());
-    }
-
-    public <U> StyxObservable<U> reduce(BiFunction<T, U, U> accumulator, U seed) {
+    public <R> StyxObservable<R> reduce(BiFunction<? super T, R, R> accumulator, R seed) {
         return new StyxCoreObservable<>(delegate.reduce(seed, (result, element) -> accumulator.apply(element, result)));
     }
 
     @Override
-    public StyxObservable<T> onError(Function<Throwable, StyxObservable<T>> errorHandler) {
-        return new StyxCoreObservable<>(delegate.onErrorResumeNext(cause -> {
-            StyxCoreObservable<T> result = (StyxCoreObservable<T>) errorHandler.apply(cause);
-            return result.delegate();
-        }));
+    public StyxObservable<T> onError(Function<Throwable, ? extends StyxObservable<? extends T>> errorHandler) {
+        return new StyxCoreObservable<>(delegate.onErrorResumeNext(cause ->
+                toObservable(errorHandler.apply(cause))));
     }
 
     public Observable<T> delegate() {
@@ -103,4 +83,21 @@ private static <T> CompletableFuture<T> fromSingleObservable(Observable<T> obser
         return future;
     }
 
+    private static <T> Observable<T> toObservable(StyxObservable<T> styxObservable) {
+        return styxObservable instanceof StyxCoreObservable
+                ? ((StyxCoreObservable<T>) styxObservable).delegate
+                : toObservable(styxObservable.asCompletableFuture());
+    }
+
+    private static <T> Observable<T> toObservable(CompletionStage<T> future) {
+        return Observable.create(subscriber ->
+                future.whenComplete((result, error) -> {
+                    if (error != null) {
+                        subscriber.onError(error);
+                    } else {
+                        subscriber.onNext(result);
+                        subscriber.onCompleted();
+                    }
+                }));
+    }
 }

components/api/src/main/java/com/hotels/styx/api/StyxObservable.java

@@ -27,16 +27,16 @@
  * <p>
  * This interface provides a is *not* intended for plugins to extend.
  *
- * @param <T>
+ * @param <T> type of object published
  */
 public interface StyxObservable<T> {
-    <U> StyxObservable<U> map(Function<T, U> transformation);
+    <R> StyxObservable<R> map(Function<? super T, ? extends R> transformation);
 
-    <U> StyxObservable<U> flatMap(Function<T, StyxObservable<U>> transformation);
+    <R> StyxObservable<R> flatMap(Function<? super T, ? extends StyxObservable<? extends R>> transformation);
 
-    <U> StyxObservable<U> reduce(BiFunction<T, U, U> accumulator, U initialValue);
+    <R> StyxObservable<R> reduce(BiFunction<? super T, R, R> accumulator, R initialValue);
 
-    StyxObservable<T> onError(Function<Throwable, StyxObservable<T>> errorHandler);
+    StyxObservable<T> onError(Function<Throwable, ? extends StyxObservable<? extends T>> errorHandler);
 
     /**
      * Converts this observable to a completable future. Note that in order to do this, it must
@@ -62,7 +62,7 @@ static <T> StyxObservable<T> of(T value) {
         return new StyxCoreObservable<>(Observable.just(value));
     }
 
-    static <T> StyxObservable<T> from(Iterable<T> values) {
+    static <T> StyxObservable<T> from(Iterable<? extends T> values) {
         return new StyxCoreObservable<>(Observable.from(values));
     }
 

system-tests/e2e-suite/src/test/scala/com/hotels/styx/plugins/AsyncRequestContentSpec.scala

@@ -94,8 +94,12 @@ class AsyncRequestContentDelayPlugin extends PluginAdapter {
           Thread.sleep(1000)
           Observable.just(byteBuf)
         })
-    StyxObservable.of(request)
+
+    // This was split apart as it no longer compiles without the type annotation StyxObservable[HttpRequest]
+    val mapped: StyxObservable[HttpRequest] = StyxObservable.of(request)
       .map(asJavaFunction((request: HttpRequest) => request.newBuilder().body(fromRxObservable(contentTransformation)).build()))
+
+    mapped
       .flatMap(asJavaFunction((request: HttpRequest) => chain.proceed(request)))
   }
 }