diff --git a/dynamic-pipeline.cabal b/dynamic-pipeline.cabal --- a/dynamic-pipeline.cabal +++ b/dynamic-pipeline.cabal @@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack name: dynamic-pipeline -version: 0.3.0.0 +version: 0.3.1.0 synopsis: Library Type Safe implementation of Dynamic Pipeline Paradigm (DPP). description: @dynamic-pipeline@ is a __/Type Safe/__ Dynamic and Parallel Streaming Library, which is an implementation of __Dynamic Pipeline Paradigm (DPP)__ proposed in this paper [DPP](https://biblioteca.sistedes.es/articulo/the-dynamic-pipeline-paradigm/). diff --git a/examples/Misc/RepeatedTwiceDP.hs b/examples/Misc/RepeatedTwiceDP.hs --- a/examples/Misc/RepeatedTwiceDP.hs +++ b/examples/Misc/RepeatedTwiceDP.hs @@ -18,9 +18,9 @@ source' :: Stage (ReadChannel String -> WriteChannel Int -> WriteChannel Int -> WriteChannel Double -> WriteChannel String -> DP st ()) source' = withSource @DPExample $ \feedback cout cout' cout'' toFilter -> do + finish cout' >> finish cout'' unfoldT ([1 .. 10] <> [1 .. 10]) cout identity - finish cout >> finish cout' >> finish cout'' - foldM_ feedback (`push` toFilter) + feedback |=> toFilter generator' :: GeneratorStage DPExample (Maybe Int) Int s generator' = @@ -40,9 +40,8 @@ genAction filter' cin cin' cdn cin'' _ _ odn cout = do let unfoldFilter = mkUnfoldFilterForAll filter' Just cin (cin' .*. cdn .*. cin'' .*. HNil) HCons ft (HCons sec _) <- unfoldF unfoldFilter - foldM_ ft $ flip push cout . show - finish cout - foldM_ sec $ flip push odn + ft |=>| cout $ show + sec |=>| odn $ id filterTemp :: Filter DPExample (Maybe Int) Int s filterTemp = mkFilter actorRepeted @@ -58,23 +57,20 @@ -> WriteChannel String -> StateT (Maybe Int) (DP s) () actorRepeted i rc rc' rd rs wc wc' wd wc'' = do - foldM_ rc $ \e -> do - putTextLn $ "1) Elem: " <> show e <> " - Param: " <> show i - if e /= i then push e wc else pure () - finish wc + rc |>=>| wc $ \e -> do + -- putTextLn $ "1) Elem: " <> show e <> " - Param: " <> show i + if e /= i then pure $ Just e else pure Nothing push i wc' - foldM_ rc' $ \e -> do - putTextLn $ "2) Elem: " <> show e <> " - Param: " <> show i - push e wc' - finish wc' + rc' |>=>| wc' $ \e -> do + -- putTextLn $ "2) Elem: " <> show e <> " - Param: " <> show i + pure $ Just e foldM_ rs $ \e -> let x = maybe 0 identity $ readMaybe @Int e in if i == x then push (fromIntegral x) wd else push e wc'' finish wc'' - foldM_ rd $ flip push wd - finish wd + rd |=>| wd $ id sink' :: Stage (ReadChannel Int -> ReadChannel Int -> ReadChannel Double -> DP s ()) diff --git a/src/DynamicPipeline.hs b/src/DynamicPipeline.hs --- a/src/DynamicPipeline.hs +++ b/src/DynamicPipeline.hs @@ -107,6 +107,14 @@ -- * Channels ReadChannel, WriteChannel, + (|=>), + (|=>|), + (|>=>), + (|>=>|), + mapF_, + map_, + mapM_, + mapMF_, foldM_, foldWithM_, push, @@ -127,15 +135,20 @@ -- The following is the Regular Grammar allowed to build a /DPP/ Flow definition: -- -- @ --- __DP__ = 'Source' __CHANS__ ':=>' 'Generator' __CHANS__ ':=>' 'Sink' --- __CHANS__ = 'Channel' __CH__ --- __CH__ = 'Eof' | 'Type' ':<+>' __CH__ +-- __DP__ -> 'Source' __CHANS__ ':=>' 'Generator' __CHANS__ ':=>' 'Sink' +-- __DP__ -> 'Source' __CHANS ':=>' 'Generator' __CHANS__ ':=>' __FEEDBACK__ ':=>' 'Sink' +-- __CHANS__ -> 'Channel' __CH__ +-- __FEEDBACK__ -> 'FeedbackChannel' __CH__ +-- __CH__ -> 'Type' ':<+>' __CH__ | 'Eof' -- @ -- -- Example: -- -- @ 'Source' ('Channel' (Int ':<+>' Int)) ':=>' 'Generator' ('Channel' (Int ':<+>' Int)) ':=>' 'Sink' @ -- +-- Or with Feedback Channel to retrofit Streamming +-- +-- @ 'Source' ('Channel' (Int ':<+>' Int)) ':=>' 'Generator' ('Channel' (Int ':<+>' Int)) ':=>' 'FeedbackChannel' ('String' ':<+>' 'Eof') ':=>' 'Sink' @ -- -- $dp -- 'DynamicPipeline' Data type is the point where all the information is contained in order the library can run our /DP/ Algorithm. diff --git a/src/DynamicPipeline/Channel.hs b/src/DynamicPipeline/Channel.hs --- a/src/DynamicPipeline/Channel.hs +++ b/src/DynamicPipeline/Channel.hs @@ -12,6 +12,14 @@ module DynamicPipeline.Channel ( ReadChannel , WriteChannel + , (|=>) + , (|=>|) + , (|>=>) + , (|>=>|) + , mapF_ + , map_ + , mapM_ + , mapMF_ , foldM_ , foldWithM_ , push @@ -30,10 +38,15 @@ ) import Data.ByteString as B import Data.Foldable as F + hiding ( mapM_ + ) import Data.HList hiding ( foldM_ + , mapM_ ) import GHC.IO.Handle as H import Relude as R + hiding ( mapM_ + ) -- | 'WriteChannel' can only write values into some Channel Queue @@ -46,6 +59,68 @@ -- [@a@]: Type that this Channel can read newtype ReadChannel a = ReadChannel { unRead :: OutChan (Maybe a) } +-- | 'map_' is a /Natural Transformation/ from consumer 'ReadChannel' to some producer 'WriteChannel' applying a transformation with function @f@ +{-# INLINE map_ #-} +map_ :: MonadIO m + => ReadChannel a -- ^'ReadChannel' + -> WriteChannel b -- ^'ReadChannel' + -> (a -> b) -- ^Monadic Transformation to do with read element + -> m () +map_ rc wc f = foldM_ rc $ flip push wc . f + +-- | Same as 'map_' but with 'id' combinator +(|=>) :: MonadIO m => ReadChannel a -> WriteChannel a -> m () +(|=>) rc wc = map_ rc wc id + +infixl 5 |=> + +-- | Same as 'map_' but mark Eof Channel after all processing +{-# INLINE mapF_ #-} +mapF_ :: MonadIO m + => ReadChannel a -- ^'ReadChannel' + -> WriteChannel b -- ^'ReadChannel' + -> (a -> b) -- ^Monadic Transformation to do with read element + -> m () +mapF_ rc wc f = map_ rc wc f >> finish wc + +-- | Alias 'mapF_' +(|=>|) :: MonadIO m => ReadChannel a -> WriteChannel b -> (a -> b) -> m () +(|=>|) = mapF_ + +infixl 5 |=>| + + +-- | Same as 'map_' But applying a Monadic mapping +{-# INLINE mapM_ #-} +mapM_ :: MonadIO m + => ReadChannel a -- ^'ReadChannel' + -> WriteChannel b -- ^'ReadChannel' + -> (a -> m (Maybe b)) -- ^Monadic Transformation to do with read element + -> m () +mapM_ rc wc f = foldM_ rc $ maybe (pure ()) (`push` wc) <=< f + +-- | Alias 'mapM_' +(|>=>) :: MonadIO m => ReadChannel a -> WriteChannel b -> (a -> m (Maybe b)) -> m () +(|>=>) = mapM_ + +infixr 5 |>=> + +-- | Same as 'mapM_' but mark Eof Channel after all processing +{-# INLINE mapMF_ #-} +mapMF_ :: MonadIO m + => ReadChannel a -- ^'ReadChannel' + -> WriteChannel b -- ^'ReadChannel' + -> (a -> m (Maybe b)) -- ^Monadic Transformation to do with read element + -> m () +mapMF_ rc wc f = mapM_ rc wc f >> finish wc + +-- | Alias 'mapMF_' +(|>=>|) :: MonadIO m => ReadChannel a -> WriteChannel b -> (a -> m (Maybe b)) -> m () +(|>=>|) = mapMF_ + +infixr 5 |>=>| + + -- | 'foldM_' is a /Catamorphism/ for consuming a 'ReadChannel' and do some Monadic @m@ computation with each element {-# INLINE foldM_ #-} foldM_ :: MonadIO m @@ -97,7 +172,7 @@ unfoldM = loop' where loop' seed fn stopIfM writeChannel = - ifM stopIfM (pure ()) (seed >>= flip push writeChannel . fn >> loop' seed fn stopIfM writeChannel) + ifM stopIfM (finish writeChannel) (seed >>= flip push writeChannel . fn >> loop' seed fn stopIfM writeChannel) -- | Using 'unfoldM', unfold from file {-# INLINE unfoldFile #-} unfoldFile :: MonadIO m @@ -111,7 +186,7 @@ -- | Idem 'unfoldM' but for 'Foldable', for example a List @[a]@. Useful for testing purpose {-# INLINE unfoldT #-} unfoldT :: (MonadIO m, Foldable t) => t a -> WriteChannel b -> (a -> b) -> m () -unfoldT ts writeChannel fn = forM_ ts (flip push writeChannel . fn) +unfoldT ts writeChannel fn = forM_ ts (flip push writeChannel . fn) >> finish writeChannel {-# WARNING newChannel "INTERNAL USE" #-} {-# NOINLINE newChannel #-} diff --git a/src/DynamicPipeline/Stage.hs b/src/DynamicPipeline/Stage.hs --- a/src/DynamicPipeline/Stage.hs +++ b/src/DynamicPipeline/Stage.hs @@ -42,7 +42,7 @@ import Control.Lens hiding ((<|)) import Data.HList import Data.List.NonEmpty -import DynamicPipeline.Channel +import DynamicPipeline.Channel hiding (mapM_) import DynamicPipeline.Flow import GHC.TypeLits import Relude as R @@ -87,9 +87,11 @@ ValidDP 'False = TypeError ( 'Text "Invalid Semantic for Building DP Program" ':$$: 'Text "Language Grammar:" - ':$$: 'Text "DP = Source CHANS :=> Generator CHANS :=> Sink" - ':$$: 'Text "CHANS = Channel CH" - ':$$: 'Text "CH = Type | Type :<+> CH" + ':$$: 'Text "DP -> Source CHANS :=> Generator CHANS :=> Sink" + ':$$: 'Text "DP -> Source CHANS :=> Generator CHANS :=> FEEDBACK :=> Sink" + ':$$: 'Text "CHANS -> Channel CH" + ':$$: 'Text "FEEDBACK -> FeedbackChannel CH" + ':$$: 'Text "CH -> Type :<+> CH | Eof" ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :=> Generator (Channel (Int :<+> Int)) :=> Sink'" ) @@ -109,9 +111,11 @@ ':<>: 'ShowType dpDefinition ':<>: 'Text "'" ':$$: 'Text "Language Grammar:" - ':$$: 'Text "DP = Source CHANS :=> Generator CHANS :=> Sink" - ':$$: 'Text "CHANS = Channel CH" - ':$$: 'Text "CH = Type | Type :<+> CH" + ':$$: 'Text "DP -> Source CHANS :=> Generator CHANS :=> Sink" + ':$$: 'Text "DP -> Source CHANS :=> Generator CHANS :=> FEEDBACK :=> Sink" + ':$$: 'Text "CHANS -> Channel CH" + ':$$: 'Text "FEEDBACK -> FeedbackChannel CH" + ':$$: 'Text "CH -> Type :<+> CH | Eof" ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :=> Generator (Channel (Int :<+> Int)) :=> Sink'" ) @@ -133,9 +137,11 @@ ':<>: 'ShowType dpDefinition ':<>: 'Text "'" ':$$: 'Text "Language Grammar:" - ':$$: 'Text "DP = Source CHANS :=> Generator CHANS :=> Sink" - ':$$: 'Text "CHANS = Channel CH" - ':$$: 'Text "CH = Type | Type :<+> CH" + ':$$: 'Text "DP -> Source CHANS :=> Generator CHANS :=> Sink" + ':$$: 'Text "DP -> Source CHANS :=> Generator CHANS :=> FEEDBACK :=> Sink" + ':$$: 'Text "CHANS -> Channel CH" + ':$$: 'Text "FEEDBACK -> FeedbackChannel CH" + ':$$: 'Text "CH -> Type :<+> CH | Eof" ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :=> Generator (Channel (Int :<+> Int)) :=> Sink'" ) @@ -162,9 +168,11 @@ ':<>: 'ShowType dpDefinition ':<>: 'Text "'" ':$$: 'Text "Language Grammar:" - ':$$: 'Text "DP = Source CHANS :=> Generator CHANS :=> Sink" - ':$$: 'Text "CHANS = Channel CH" - ':$$: 'Text "CH = Type | Type :<+> CH" + ':$$: 'Text "DP -> Source CHANS :=> Generator CHANS :=> Sink" + ':$$: 'Text "DP -> Source CHANS :=> Generator CHANS :=> FEEDBACK :=> Sink" + ':$$: 'Text "CHANS -> Channel CH" + ':$$: 'Text "FEEDBACK -> FeedbackChannel CH" + ':$$: 'Text "CH -> Type :<+> CH | Eof" ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :=> Generator (Channel (Int :<+> Int)) :=> Sink'" ) @@ -181,9 +189,11 @@ ':<>: 'ShowType dpDefinition ':<>: 'Text "'" ':$$: 'Text "Language Grammar:" - ':$$: 'Text "DP = Source CHANS :=> Generator CHANS :=> Sink" - ':$$: 'Text "CHANS = Channel CH" - ':$$: 'Text "CH = Type | Type :<+> CH" + ':$$: 'Text "DP -> Source CHANS :=> Generator CHANS :=> Sink" + ':$$: 'Text "DP -> Source CHANS :=> Generator CHANS :=> FEEDBACK :=> Sink" + ':$$: 'Text "CHANS -> Channel CH" + ':$$: 'Text "FEEDBACK -> FeedbackChannel CH" + ':$$: 'Text "CH -> Type :<+> CH | Eof" ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :=> Generator (Channel (Int :<+> Int)) :=> Sink'" )