Purpose of rx.util.Opening and rx.util.Closing interfaces?

[akarnokd]

Is there a particular reason these two interfaces are used by the various buffer() and window() operators instead of allowing arbitrary generic types for opening and closing buffers/windows similar to the duration selectors in join() & groupByUntil()?

[headinthebox]

I think they are historical artifacts, less types == better IMHO.

On Thu, Nov 28, 2013 at 2:47 PM, akarnokd [email protected] wrote:

Is there a particular reason these two interfaces are used by the various
buffer() and window() operators instead of allowing arbitrary generic types
for opening and closing buffers/windows similar to the duration selectors
in join() & groupByUntil()?

—
Reply to this email directly or view it on GitHubhttps://github.com//issues/540
.

[headinthebox]

On a related issue, we should move ObserverBase from rx.joins to rx right next to the Observer interface. Nobody should directly implement Observer. ObserverBase ensure that the Rx contract is enforced.

It is a bit ugly that you need to override XXXCore, so we should also add Observer.create or a constructor with overloads that take funcs, as well as one that takes an Observer itself (which can be unsafe).

You want to take this?

See

http://msdn.microsoft.com/en-us/library/hh211985(v=vs.103).aspx.
https://github.com/mono/rx/blob/master/Rx/NET/Source/System.Reactive.Core/Reactive/ObserverBase.cs (did I say already that the MSDN docs suck!)

[akarnokd]

@headinthebox Sure I would do the refactorings, but I think ObserverBase should be placed under rx.operators similar to the SafeObserver and SynchronizedObserver classes.

[headinthebox]

Probably yes, and there is a lot of duplication between all of these various observers, so you may want to have a look at that as well.

[benjchristensen]

If its job is to ensure the contract is kept then that is what SafeObserver does.

[benjchristensen]

but I think ObserverBase should be placed under rx.operators similar to the SafeObserver and SynchronizedObserver classes.

Yes, we have kept the rx package to only the most top-level things, so I agree this should be elsewhere. However, thus far the rx.operators package is not considered part of the public API and is excluded from Javadocs. If we are to have public Observers then it would be an rx.observers package like rx.observables. The rx.operators package is internal implementations and can be changed without breaking users.

[akarnokd]

I did the refactorings but couldn't get the Observable.scala working with it: akarnokd@7f36286

Could someone help with that part?

RxJava\language-adaptors\rxjava-scala\src\main\scala\rx\lang\scala\Observable.scala:307: error: type mismatch;
   found   : rx.lang.scala.Observable[Closing]
   required: rx.util.functions.Func0[_ <: rx.Observable[Closing]]
      val f: Func0[_ <: rx.Observable[Closing]] = closings().asJavaObservable
                                                                              ^
  RxJava\language-adaptors\rxjava-scala\src\main\scala\rx\lang\scala\Observable.scala:332: error: type mismatch;
   found   : Opening => rx.Observable[_ <: Closing]
   required: rx.util.functions.Func1[Opening, _ <: rx.Observable[Closing]]
      val closing: Func1[Opening, _ <: rx.Observable[Closing]] = (o: Opening) => closings(o).asJavaObservable
                                                                              ^
  RxJava\language-adaptors\rxjava-scala\src\main\scala\rx\lang\scala\Observable.scala:333: error: type mismatch;
   found   : rx.Observable[_$15] where type _$15 <: Opening
   required: rx.Observable[Opening]
  Note: _$15 <: Opening, but Java-defined class Observable is invariant in type T.
  You may wish to investigate a wildcard type such as `_ <: Opening`. (SLS 3.2.10)
      val jObs: rx.Observable[_ <: java.util.List[_]] = asJavaObservable.buffer[Opening, Closing](opening, closing)
                                                                                                  ^
  RxJava\language-adaptors\rxjava-scala\src\main\scala\rx\lang\scala\Observable.scala:545: error: overloaded method window needs result type
        asJavaObservable.window(openings.asJavaObservable, (op: Opening) => closings(op).asJavaObservable))
                               ^
  RxJava\language-adaptors\rxjava-scala\src\main\scala\rx\lang\scala\Observable.scala:519: error: overloaded method value window with alternatives:
    (x$1: Int)rx.Observable[rx.Observable[_$1]] <and>
    [TClosing](x$1: rx.util.functions.Func0[_ <: rx.Observable[TClosing]])rx.Observable[rx.Observable[_$1]]
   cannot be applied to (rx.util.functions.Func0[_$44])
      val o1: rx.Observable[_ <: rx.Observable[_]] = asJavaObservable.window(func)

[samuelgruetter]

Observable is covariant in its type parameter. In Scala, when you use a Scala type, scalac knows this automatically, because it's declared where the type is declared (using trait Observable[+T], + means covariant, - means contravariant). So rx.lang.scala.Observable[T] actually means "an rx.lang.scala.Observable whose type parameter is T or any subclass of T. However, if you use a Java type, there's no + or - at declaration site, so whenever you use rx.Observable in Scala, you have to write rx.Observable[_ <: T]. Never write rx.Observable[T].