title: Connecting Pipes with Consumers and Producers that return different values
body: <p>I am writing a  streaming function with the pipes ecosystem, and pipes-concurrency in particular, which is based on the operational library to allow me to quickly make little program snippets which I yield commands to a server over the network or to the stdin/out of a shell command and then read back the response.  In this case it is asterisk, but could be generalized to be anything similar.</p>
  
  <p>I initially wrote this with pipes in mind, but it doesn't work.  The reason the following code doesn't work is that astPipe returns a <code>Pipe _ _ IO a</code>, whereas both i and o, from pipes-concurrency both return <code>Consumer/Producer _ IO ()</code>.  I thought about having <code>astPipe</code> yield <code>Maybe ByteString</code>, and then making the output <code>Consumer</code> consume <code>Maybe ByteString</code>, but that still doesn't solve the problem of the <code>Producer</code> returning <code>()</code>.</p>
  
  <p>I feel like I'm really close to a solution, but I can't quite eek it out.  You should be able to just run stack on this file to replicate.</p>
  
  <pre><code>#!/usr/bin/env stack
  -- stack --resolver lts-6.20 runghc --package pipes  --package pipes-concurrency  --package operational --package process-streaming
  
  {-# LANGUAGE OverloadedStrings, LambdaCase #-}
  {-# LANGUAGE ScopedTypeVariables #-}
  {-# LANGUAGE GADTs #-}
  module West.Asterisk where
  
  import System.Process.Streaming as PS
  import Control.Monad.Operational as Op
  
  import Pipes as P
  import Pipes.Concurrent as PC;
  
  import qualified Data.ByteString.Char8 as B
  
  import Control.Concurrent.Async
  
  import GHC.IO.Exception (ExitCode)
  
  data Version = Version String
  data Channels = Channels
  
  data AsteriskInstruction a where
      Login :: AsteriskInstruction (Maybe Version)
      CoreShowChannels :: AsteriskInstruction (Maybe Channels)
  
  type Asterisk a = Program AsteriskInstruction a
  
  runAsterisk :: forall a. Asterisk a -&gt; IO a
  runAsterisk m = 
    let
  
      runAsterisk' :: Producer B.ByteString {- TODO Response -} IO () -&gt; Consumer B.ByteString IO () -&gt; Asterisk a -&gt; IO a
      runAsterisk' i o m' = runEffect $ i &gt;-&gt; astPipe m' &gt;-&gt; o
        where
          astPipe :: Asterisk a -&gt; Pipe B.ByteString B.ByteString IO a
          astPipe k = 
            case Op.view m' of
  
              Return a -&gt; return a
  
              Login :&gt;&gt;= k -&gt; do
                yield logincmd
                resp &lt;- await -- :: Response
                let v = undefined resp :: Maybe Version
                astPipe (k v)
  
              CoreShowChannels :&gt;&gt;= k -&gt; do
                yield coreshowchannelscmd
                resp &lt;- await
                let c = undefined resp :: Maybe Channels
                astPipe (k c)
  
    in do
      withSpawn unbounded $ \(out1, in1) -&gt; do
          async $ asteriskManager (fromInput in1) (toOutput out1)
          runAsterisk' (fromInput in1) (toOutput out1) m 
  
  asteriskManager :: Producer B.ByteString IO () -&gt; Consumer B.ByteString IO () -&gt; IO ExitCode
  asteriskManager prod cons = do
    let ssh = shell "nc someserver 5038"
    execute (piped ssh) (foldOut (withConsumer cons) *&gt; feedProducer prod *&gt; exitCode)
  
  
  logincmd, coreshowchannelscmd :: B.ByteString
  logincmd = "action: login\nusername: username\nsecret: pass\nevents: off\n\n"
  coreshowchannelscmd = "action: coreshowchannels\n\n"
  </code></pre>
  
  <p>The error:</p>
  
  <pre><code>  Blah.hs:38:45:
      Couldn't match type ‘a’ with ‘()’
        ‘a’ is a rigid type variable bound by
            the type signature for runAsterisk :: Asterisk a -&gt; IO a
            at Blah.hs:33:23
      Expected type: Proxy () B.ByteString () B.ByteString IO ()
        Actual type: Pipe B.ByteString B.ByteString IO a
      Relevant bindings include
        astPipe :: Asterisk a -&gt; Pipe B.ByteString B.ByteString IO a
          (bound at Blah.hs:41:9)
        m' :: Asterisk a (bound at Blah.hs:38:22)
        runAsterisk' :: Producer B.ByteString IO ()
                        -&gt; Consumer B.ByteString IO () -&gt; Asterisk a -&gt; IO a
          (bound at Blah.hs:38:5)
        m :: Asterisk a (bound at Blah.hs:34:13)
        runAsterisk :: Asterisk a -&gt; IO a (bound at Blah.hs:34:1)
      In the second argument of ‘(&gt;-&gt;)’, namely ‘astPipe m'’
      In the first argument of ‘(&gt;-&gt;)’, namely ‘i &gt;-&gt; astPipe m'’
  </code></pre>