{"action":"created","issue":{"url":"https://api.github.com/repos/scalaz/scalaz-stream/issues/351","labels_url":"https://api.github.com/repos/scalaz/scalaz-stream/issues/351/labels{/name}","comments_url":"https://api.github.com/repos/scalaz/scalaz-stream/issues/351/comments","events_url":"https://api.github.com/repos/scalaz/scalaz-stream/issues/351/events","html_url":"https://github.com/scalaz/scalaz-stream/issues/351","id":65004473,"number":351,"title":"Transducers to generalize Process1","user":{"login":"djspiewak","id":752,"avatar_url":"https://avatars.githubusercontent.com/u/752?v=3","gravatar_id":"","url":"https://api.github.com/users/djspiewak","html_url":"https://github.com/djspiewak","followers_url":"https://api.github.com/users/djspiewak/followers","following_url":"https://api.github.com/users/djspiewak/following{/other_user}","gists_url":"https://api.github.com/users/djspiewak/gists{/gist_id}","starred_url":"https://api.github.com/users/djspiewak/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/djspiewak/subscriptions","organizations_url":"https://api.github.com/users/djspiewak/orgs","repos_url":"https://api.github.com/users/djspiewak/repos","events_url":"https://api.github.com/users/djspiewak/events{/privacy}","received_events_url":"https://api.github.com/users/djspiewak/received_events","type":"User","site_admin":false},"labels":[{"url":"https://api.github.com/repos/scalaz/scalaz-stream/labels/design/refactoring","name":"design/refactoring","color":"fbca04"},{"url":"https://api.github.com/repos/scalaz/scalaz-stream/labels/feature","name":"feature","color":"009800"}],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":4,"created_at":"2015-03-29T00:11:39Z","updated_at":"2015-03-31T02:07:14Z","closed_at":null,"body":"```scala\r\n  sealed trait AcceptT[+F[_], -I, +O]\r\n\r\n  object AcceptT {\r\n    trait Partial[F[_], I] {\r\n      type Aux[α] = AcceptT[F, I, α]\r\n    }\r\n\r\n    final case class Effect[+F[_], O](f: F[O]) extends AcceptT[F, Any, O]\r\n    final case class Accept[+F[_], I]() extends AcceptT[F, I, I]\r\n  }\r\n\r\n  type Transducer[+F[_], -I, +O] = Process[AcceptT.Partial[F, I]#Aux, O]\r\n  // type Process1[-I, +O] = Transducer[Nothing, I, O]\r\n  // type Channel[+F[_], -I, +O] = Obsolete!!\r\n  // type Sink[+F[_], +A] = Transducer[F, A, Unit]\r\n\r\n  object transducer {\r\n\r\n    def receiveF[F[_], A, I, O](effect: F[A])(f: A => Transducer[F, I, O]): Transducer[F, I, O] =\r\n      await[AcceptT.Partial[F, I]#Aux, A, O](AcceptT.Effect(effect))(f)\r\n\r\n    def receive1[F[_], I, O](f: I => Transducer[F, I, O]): Transducer[F, I, O] =\r\n      await[AcceptT.Partial[F, I]#Aux, I, O](AcceptT.Accept[F, I]())(f)\r\n  }\r\n\r\n  implicit class TransducerSyntax[F[_], A](val self: Process[F, A]) extends AnyVal {\r\n    import Cause.{EarlyCause, End, Kill}\r\n\r\n    def transduce[B](trans: Transducer[F, A, B])(implicit F: Applicative[F], C: Catchable[F]): Process[F, B] = trans.suspendStep flatMap {\r\n      // fail suck ugliness to get around limitations in GADT pattern matching\r\n      case Halt(rsn) => self.kill onHalt { _ => Halt(rsn) }\r\n\r\n      case step: Step[AcceptT.Partial[F, A]#Aux, B] => {\r\n        val cont1 = step.next\r\n\r\n        step.head match {\r\n          case awt: Await[AcceptT.Partial[F, A]#Aux, a, B] => {\r\n            val req = awt.req\r\n            val rcv = awt.rcv\r\n\r\n            req match {\r\n              case AcceptT.Effect(eff) => {\r\n                val fp = eff.attempt map { _ leftMap { Error(_) } } map { res =>\r\n                  self transduce (cont1.+:[AcceptT.Partial[F, A]#Aux, B](rcv(res).run))    // TODO errors?  I don't have access to Util.Try, because... blergh\r\n                }\r\n\r\n                eval(fp) flatMap { a => a }\r\n              }\r\n\r\n              case AcceptT.Accept() => self.step match {\r\n                case Step(awt @ Await(_, _), cont) => awt extend { p => (p +: cont) transduce step.toProcess }\r\n\r\n                case Step(Emit(Seq()), cont) => cont.continue transduce step.toProcess\r\n\r\n                case Step(Emit(is), cont) => {\r\n                  val nextT = cont1.+:[AcceptT.Partial[F, A]#Aux, B](rcv(\\/-(is.head.asInstanceOf[a])).run)      // this is sound because Accept proves it so\r\n                  val nextS = emitAll(is.tail) +: cont\r\n\r\n                  new TransducerSyntax[F, A](nextS) transduce nextT\r\n                }\r\n\r\n                case hlt @ Halt(End) => new TransducerSyntax[F, A](hlt) transduce (step.toProcess disconnect Kill swallowKill)\r\n                case hlt @ Halt(rsn: Cause.EarlyCause) => new TransducerSyntax[F, A](hlt) transduce (step.toProcess disconnect rsn)\r\n              }\r\n            }\r\n          }\r\n\r\n          case emt @ Emit(os) => {\r\n            emt onHalt {\r\n              case End => self transduce cont1.continue\r\n              case early => self transduce (Halt(early) +: cont1) causedBy early\r\n            }\r\n          }\r\n        }\r\n      }\r\n    }\r\n  }\r\n```\r\n\r\nI have some comments to this effect, but basically `Transducer` completely subsumes `Process1` and provides a strictly more powerful construct to replace `Channel` and `Sink`.  As mentioned in the gist, `Process1[A, B] = Transducer[Nothing, A, B]`, which is pretty cool!\r\n\r\nHere's the motivation…  `Channel` allows us to define an effectful transformation on some input `Process`.  However, unlike with `Process1`, it is not possible to define a `Channel` as some sort of state machine based on input!  The transduction input elements to a `Channel` are received by the inner composite function.  If the enclosing effect is `IO`-like, it is possible for a function to *mutate* some register to change what function \"comes next\" based on the input observed, but outside of this sort of spooky action, it isn't possible.\r\n\r\nBy comparison, `Process1` makes it very very easy to define a state machine on the input stream.  That is in fact the whole purpose of `Process1`!  `Transducer` generalizes this notion and allows us to define a state machine on an input stream…while still embedding effects!  This greatly expands the space of possible effectful transformations that can be conveniently encoded, and completely obsoletes both `Channel` and `Sink` (well, `Sink` would be a transducer, but you get the idea).\r\n\r\nHere's the catch…  As with monad transformers, Scala isn't too happy inferring certain chained type constructors.  You'll notice some consequences of this in the implementation above.  Specifically:\r\n\r\n```scala\r\nself transduce (cont1.+:[AcceptT.Partial[F, A]#Aux, B](rcv(res).run))\r\n```\r\n\r\nObviously this is ugly.  The problem is that Scala is unable to *infer* the correct type parameters for `+:`.  Originally I thought that this was just a problem with `+:` and `Cont`, but unfortunately it also extends to much more useful combinators.  Like, for example, `++` on `Process`.  And before you ask, no, variance doesn't have anything to do with the problem.  I have a [relatively minimized case](https://gist.github.com/djspiewak/ac9af4519f65c6e5ad01) with no variance or subtyping of any kind.\r\n\r\nAnyway, as with monad transformers, the solution here is to punt the problem over to implicit resolution (similar to `scalaz.Unapply`), which gives us some control over the type constructor inference.  I'm still working on sorting this part out, but I have no doubt that it will work.  Hat-tip to @milessabin for putting me on to that solution…\r\n\r\nHere's the catch…  In order to make this work, in order for us to have this really nice generalization to unify *all* of our stream transducers into a single framework, in order to enable straightforward definition of *effectful* state machines on input streams (very desirable!), we need to adjust some type signatures.  Most notably, `++` is going to need to take an implicit parameter which performs the type constructor inference.  In a sense, `append` would get removed and replaced with `appendU` (think: the `U` variants of certain functions in Scalaz).  This significantly complicates our type signatures and does break binary compatibility.  Everything should infer nicely though, regardless of whether we're using `Transducer` with `++` or just a plain-old `Process[Task, _]`.  So it wouldn't change *call-site syntax* at all, just the raw types.\r\n\r\nAre there any thoughts on this?  @pchiusano?"},"comment":{"url":"https://api.github.com/repos/scalaz/scalaz-stream/issues/comments/87899410","html_url":"https://github.com/scalaz/scalaz-stream/issues/351#issuecomment-87899410","issue_url":"https://api.github.com/repos/scalaz/scalaz-stream/issues/351","id":87899410,"user":{"login":"pchiusano","id":11074,"avatar_url":"https://avatars.githubusercontent.com/u/11074?v=3","gravatar_id":"","url":"https://api.github.com/users/pchiusano","html_url":"https://github.com/pchiusano","followers_url":"https://api.github.com/users/pchiusano/followers","following_url":"https://api.github.com/users/pchiusano/following{/other_user}","gists_url":"https://api.github.com/users/pchiusano/gists{/gist_id}","starred_url":"https://api.github.com/users/pchiusano/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/pchiusano/subscriptions","organizations_url":"https://api.github.com/users/pchiusano/orgs","repos_url":"https://api.github.com/users/pchiusano/repos","events_url":"https://api.github.com/users/pchiusano/events{/privacy}","received_events_url":"https://api.github.com/users/pchiusano/received_events","type":"User","site_admin":false},"created_at":"2015-03-31T02:07:13Z","updated_at":"2015-03-31T02:07:13Z","body":"@milessabin PING :) Could you elaborate on your remarks? Maybe put together a gist of just a type-aligned cons list, in the style you think would be simpler."}}