Relevant discussion: http://www.reddit.com/r/haskell/comments/1xmmtn/some_ideas_for_pipesparse/cfcs8om

conduit.hs

{-# LANGUAGE DeriveDataTypeable #-}
import           Control.Applicative       ((<$>), (<*>))
import           Control.Exception         (Exception, throwIO)
import           Control.Monad.Trans.Class (lift)
import           Data.Conduit
import           Data.Conduit.Extra        (fuseLeftovers)
import qualified Data.Conduit.List         as CL
import           Data.Typeable             (Typeable)
import           Debug.Trace               (traceShow)

data A = A | AEnd
    deriving Show
newtype B = B Int
    deriving Show

btoas (B i) = traceShow ("btoas", i) $ replicate i A ++ [AEnd]

astob :: Conduit A IO B
astob =
    loop 0
  where
    loop i = await >>= maybe (end i) (go i)

    end 0 = return ()
    end _ = lift $ throwIO IncompleteSequence
    go i A = loop (succ i)
    go i AEnd = yield (B i) >> loop 0

data IncompleteSequence = IncompleteSequence
    deriving (Show, Typeable)
instance Exception IncompleteSequence

main :: IO ()
main = do
    let src = mapM_ yield [A, A, A, A, AEnd, A, A, AEnd, A]
    res <- src $$ (,)
        <$> fuseLeftovers (concatMap btoas) astob CL.peek
        <*> CL.consume
    print res

pipes.hs

{-# LANGUAGE DeriveDataTypeable #-}
import           Control.Applicative
import           Control.Exception          (Exception, throwIO)
import           Control.Lens               (Iso', from, iso, view, zoom)
import           Control.Monad.State.Strict (evalStateT)
import           Data.Typeable              (Typeable)
import           Debug.Trace                (traceShow)
import           Pipes
import           Pipes.Core                 as Pc
import qualified Pipes.Parse                as Pp
import qualified Pipes.Prelude              as P

data A = A | AEnd
    deriving Show
newtype B = B Int
    deriving Show

btoas :: Monad m => Producer B m r -> Producer A m r
btoas p = p >-> P.map (\(B i) -> traceShow ("btoas", i) $ replicate i A ++ [AEnd]) >-> P.concat

astob :: Producer A IO r -> Producer B IO r
astob =
    loop 0
  where
    loop i p = do
        res <- lift $ next p
        case res of
            Left r
                | i == 0 -> return r
                | otherwise -> lift $ throwIO IncompleteSequence
            Right (A, p') -> loop (succ i) p'
            Right (AEnd, p') -> yield (B i) >> loop 0 p'

data IncompleteSequence = IncompleteSequence
    deriving (Show, Typeable)
instance Exception IncompleteSequence

abiso :: Iso' (Producer A IO r) (Producer B IO r)
abiso = iso astob btoas

main :: IO ()
main = do
  let src = mapM_ yield [A, A, A, A, AEnd, A, A, AEnd, A]
  let parser = (,) <$> zoom abiso Pp.peek
                   <*> Pp.drawAll
  res <- evalStateT parser src
  print res