Add ZPipeline to Processor conversion

[pragmaxim]

org.reactivestreams.Processor is an equivalent of ZPipeline or akka.stream.scaladsl.Flow and when integrating between different reactive "implementation", it is possible to create it using org.reactivestreams.FlowAdapters.toFlowProcessor ... Akka-stream has flow.toProcessor method but I can't find a way how to turn ZPipeline into a Processor.

Any idea?

[runtologist]

Not tested, bu shouldn't this work?

  def processorToZPipeline[In, Out](processor: Processor[In, Out]): ZPipeline[Scope, Throwable, In, Out] =
    ZPipeline.unwrap(
      for {
        signalErrorSink    <- processor.toZIOSink
        (signalError, sink) = signalErrorSink
        stream              = processor.toZIOStream()
        sinkPipe            = ZPipeline.fromSink(sink)
        sourcePipe          = ZPipeline.fromChannel(stream.channel)
      } yield sinkPipe >>> sourcePipe
    )

[runtologist]

Well, it doesn't.

[mschuwalow]

@runtologist are you looking at this one? If not -- I just hit a usecase for this at work, so I can take care of it.