Remove Processor Interface

[benjchristensen]

What is the purpose of Processor being in the API?

Based on my reading of the code and understanding of the intents of this API, I suggest removing the Processor class: https://github.com/reactive-streams/reactive-streams/blob/master/spi/src/main/java/org/reactivestreams/api/Processor.java

It does not seem relevant to the purpose of this project. Libraries such as Akka, RxJava, Reactor, Finagle etc can and will layer logic on top of the streams of data and integrate via the Subscriber class to receive and transform data.

[viktorklang]

It is crucial to be able to represent a detached "stage" (having an inlet and an outlet), and since Java has a, erm, "rustic" type system, you cannot represent Consumer[T] with Producer[T], which means that you need an interface that extends both.
Hence, Processor.

[benjchristensen]

Please provide some pseudo code as I don't see how the interop goals have anything to do with detached stages.

[smaldini]

I will feed the thread with some reactor code which implements Processor.

Sent from my iPhone

On 15 Apr 2014, at 17:12, Ben Christensen notifications@github.com wrote:

Please provide some pseudo code as I don't see how the interop goals have anything to do with detached stages.

—
Reply to this email directly or view it on GitHub.

[viktorklang]

Pseudo:

   def wrap[T](p: Processor[T]): Processor[T] = doSomethingBefore() produceTo p produceTo doSomethingAfter()

[benjchristensen]

Sorry I'm so slow, but I'm still not seeing what that example has to do with interop between libraries.

Can you give me an example of how Akka -> Rx -> Reactor would need Processor?

[viktorklang]

I'm thinking:

Rx -> AnyGenericReactiveStreamsImpl -> Rx

as in:

  def splice[T](anyGenericReactiveStreamsImpl: Processor[T]) =
    doRxyThings() subscribe anyGenericReactiveStreamsImpl subscribe doMoarRxyThings()

As mentioned before, the only reason it is needed is because Scala is the only(?) jvm-lang with a typesystem that supports saying Subscriber with Publisher without requiring that to be a concrete type, i.e. interface Processor extends Subscriber with Publisher.

The alternative encoding would be to do something runtime-fail-y like:

   def splice[T](genericSubscriber: Subscriber[T], genericPublisher: Publisher[T]) = {
       assert(genericSubscriber == genericPublisher, "You broke it")
       doRxyThings() subscribe genericSubscriber
       genericPublisher subscribe doMoarRxyThings()
    }

Thoughts?

[benjchristensen]

Would it be correct to say that the only reason this would be needed is if we are wanting to define interoperable transformations? If so, that's beyond the original scope of allowing interop of streams but moves into interoperable stream processing.

[viktorklang]

@benjchristensen No it's just a consequence of a Publisher also being able to be a Subscriber.

[rkuhn]

I think I understand what Ben means: passing around a Processor means that I give someone a piece of hose to fit it into a pipeline, which emphasizes the transformation being passed around. We have such a case in our TCP code, when you connect outwards you get back a Processor since you can view the server end as a transformation you send ByteStrings through. It would not be a big difference to say that you get a pair of Publisher and Subscriber instead, I think.

The main question here is whether or not we will see many incompatible types being created in implementations, and I think the answer is yes: if you provide a DSL for creating some asynchronous processing steps, then the return type of the “create me the machine” method must return both interfaces at once (with Scala being the only possible exception).

By not providing this type in the SPI we will foster the creation of many copies that only inter-operate single-sidedly (i.e. you can pass an Akka Processor as either a Publisher or Subscriber, but there is no shared Processor).

I would argue for keeping this unification since most implementations will otherwise create one.

[benjchristensen]

This conversation has long since been superseded by discussions in #37 and #19

[benjchristensen]

Reopening as this discussion has been revived: #56 (comment)