{-# LANGUAGE RankNTypes #-}

{- |
   Module      : Streaming.Conduit
   Description : Bidirectional support for the streaming and conduit libraries
   Copyright   : Ivan Lazar Miljenovic
   License     : MIT
   Maintainer  : Ivan.Miljenovic@gmail.com


  This provides interoperability between the
  <http://hackage.haskell.org/package/streaming streaming> and
  <http://hackage.haskell.org/package/conduit conduit> libraries.

  Not only can you convert between one streaming data representation
  to the other, there is also support to use one in the middle of a
  pipeline.

  No 'B.ByteString'-based analogues of 'asConduit' and 'asStream' are
  provided as it would be of strictly less utility, requiring both the
  input and output of the 'ConduitM' to be 'ByteString'.

 -}
module Streaming.Conduit
  ( -- * Converting from Streams
    fromStream
  , fromStreamSource
  , fromStreamProducer
  , asConduit
    -- ** ByteString support
  , fromBStream
  , fromBStreamProducer
    -- * Converting from Conduits
  , toStream
  , asStream
  , sinkStream
    -- ** ByteString support
  , toBStream
  , sinkBStream
  ) where

import           Control.Monad             (join, void)
import           Control.Monad.Trans.Class (lift)
import           Data.ByteString           (ByteString)
import qualified Data.ByteString.Streaming as B
import           Data.Conduit              (Conduit, ConduitM, Producer, Source, Consumer,
                                            await, runConduit, transPipe, (.|))
import qualified Data.Conduit.List         as CL
import           Streaming                 (Of, Stream)
import qualified Streaming.Prelude         as S

--------------------------------------------------------------------------------

-- | The result of this is slightly generic than a 'Source' or a
--   'Producer'.  Subject to fusion.
fromStream :: (Monad m) => Stream (Of o) m r -> ConduitM i o m r
fromStream :: forall (m :: * -> *) o r i.
Monad m =>
Stream (Of o) m r -> ConduitM i o m r
fromStream = forall (m :: * -> *) b r a i.
Monad m =>
(b -> m (Either r (a, b))) -> b -> ConduitT i a m r
CL.unfoldEitherM forall (m :: * -> *) a r.
Monad m =>
Stream (Of a) m r -> m (Either r (a, Stream (Of a) m r))
S.next

-- | A type-specialised variant of 'fromStream' that ignores the
--   result.
fromStreamSource :: (Monad m) => Stream (Of a) m r -> Source m a
fromStreamSource :: forall (m :: * -> *) a r.
Monad m =>
Stream (Of a) m r -> Source m a
fromStreamSource = forall (f :: * -> *) a. Functor f => f a -> f ()
void forall b c a. (b -> c) -> (a -> b) -> a -> c
. forall (m :: * -> *) o r i.
Monad m =>
Stream (Of o) m r -> ConduitM i o m r
fromStream

-- | A more specialised variant of 'fromStream' that is subject to
--   fusion.
fromStreamProducer :: (Monad m) => Stream (Of a) m r -> Producer m a
fromStreamProducer :: forall (m :: * -> *) a r.
Monad m =>
Stream (Of a) m r -> Producer m a
fromStreamProducer = forall (m :: * -> *) b a i.
Monad m =>
(b -> m (Maybe (a, b))) -> b -> ConduitT i a m ()
CL.unfoldM forall (m :: * -> *) a r.
Monad m =>
Stream (Of a) m r -> m (Maybe (a, Stream (Of a) m r))
S.uncons forall b c a. (b -> c) -> (a -> b) -> a -> c
. forall (f :: * -> *) a. Functor f => f a -> f ()
void

-- | Convert a streaming 'B.ByteString' into a 'Source'; subject to fusion.
fromBStream :: (Monad m) => B.ByteString m r -> ConduitM i ByteString m r
fromBStream :: forall (m :: * -> *) r i.
Monad m =>
ByteString m r -> ConduitM i ByteString m r
fromBStream = forall (m :: * -> *) b r a i.
Monad m =>
(b -> m (Either r (a, b))) -> b -> ConduitT i a m r
CL.unfoldEitherM forall (m :: * -> *) r.
Monad m =>
ByteStream m r -> m (Either r (ByteString, ByteStream m r))
B.nextChunk

-- | A more specialised variant of 'fromBStream'.
fromBStreamProducer :: (Monad m) => B.ByteString m r -> Producer m ByteString
fromBStreamProducer :: forall (m :: * -> *) r.
Monad m =>
ByteString m r -> Producer m ByteString
fromBStreamProducer = forall (m :: * -> *) b r a i.
Monad m =>
(b -> m (Either r (a, b))) -> b -> ConduitT i a m r
CL.unfoldEitherM forall (m :: * -> *) r.
Monad m =>
ByteStream m r -> m (Either r (ByteString, ByteStream m r))
B.unconsChunk forall b c a. (b -> c) -> (a -> b) -> a -> c
. forall (f :: * -> *) a. Functor f => f a -> f ()
void

-- | Convert a 'Producer' to a 'Stream'.  Subject to fusion.
--
--   It is not possible to generalise this to be a 'ConduitM' as input
--   values are required.  If you need such functionality, see
--   'asStream'.
toStream :: (Monad m) => Producer m o -> Stream (Of o) m ()
toStream :: forall (m :: * -> *) o.
Monad m =>
Producer m o -> Stream (Of o) m ()
toStream Producer m o
cnd = forall (m :: * -> *) r. Monad m => ConduitT () Void m r -> m r
runConduit (forall (m :: * -> *) (n :: * -> *) i o r.
Monad m =>
(forall a. m a -> n a) -> ConduitT i o m r -> ConduitT i o n r
transPipe forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift Producer m o
cnd forall (m :: * -> *) a b c r.
Monad m =>
ConduitT a b m () -> ConduitT b c m r -> ConduitT a c m r
.| forall (m :: * -> *) a o.
Monad m =>
(a -> m ()) -> ConduitT a o m ()
CL.mapM_ forall (m :: * -> *) a. Monad m => a -> Stream (Of a) m ()
S.yield)

-- | Convert a 'Producer' to a 'B.ByteString' stream.  Subject to
--   fusion.
toBStream :: (Monad m) => Producer m ByteString -> B.ByteString m ()
toBStream :: forall (m :: * -> *).
Monad m =>
Producer m ByteString -> ByteString m ()
toBStream Producer m ByteString
cnd = forall (m :: * -> *) r. Monad m => ConduitT () Void m r -> m r
runConduit (forall (m :: * -> *) (n :: * -> *) i o r.
Monad m =>
(forall a. m a -> n a) -> ConduitT i o m r -> ConduitT i o n r
transPipe forall (t :: (* -> *) -> * -> *) (m :: * -> *) a.
(MonadTrans t, Monad m) =>
m a -> t m a
lift Producer m ByteString
cnd forall (m :: * -> *) a b c r.
Monad m =>
ConduitT a b m () -> ConduitT b c m r -> ConduitT a c m r
.| forall (m :: * -> *) a o.
Monad m =>
(a -> m ()) -> ConduitT a o m ()
CL.mapM_ forall (m :: * -> *). ByteString -> ByteStream m ()
B.chunk)

-- | Treat a 'Conduit' as a function between 'Stream's.  Subject to
--   fusion.
asStream :: (Monad m) => Conduit i m o -> Stream (Of i) m () -> Stream (Of o) m ()
asStream :: forall (m :: * -> *) i o.
Monad m =>
Conduit i m o -> Stream (Of i) m () -> Stream (Of o) m ()
asStream Conduit i m o
cnd Stream (Of i) m ()
stream = forall (m :: * -> *) o.
Monad m =>
Producer m o -> Stream (Of o) m ()
toStream (forall (m :: * -> *) o r i.
Monad m =>
Stream (Of o) m r -> ConduitM i o m r
fromStream Stream (Of i) m ()
stream forall (m :: * -> *) a b c r.
Monad m =>
ConduitT a b m () -> ConduitT b c m r -> ConduitT a c m r
.| Conduit i m o
cnd)

-- | Treat a 'Consumer' as a function which consumes a 'Stream'.
--   Subject to fusion.
sinkStream :: (Monad m) => Consumer i m r -> Stream (Of i) m () -> m r
sinkStream :: forall (m :: * -> *) i r.
Monad m =>
Consumer i m r -> Stream (Of i) m () -> m r
sinkStream Consumer i m r
cns Stream (Of i) m ()
stream = forall (m :: * -> *) r. Monad m => ConduitT () Void m r -> m r
runConduit (forall (m :: * -> *) o r i.
Monad m =>
Stream (Of o) m r -> ConduitM i o m r
fromStream Stream (Of i) m ()
stream forall (m :: * -> *) a b c r.
Monad m =>
ConduitT a b m () -> ConduitT b c m r -> ConduitT a c m r
.| Consumer i m r
cns)

-- | Treat a 'Consumer' as a function which consumes a 'B.ByteString'.
--   Subject to fusion.
sinkBStream :: (Monad m) => Consumer ByteString m r -> B.ByteString m () -> m r
sinkBStream :: forall (m :: * -> *) r.
Monad m =>
Consumer ByteString m r -> ByteString m () -> m r
sinkBStream Consumer ByteString m r
cns ByteString m ()
stream = forall (m :: * -> *) r. Monad m => ConduitT () Void m r -> m r
runConduit (forall (m :: * -> *) r i.
Monad m =>
ByteString m r -> ConduitM i ByteString m r
fromBStream ByteString m ()
stream forall (m :: * -> *) a b c r.
Monad m =>
ConduitT a b m () -> ConduitT b c m r -> ConduitT a c m r
.| Consumer ByteString m r
cns)

-- | Treat a function between 'Stream's as a 'Conduit'.  May be
--   subject to fusion.
asConduit :: (Monad m) => (Stream (Of i) m () -> Stream (Of o) m r) -> ConduitM i o m r
asConduit :: forall (m :: * -> *) i o r.
Monad m =>
(Stream (Of i) m () -> Stream (Of o) m r) -> ConduitM i o m r
asConduit Stream (Of i) m () -> Stream (Of o) m r
f = forall (m :: * -> *) a. Monad m => m (m a) -> m a
join forall b c a. (b -> c) -> (a -> b) -> a -> c
. forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap (forall (m :: * -> *) o r i.
Monad m =>
Stream (Of o) m r -> ConduitM i o m r
fromStream forall b c a. (b -> c) -> (a -> b) -> a -> c
. Stream (Of i) m () -> Stream (Of o) m r
f) forall a b. (a -> b) -> a -> b
$ forall {a} {o}. ConduitT a o m (Stream (Of a) m ())
go
  where
    -- Probably not the best way to go about it, but it works.
    go :: ConduitT a o m (Stream (Of a) m ())
go = do Maybe a
mo <- forall (m :: * -> *) i o. Monad m => ConduitT i o m (Maybe i)
await
            case Maybe a
mo of
              Maybe a
Nothing -> forall (m :: * -> *) a. Monad m => a -> m a
return (forall (m :: * -> *) a. Monad m => a -> m a
return ())
              Just a
o  -> forall (m :: * -> *) a r.
Monad m =>
a -> Stream (Of a) m r -> Stream (Of a) m r
S.cons a
o forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> ConduitT a o m (Stream (Of a) m ())
go