dynamic-pipeline 0.2.0.0 → 0.3.0.0

raw patch · 9 files changed

+386/−270 lines, 9 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• dynamic-pipeline.cabal +2/−1
• examples/Graph/ConnectedComp.hs +6/−6
• examples/Main.hs +16/−4
• examples/Misc/RepeatedDP.hs +3/−3
• examples/Misc/RepeatedTwiceDP.hs +84/−0
• src/DynamicPipeline.hs +11/−9
• src/DynamicPipeline/Channel.hs +25/−16
• src/DynamicPipeline/Flow.hs +123/−144
• src/DynamicPipeline/Stage.hs +116/−87

@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack  name:           dynamic-pipeline-version:        0.2.0.0+version:        0.3.0.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/).@@ -110,6 +110,7 @@       Graph.ConnComp       Graph.ConnectedComp       Misc.RepeatedDP+      Misc.RepeatedTwiceDP   hs-source-dirs:       examples   default-extensions:

@@ -14,8 +14,8 @@  -- brittany-disable-next-binding type DPConnComp = Source (Channel (Edge :<+> ConnectedComponents :<+> Eof))-                :>> Generator (Channel (Edge :<+> ConnectedComponents :<+> Eof))-                :>> Sink+                :=> Generator (Channel (Edge :<+> ConnectedComponents :<+> Eof))+                :=> Sink  source' :: FilePath         -> Stage@@ -24,7 +24,7 @@   $ \edgeOut _ -> unfoldFile filePath edgeOut (toEdge . decodeUtf8)  sink' :: Stage (ReadChannel Edge -> ReadChannel ConnectedComponents -> DP st ())-sink' = withSink @DPConnComp $ \_ cc -> withDP $ foldM cc print+sink' = withSink @DPConnComp $ \_ cc -> withDP $ foldM_ cc print  generator' :: GeneratorStage DPConnComp ConnectedComponents Edge st generator' =@@ -41,7 +41,7 @@        -> WriteChannel ConnectedComponents        -> StateT ConnectedComponents (DP st) () actor1 _ readEdge _ writeEdge _ = -  foldM readEdge $ \e -> get >>= doActor e+  foldM_ readEdge $ \e -> get >>= doActor e  where   doActor v conn     | toConnectedComp v `intersect` conn = modify (toConnectedComp v <>)@@ -54,7 +54,7 @@        -> WriteChannel ConnectedComponents        -> StateT ConnectedComponents (DP st) () actor2 _ _ readCC _ writeCC = do -  foldM' readCC pushMemory $ \e -> get >>= doActor e+  foldWithM_ readCC pushMemory $ \e -> get >>= doActor e   where    pushMemory = get >>= flip push writeCC@@ -73,7 +73,7 @@ genAction filter' readEdge readCC _ writeCC = do   let unfoldFilter = mkUnfoldFilterForAll filter' toConnectedComp readEdge (readCC .*. HNil)    results <- unfoldF unfoldFilter-  foldM (hHead results) (`push` writeCC)+  foldM_ (hHead results) (`push` writeCC)   program :: FilePath -> IO ()

@@ -2,11 +2,13 @@  import           Graph.ConnectedComp                               as CC import           Misc.RepeatedDP                                   as Repeated+import           Misc.RepeatedTwiceDP                              as RepeatedTwice import           Options.Applicative                               as Opt import           Relude   data ProgramOptions = RepeatedElements+                    | RepeatedFeedback                     | ConnectedComponents FilePath                     deriving(Show) @@ -14,8 +16,11 @@ programDesc :: ParserInfo ProgramOptions programDesc = info (everyProgram <**> helper) (fullDesc <> header "Examples on dynamic-pipeline library")  where-  everyProgram =-    subparser (command "repeated-elements" subCmdRepeated <> command "connected-components" subCmdConnectedComp)+  everyProgram = subparser+    (  command "repeated-elements"    subCmdRepeated+    <> command "repeated-feedback"    subCmdRepeatedFeed+    <> command "connected-components" subCmdConnectedComp+    )  subCmdRepeated :: ParserInfo ProgramOptions subCmdRepeated = info@@ -24,6 +29,12 @@     "Given a list of 2000 repeated Integers filter and output unique 1000 integers"   ) +subCmdRepeatedFeed :: ParserInfo ProgramOptions+subCmdRepeatedFeed = info+  (pure RepeatedFeedback)+  (fullDesc <> header "repeated-feedback - Dynamic Pipeline Examples" <> progDesc+    "Given a list of 2000 repeated Integers filter and output unique 1000 integers and do it twice with feedback"+  )  subCmdConnectedComp :: ParserInfo ProgramOptions subCmdConnectedComp = info@@ -45,6 +56,7 @@  main' :: ProgramOptions -> IO () main' = \case-    RepeatedElements -> Repeated.program-    ConnectedComponents file -> CC.program file+  RepeatedElements         -> Repeated.program+  RepeatedFeedback         -> RepeatedTwice.program+  ConnectedComponents file -> CC.program file 

@@ -11,7 +11,7 @@ import           DynamicPipeline import           Relude -type DPExample = Source (Channel (Int :<+> Eof)) :>> Generator (Channel (Int :<+> Eof)) :>> Sink+type DPExample = Source (Channel (Int :<+> Eof)) :=> Generator (Channel (Int :<+> Eof)) :=> Sink  source' :: Stage (WriteChannel Int -> DP s ()) source' = withSource @DPExample $ \cout -> unfoldT ([1 .. 1000] <> [1 .. 1000]) cout identity@@ -36,10 +36,10 @@              -> ReadChannel Int              -> WriteChannel Int              -> StateT (Maybe Int) (DP s) ()-actorRepeted i rc wc = foldM rc $ \e -> if e /= i then push e wc else pure ()+actorRepeted i rc wc = foldM_ rc $ \e -> if e /= i then push e wc else pure ()  sink' :: Stage (ReadChannel Int -> DP s ())-sink' = withSink @DPExample $ flip foldM print+sink' = withSink @DPExample $ flip foldM_ print  program :: IO () program = runDP $ mkDP @DPExample source' generator' sink'

@@ -0,0 +1,84 @@+-- |+-- Module      : Misc.RepeatedTwiceDP+-- Copyright   : (c) 2021 Juan Pablo Royo Sales+--+-- License     : BSD3+-- Maintainer  : juanpablo.royo@gmail.com+-- Stability   : experimental+-- Portability : GHC+module Misc.RepeatedTwiceDP where++import           DynamicPipeline+import           Relude++type DPExample =   Source (Channel (Int :<+> Int :<+> Double :<+> String :<+> Eof)) +               :=> Generator (Channel (Int :<+> Int :<+> Double :<+> Eof)) +               :=> FeedbackChannel (String :<+> Eof)+               :=> Sink++source' :: Stage (ReadChannel String -> WriteChannel Int -> WriteChannel Int -> WriteChannel Double -> WriteChannel String -> DP st ())+source' = withSource @DPExample $ \feedback cout cout' cout'' toFilter -> do +    unfoldT ([1 .. 10] <> [1 .. 10]) cout identity+    finish cout >> finish cout' >> finish cout''+    foldM_ feedback (`push` toFilter)++generator' :: GeneratorStage DPExample (Maybe Int) Int s+generator' =+  let gen = withGenerator @DPExample genAction+   in  mkGenerator gen filterTemp++genAction :: Filter DPExample (Maybe Int) Int st +          -> ReadChannel Int +          -> ReadChannel Int +          -> ReadChannel Double+          -> ReadChannel String +          -> WriteChannel Int +          -> WriteChannel Int +          -> WriteChannel Double+          -> WriteChannel String +          -> DP st ()+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 ++filterTemp :: Filter DPExample (Maybe Int) Int s +filterTemp = mkFilter actorRepeted++actorRepeted :: Int+             -> ReadChannel Int+             -> ReadChannel Int+             -> ReadChannel Double+             -> ReadChannel String+             -> WriteChannel Int+             -> WriteChannel Int+             -> WriteChannel Double+             -> 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+  push i wc'+  foldM_ rc' $ \e -> do +    putTextLn $ "2) Elem: " <> show e <> " - Param: " <> show i+    push e wc'+  finish wc'+  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+++sink' :: Stage (ReadChannel Int -> ReadChannel Int -> ReadChannel Double -> DP s ())+sink' = withSink @DPExample $ \_ _ ci -> foldM_ ci print++program :: IO ()+program = runDP $ mkDP @DPExample source' generator' sink'

@@ -22,7 +22,7 @@ -- @ -- import "DynamicPipeline" ----- type DPExample = 'Source' ('Channel' (Int ':<+>' 'Eof')) ':>>' 'Generator' ('Channel' (Int ':<+>' 'Eof')) ':>>' 'Sink'+-- type DPExample = 'Source' ('Channel' (Int ':<+>' 'Eof')) ':=>' 'Generator' ('Channel' (Int ':<+>' 'Eof')) ':=>' 'Sink' --  -- source' :: 'Stage' ('WriteChannel' Int -> 'DP' s ()) -- source' = 'withSource' @DPExample $ \cout -> 'unfoldT' ([1 .. 1000] <> [1 .. 1000]) cout identity@@ -73,7 +73,8 @@       Generator,       Source,       Channel,-      type (:>>)(..),+      FeedbackChannel,+      type (:=>)(..),       type (:<+>)(..),        -- * Smart Constructors        DynamicPipeline,@@ -102,21 +103,22 @@       mkUnfoldFilter',       mkUnfoldFilterForAll,       mkUnfoldFilterForAll',-      (.*.), HList(HNil), hHead, +      (.*.), HList(HNil,HCons), hHead, hTail,       -- * Channels       ReadChannel,       WriteChannel,-      foldM,-      foldM',+      foldM_,+      foldWithM_,       push,       pull,+      finish,       unfoldM,       unfoldFile,       unfoldT     )     where -import Data.HList ((.*.), HList(HNil), hHead)+import Data.HList ((.*.), HList(HNil,HCons), hHead, hTail) import DynamicPipeline.Flow import DynamicPipeline.Channel import DynamicPipeline.Stage@@ -125,14 +127,14 @@ -- The following is the Regular Grammar allowed to build a /DPP/ Flow definition: --  -- @--- __DP__     = 'Source'  __CHANS__ ':>>' 'Generator' __CHANS__ ':>>' 'Sink'+-- __DP__     = 'Source'  __CHANS__ ':=>' 'Generator' __CHANS__ ':=>' 'Sink' -- __CHANS__  = 'Channel' __CH__ -- __CH__     = 'Eof' | 'Type' ':<+>' __CH__ -- @ -- -- Example:  -- --- @ 'Source' ('Channel' (Int ':<+>' Int)) ':>>' 'Generator' ('Channel' (Int ':<+>' Int)) ':>>' 'Sink' @+-- @ 'Source' ('Channel' (Int ':<+>' Int)) ':=>' 'Generator' ('Channel' (Int ':<+>' Int)) ':=>' 'Sink' @ -- -- -- $dp@@ -149,7 +151,7 @@ -- -- >>> import Relude -- >>> import DynamicPipeline--- >>> type DPEx = Source (Channel (Int :<+> Eof)) :>> Generator (Channel (Int :<+> Eof)) :>> Sink+-- >>> type DPEx = Source (Channel (Int :<+> Eof)) :=> Generator (Channel (Int :<+> Eof)) :=> Sink -- >>> :t withSource @DPEx -- withSource @DPEx --   :: forall k (st :: k).

@@ -12,8 +12,8 @@ module DynamicPipeline.Channel   ( ReadChannel   , WriteChannel-  , DynamicPipeline.Channel.foldM-  , foldM'+  , foldM_+  , foldWithM_   , push   , pull   , unfoldM@@ -21,6 +21,7 @@   , unfoldT   , newChannel   , end+  , finish   ) where  import qualified Control.Concurrent                                as CC@@ -29,7 +30,8 @@                                                                                                                       ) import           Data.ByteString                                   as B import           Data.Foldable                                     as F-import           Data.HList+import           Data.HList                                                                                    hiding ( foldM_+                                                                                                                      ) import           GHC.IO.Handle                                     as H import           Relude                                            as R @@ -44,23 +46,24 @@ -- [@a@]: Type that this Channel can read newtype ReadChannel a = ReadChannel { unRead :: OutChan (Maybe a) } --- | 'foldM' is a /Catamorphism/ for consuming a 'ReadChannel' and do some Monadic @m@ computation with each element-{-# INLINE foldM #-}-foldM :: MonadIO m-      => ReadChannel a -- ^'ReadChannel'-      -> (a -> m ()) -- ^Computation to do with read element-      -> m ()-foldM = flip foldM' (pure ())---- | Idem 'foldM' but allows pass a monadic computation to perform at the end of the Channel-{-# INLINE foldM' #-}-foldM' :: MonadIO m+-- | 'foldM_' is a /Catamorphism/ for consuming a 'ReadChannel' and do some Monadic @m@ computation with each element+{-# INLINE foldM_ #-}+foldM_ :: MonadIO m        => ReadChannel a -- ^'ReadChannel'-       -> m () -- ^Computation to do at the end of the channel        -> (a -> m ()) -- ^Computation to do with read element        -> m ()-foldM' = loop' where loop' c onNothing io = maybe onNothing (\e -> io e >> loop' c onNothing io) =<< liftIO (pull c)+foldM_ = flip foldWithM_ (pure ()) +-- | Idem 'foldM_' but allows pass a monadic computation to perform at the end of the Channel+{-# INLINE foldWithM_ #-}+foldWithM_ :: MonadIO m+           => ReadChannel a -- ^'ReadChannel'+           -> m () -- ^Computation to do at the end of the channel+           -> (a -> m ()) -- ^Computation to do with read element+           -> m ()+foldWithM_ = loop'+  where loop' c onNothing io = maybe onNothing (\e -> io e >> loop' c onNothing io) =<< liftIO (pull c)+ -- | Push element @a@ into 'WriteChannel' {-# INLINE push #-} push :: MonadIO m => a -> WriteChannel a -> m ()@@ -71,6 +74,11 @@ pull :: MonadIO m => ReadChannel a -> m (Maybe a) pull = liftIO . readChan (CC.threadDelay 100) . unRead +-- | Finalize Channel to indicate EOF mark and allow progress on following consumers+finish :: MonadIO m => WriteChannel a -> m ()+finish = liftIO . end++ -- | Coalgebra with Monadic computation to Feed some 'WriteChannel' -- -- [@m@]: Monadic computation wrapping Coalgebra@@ -114,4 +122,5 @@ {-# INLINE end #-} end :: WriteChannel a -> IO () end = flip writeChan Nothing . unWrite+ 

@@ -15,32 +15,35 @@       Generator,       Source,       Channel,-      type (:>>)(..),+      FeedbackChannel,+      type (:=>)(..),       type (:<+>)(..),       ChanIn,+      ChanInIn,       ChanOut,       ChanOutIn,       ChansFilter,       ChanWriteSource,       ChanReadWriteGen,       ChanReadOut,+      ChanRecord(..),+      InOutChan(..),       MkCh(..),       MkChans(..),       ExpandGenToCh,        ExpandSinkToCh,       ExpandSourceToCh,       ExpandFilterToCh,-      AllChans,-      FilterChans,       inGenOut,+      toHList,        makeChans,+      makeChansF,       getFilterChannels     ) where  import           Control.Lens                             hiding ((<|)) import           Data.Foldable                            as F import           Data.HList-import           Data.HList.Labelable import           DynamicPipeline.Channel import           Relude                                   as R @@ -65,6 +68,12 @@ -- @ a ~ (Type ':<+>' Type ':<+>' ... ':<+>' Eof) @ data Channel (a :: Type) +-- |'FeedbackChannel' is the Container Type of /Open Union Type/ which is going to be defined with ':<+>' and indicates that this+-- | Channel is for feedback to Source+--+-- @ a ~ (Type ':<+>' Type ':<+>' ... ':<+>' Eof) @+data FeedbackChannel (a :: Type)+ -- | This is the Type level function of the /Open Union Type/ for Channels.  --  -- Channels forms an /Open Union Type/ in each stage because according to __DPP__ we can have multiple /In/ and /Out/ Channels @@ -83,10 +92,10 @@ --  -- This should have the form: ----- @ 'Source' ('Channel' ..) ':>>' 'Generator' ('Channel' ..) ':>>' 'Sink' @-data a :>> b = a :>> b+-- @ 'Source' ('Channel' ..) ':=>' 'Generator' ('Channel' ..) ':=>' 'Sink' @+data a :=> b = a :=> b   deriving (Typeable, Eq, Show, Functor, Traversable, Foldable, Bounded)-infixr 5 :>>+infixr 5 :=>  -- Internal Data Types for expanding function based on Channel definitions {-# WARNING ChanIn "INTERNAL USE" #-}@@ -95,6 +104,8 @@ data ChanOut (a :: Type) {-# WARNING ChanOutIn "INTERNAL USE" #-} data ChanOutIn (a :: Type) (b :: Type)+{-# WARNING ChanInIn "INTERNAL USE" #-}+data ChanInIn (a :: Type) (b :: Type) {-# WARNING ChansFilter "INTERNAL USE" #-} data ChansFilter (a :: Type) {-# WARNING ChanWriteSource "INTERNAL USE" #-}@@ -107,7 +118,7 @@ -- Type encoding for Building Chans. Only for internal use in the Associated Type Family and combinators of MkCh and MkChans -- For accessing Dynamic Indexed Records of Channels {-# WARNING inLabel "INTERNAL USE" #-}-inLabel :: Label "Source"+inLabel :: Label "source" inLabel = Label  {-# WARNING genLabel "INTERNAL USE" #-}@@ -115,7 +126,7 @@ genLabel = Label  {-# WARNING outLabel "INTERNAL USE" #-}-outLabel :: Label "Sink"+outLabel :: Label "sink" outLabel = Label  {-# WARNING inChLabel "INTERNAL USE" #-}@@ -134,10 +145,10 @@   mkCh :: Proxy a -> IO (HList (HChI a), HList (HChO a))  instance MkCh more => MkCh (a :<+> more) where-  type HChI (a :<+> more) = WriteChannel a ': HChI more-  type HChO (a :<+> more) = ReadChannel a ': HChO more+  type HChI (a :<+> more) = ReadChannel a ': HChI more+  type HChO (a :<+> more) = WriteChannel a ': HChO more   mkCh _ = do-    (i, o) <- newChannel @a+    (o, i) <- newChannel @a     (il, ol) <- mkCh (Proxy @more)     return (i .*. il, o .*. ol) @@ -150,19 +161,34 @@ {-# WARNING ExpandToHList "INTERNAL USE" #-} type family ExpandToHList (a :: Type) (param :: Type) :: [Type] type instance ExpandToHList (ChanWriteSource ( Source (Channel inToGen)-                                          :>> Generator (Channel genToOut)-                                          :>> Sink )-                            ) _ = HChI inToGen+                                          :=> Generator (Channel genToOut)+                                          :=> Sink )+                            ) _ = HChO inToGen+type instance ExpandToHList (ChanWriteSource ( Source (Channel inToGen)+                                          :=> Generator (Channel genToOut)+                                          :=> FeedbackChannel toSource+                                          :=> Sink )+                            ) _ = HAppendListR (HChI toSource) (HChO inToGen)  type instance ExpandToHList (ChanReadWriteGen ( Source (Channel inToGen)-                                            :>> Generator (Channel genToOut)-                                            :>> Sink)-                            ) filter = filter ': HAppendListR (HChO inToGen) (HChI genToOut)+                                            :=> Generator (Channel genToOut)+                                            :=> Sink)+                            ) filter = filter ': HAppendListR (HChI inToGen) (HChO genToOut)+type instance ExpandToHList (ChanReadWriteGen ( Source (Channel inToGen)+                                            :=> Generator (Channel genToOut)+                                            :=> FeedbackChannel toSource+                                            :=> Sink)+                            ) filter = filter ': HAppendListR (HAppendListR (HChI inToGen) (HChO genToOut)) (HChO toSource)  type instance ExpandToHList (ChanReadOut ( Source (Channel inToGen)-                                       :>> Generator (Channel genToOut)-                                       :>> Sink )-                            ) filter = HChO genToOut+                                       :=> Generator (Channel genToOut)+                                       :=> Sink )+                            ) filter = HChI genToOut+type instance ExpandToHList (ChanReadOut ( Source (Channel inToGen)+                                       :=> Generator (Channel genToOut)+                                       :=> FeedbackChannel toSource+                                       :=> Sink )+                            ) filter = HChI genToOut  {-# WARNING ExpandSourceToCh "INTERNAL USE" #-} type ExpandSourceToCh a = ExpandToHList (ChanWriteSource a) Void@@ -173,164 +199,117 @@ {-# WARNING ExpandSinkToCh "INTERNAL USE" #-} type ExpandSinkToCh a = ExpandToHList (ChanReadOut a) Void +data InOutChan lr lw = InOutChan+  { _iocReadChans  :: HList lr+  , _iocWriteChans :: HList lw+  }++toHList :: HAppendList lr lw => InOutChan lr lw -> HList (HAppendListR lr lw)+toHList InOutChan{..} = _iocReadChans `hAppend` _iocWriteChans++data ChanRecord slr slw glr glw silr silw = ChanRecord+  { _crSource :: InOutChan slr slw+  , _crGenerator :: InOutChan glr glw+  , _crSink :: InOutChan silr silw+  }+ -- Class for building Channels base on a DP Definition on `a` Type {-# WARNING MkChans "INTERNAL USE" #-} class MkChans (a :: Type) where   type HChan a :: Type   mkChans :: Proxy a -> IO (HChan a) --- Instance for Building Channels for all the Chain Source :>> Generator :>> Sink+-- Instance for Building Channels for all the Chain Source :=> Generator :=> Sink instance ( MkCh inToGen          , MkCh genToOut)-    => MkChans (Source (Channel inToGen) :>> Generator (Channel genToOut) :>> Sink) where+    => MkChans (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> Sink) where -  type HChan (Source (Channel inToGen) :>> Generator (Channel genToOut) :>> Sink)-    = Record '[ Tagged "Source" (Record '[ Tagged "in-ch" (HList (HChI inToGen))-                                        , Tagged "out-ch" (HList (HChO inToGen))-                                        ]-                        )-       , Tagged "generator" (Record '[ Tagged "in-ch" (HList (HChI genToOut))-                                     , Tagged "out-ch" (HList (HChO genToOut))-                                     ]-                            )-       , Tagged "Sink" (Record '[ Tagged "in-ch" (HList (HChI genToOut))])-       ]+  type HChan (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> Sink)+      = ChanRecord '[] (HChO inToGen) (HChI inToGen) (HChO genToOut) (HChI genToOut) '[]    mkChans _ =  do-    (ii, io) <- mkCh (Proxy @inToGen)-    (gi, go) <- mkCh (Proxy @genToOut)-    (oi, _)  <- mkCh (Proxy @genToOut)-    return $ (inLabel .=. (inChLabel .=. ii .*. outChLabel .=. io .*. emptyRecord))-              .*.-              (genLabel .=. (inChLabel .=. gi .*. outChLabel .=. go .*. emptyRecord))-              .*.-              (outLabel .=. (inChLabel .=. oi .*. emptyRecord))-              .*.-              emptyRecord+    (rs, ws) <- mkCh (Proxy @inToGen)+    (rg, wg) <- mkCh (Proxy @genToOut)+    return $ ChanRecord+      { _crSource = InOutChan HNil ws+      , _crGenerator = InOutChan rs wg+      , _crSink = InOutChan rg HNil+      } +instance ( MkCh inToGen+         , MkCh genToOut+         , MkCh toSource+         , HAppendList (HChO genToOut) (HChO toSource))+    => MkChans (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> FeedbackChannel toSource :=> Sink) where++  type HChan (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> FeedbackChannel toSource :=> Sink)+    = ChanRecord (HChI toSource) (HChO inToGen) +                 (HChI inToGen) (HAppendListR (HChO genToOut) (HChO toSource)) +                 (HChI genToOut) '[]++  mkChans _ =  do+    (rf, wf) <- mkCh (Proxy @toSource)+    (rs, ws) <- mkCh (Proxy @inToGen)+    (rg, wg) <- mkCh (Proxy @genToOut)+    return $ ChanRecord+      { _crSource = InOutChan rf ws+      , _crGenerator = InOutChan rs (wg `hAppend` wf)+      , _crSink = InOutChan rg HNil+      }+ -- Instance for Building Only Channels for Filters on each Generator action instance MkCh inToGen-    => MkChans (ChansFilter (Source (Channel inToGen) :>> Generator (Channel genToOut) :>> Sink)) where+    => MkChans (ChansFilter (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> Sink)) where -  type HChan (ChansFilter (Source (Channel inToGen) :>> Generator (Channel genToOut) :>> Sink))-    = Record '[ Tagged "in-ch" (HList (HChO inToGen))-              , Tagged "out-ch" (HList (HChI inToGen))-              ]+  type HChan (ChansFilter (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> Sink))+      = InOutChan (HChI inToGen) (HChO inToGen) -  mkChans _ =  do-    (writes', reads') <- mkCh (Proxy @inToGen)-    return $ mkRecord (inChLabel .=. reads' .*. outChLabel .=. writes' .*. HNil)+  mkChans _ =  uncurry InOutChan <$> mkCh (Proxy @inToGen) +instance ( MkCh inToGen+         )+    => MkChans (ChansFilter (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> FeedbackChannel toSource :=> Sink)) where +  type HChan (ChansFilter (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> FeedbackChannel toSource :=> Sink))+    = InOutChan (HChI inToGen) (HChO inToGen)++  mkChans _ =  uncurry InOutChan <$> mkCh (Proxy @inToGen)++ {-# WARNING makeChans "INTERNAL USE" #-} {-# INLINE makeChans #-}-makeChans :: forall (a :: Type). MkChans a => IO (HChan a)+makeChans :: forall (a :: Type) slr slw glr glw silr silw. (MkChans a, HChan a ~ ChanRecord slr slw glr glw silr silw) => IO (ChanRecord slr slw glr glw silr silw) makeChans = mkChans (Proxy @a) +{-# WARNING makeChansF "INTERNAL USE" #-}+{-# INLINE makeChansF #-}+makeChansF :: forall (a :: Type) flr flw. (MkChans a, HChan a ~ InOutChan flr flw) => IO (InOutChan flr flw)+makeChansF = mkChans (Proxy @a)+ -- Ugly Dynamic Indexed Record Viewer to generate specific list of channels {-# WARNING sourceChans "INTERNAL USE" #-} {-# INLINE sourceChans #-}-sourceChans :: ( LabeledOpticF (LabelableTy r1) (Const t1)-              , LabeledOpticP (LabelableTy r1) (->)-              , LabeledOpticTo (LabelableTy r1) "in-ch" (->)-              , LabeledOpticF (LabelableTy r2) (Const t1)-              , LabeledOpticP (LabelableTy r2) (->)-              , LabeledOpticTo (LabelableTy r2) "Source" (->)-              , Labelable "in-ch" r1 s t2 t1 t1-              , Labelable "Source" r2 t3 t3 (r1 s) (r1 t2))-           => r2 t3 -> t1-sourceChans = let inl  = hLens' inLabel-                  inch = hLens' inChLabel-               in view (inl . inch)+sourceChans :: HAppendList slr slw => ChanRecord slr slw _ _ _ _ -> HList (HAppendListR slr slw)+sourceChans = toHList . _crSource  {-# WARNING generatorChans "INTERNAL USE" #-} {-# INLINE generatorChans #-}-generatorChans :: ( LabeledOpticF (LabelableTy r1) (Const (HList l1))-                  , LabeledOpticP (LabelableTy r1) (->)-                  , LabeledOpticTo (LabelableTy r1) "out-ch" (->)-                  , LabeledOpticF (LabelableTy r2) (Const (HList l2))-                  , LabeledOpticP (LabelableTy r2) (->)-                  , LabeledOpticTo (LabelableTy r2) "in-ch" (->)-                  , LabeledOpticF (LabelableTy r3) (Const (HList l2))-                  , LabeledOpticP (LabelableTy r3) (->)-                  , LabeledOpticTo (LabelableTy r3) "generator" (->)-                  , LabeledOpticF (LabelableTy r3) (Const (HList l1))-                  , LabeledOpticTo (LabelableTy r3) "Source" (->)-                  , HAppendList l1 l2-                  , Labelable "generator" r3 t1 t1 (r2 s1) (r2 t2)-                  , Labelable "in-ch" r2 s1 t2 (HList l2) (HList l2)-                  , Labelable "Source" r3 t1 t1 (r1 s2) (r1 t3)-                  , Labelable "out-ch" r1 s2 t3 (HList l1) (HList l1))-               => r3 t1 -> HList (HAppendListR l1 l2)-generatorChans ch = let inl  = hLens' inLabel-                        genl  = hLens' genLabel-                        inch = hLens' inChLabel-                        outch = hLens' outChLabel-                        outsIn = view (inl . outch) ch-                        insGen = view (genl . inch) ch-                     in outsIn `hAppendList` insGen+generatorChans :: HAppendList glr glw => ChanRecord _ _ glr glw _ _ -> HList (HAppendListR glr glw)+generatorChans = toHList . _crGenerator + {-# WARNING sinkChans "INTERNAL USE" #-} {-# INLINE sinkChans #-}-sinkChans :: ( LabeledOpticF (LabelableTy r1) (Const t1)-               , LabeledOpticP (LabelableTy r1) (->)-               , LabeledOpticTo (LabelableTy r1) "out-ch" (->)-               , LabeledOpticF (LabelableTy r2) (Const t1)-               , LabeledOpticP (LabelableTy r2) (->)-               , LabeledOpticTo (LabelableTy r2) "generator" (->)-               , Labelable "generator" r2 t2 t2 (r1 s) (r1 t3)-               , Labelable "out-ch" r1 s t3 t1 t1)-            => r2 t2 -> t1-sinkChans = let genl  = hLens' genLabel-                outch = hLens' outChLabel-             in view (genl . outch)--{-# WARNING AllChans "INTERNAL USE" #-}-type AllChans r2 r3 l1 r4 l2 t2 s t1 s2 t5 l3 l4 = (LabeledOpticTo (LabelableTy r2) "in-ch" (->),-            LabeledOpticF (LabelableTy r3) (Const (HList l3)),-            LabeledOpticP (LabelableTy r3) (->),-            LabeledOpticTo (LabelableTy r3) "Source" (->),-            LabeledOpticF (LabelableTy r2) (Const (HList l1)),-            LabeledOpticP (LabelableTy r2) (->),-            LabeledOpticTo (LabelableTy r2) "out-ch" (->),-            LabeledOpticTo (LabelableTy r4) "in-ch" (->),-            LabeledOpticF (LabelableTy r3) (Const (HList l2)),-            LabeledOpticTo (LabelableTy r3) "generator" (->),-            LabeledOpticF (LabelableTy r3) (Const (HList l1)),-            LabeledOpticF (LabelableTy r2) (Const (HList l3)),-            LabeledOpticF (LabelableTy r4) (Const (HList l4)),-            LabeledOpticP (LabelableTy r4) (->),-            LabeledOpticTo (LabelableTy r4) "out-ch" (->),-            LabeledOpticF (LabelableTy r3) (Const (HList l4)),-            LabeledOpticF (LabelableTy r4) (Const (HList l2)),-            HAppendList l1 l2, Labelable "generator" r3 t2 t2 (r4 s) (r4 t1),-            Labelable "in-ch" r2 s2 t5 (HList l3) (HList l3),-            Labelable "in-ch" r4 s t1 (HList l2) (HList l2),-            Labelable "Source" r3 t2 t2 (r2 s2) (r2 t5),-            Labelable "out-ch" r2 s2 t5 (HList l1) (HList l1),-            Labelable "out-ch" r4 s t1 (HList l4) (HList l4))+sinkChans :: HAppendList silr silw => ChanRecord _ _ _ _ silr silw -> HList (HAppendListR silr silw)+sinkChans = toHList . _crSink  {-# WARNING inGenOut "INTERNAL USE" #-} {-# INLINE inGenOut #-}-inGenOut :: AllChans r2 r3 l1 r4 l2 t2 s t1 s2 t5 l3 l4 => r3 t2 -> (HList l3, HList (HAppendListR l1 l2), HList l4)+inGenOut :: (HAppendList slr slw, HAppendList glr glw, HAppendList silr silw) => ChanRecord slr slw glr glw silr silw -> (HList (HAppendListR slr slw), HList (HAppendListR glr glw), HList (HAppendListR silr silw)) inGenOut ch = (sourceChans ch, generatorChans ch, sinkChans ch) --{-# WARNING FilterChans "INTERNAL USE" #-}-type FilterChans r b t a = (LabeledOpticF (LabelableTy r) (Const b),-                            LabeledOpticTo (LabelableTy r) "out-ch" (->),-                            Labelable "out-ch" r t t b b,-                            LabeledOpticF (LabelableTy r) (Const a),-                            LabeledOpticP (LabelableTy r) (->),-                            LabeledOpticTo (LabelableTy r) "in-ch" (->),-                            Labelable "in-ch" r t t a a)- {-# WARNING getFilterChannels "INTERNAL USE" #-} {-# INLINE getFilterChannels #-}-getFilterChannels :: FilterChans r b t a => r t -> (a, b)-getFilterChannels ch =-   let inch = hLens' inChLabel-       outch = hLens' outChLabel-       reads' = ch^.inch-       writes' = ch^.outch-    in (reads', writes')+getFilterChannels :: InOutChan lr lw -> (HList lr, HList lw)+getFilterChannels InOutChan{..} = (_iocReadChans, _iocWriteChans)+

@@ -56,15 +56,21 @@ -- | FCF - Type Level Defunctionalization -- 'IsDP' Validates if /DP/ Flow Type Level Definition is Correct according to the Grammar ----- [@dpDefinition ~ 'Source' ('Channel' ..) ':>>' 'Generator' ('Channel' ..) ':>>' 'Sink'@]: /DP/ Type level Flow Definition+-- [@dpDefinition ~ 'Source' ('Channel' ..) ':=>' 'Generator' ('Channel' ..) ':=>' 'Sink'@]: /DP/ Type level Flow Definition -- type family IsDP (dpDefinition :: k) :: Bool where   IsDP (Source (Channel inToGen)-        :>> Generator (Channel genToOut)-        :>> Sink)+        :=> Generator (Channel genToOut)+        :=> Sink)                                             = And (IsDP (Source (Channel inToGen))) (IsDP (Generator (Channel genToOut)))-  IsDP (Source (Channel (a :<+> more)))      = IsDP (Source (Channel more))-  IsDP (Source (Channel Eof))                = 'True+  IsDP ( Source (Channel inToGen)+         :=> Generator (Channel genToOut)+         :=> FeedbackChannel toSource +         :=> Sink +        )+                                            = And (IsDP (Source (Channel inToGen))) (IsDP (Generator (Channel genToOut)))+  IsDP (Source (Channel (a :<+> more)))     = IsDP (Source (Channel more))+  IsDP (Source (Channel Eof))               = 'True   IsDP (Generator (Channel (a :<+> more)))  = IsDP (Generator (Channel more))   IsDP (Generator (Channel a))              = 'True   IsDP x                                    = 'False@@ -81,16 +87,18 @@   ValidDP 'False = TypeError                     ( 'Text "Invalid Semantic for Building DP Program"                       ':$$: 'Text "Language Grammar:"-                      ':$$: 'Text "DP    = Source CHANS :>> Generator CHANS :>> Sink"+                      ':$$: 'Text "DP    = Source CHANS :=> Generator CHANS :=> Sink"                       ':$$: 'Text "CHANS = Channel CH"                       ':$$: 'Text "CH    = Type | Type :<+> CH"-                      ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :>> Generator (Channel (Int :<+> Int)) :>> Sink'"+                      ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :=> Generator (Channel (Int :<+> Int)) :=> Sink'"                     )  -- Inductive Type Family for Expanding and building Source, Generator, Filter and Sink Functions Signatures type family WithSource (dpDefinition :: Type) (monadicAction :: Type -> Type) :: Type where-  WithSource (Source (Channel inToGen) :>> Generator (Channel genToOut) :>> Sink) monadicAction-                                                                    = WithSource (ChanIn inToGen) monadicAction+  WithSource (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> Sink) monadicAction+                                                                     = WithSource (ChanIn inToGen) monadicAction+  WithSource (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> FeedbackChannel toSource :=> Sink) monadicAction+                                                                     = WithSource (ChanOutIn toSource inToGen) monadicAction   WithSource (ChanIn (dpDefinition :<+> more)) monadicAction         = WriteChannel dpDefinition -> WithSource (ChanIn more) monadicAction   WithSource (ChanIn Eof) monadicAction                              = monadicAction ()   WithSource (ChanOutIn (dpDefinition :<+> more) ins) monadicAction  = ReadChannel dpDefinition -> WithSource (ChanOutIn more ins) monadicAction@@ -101,53 +109,70 @@                                                                           ':<>: 'ShowType dpDefinition                                                                           ':<>: 'Text "'"                                                                           ':$$: 'Text "Language Grammar:"-                                                                          ':$$: 'Text "DP    = Source CHANS :>> Generator CHANS :>> Sink"+                                                                          ':$$: 'Text "DP    = Source CHANS :=> Generator CHANS :=> Sink"                                                                           ':$$: 'Text "CHANS = Channel CH"                                                                           ':$$: 'Text "CH    = Type | Type :<+> CH"-                                                                          ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :>> Generator (Channel (Int :<+> Int)) :>> Sink'"+                                                                          ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :=> Generator (Channel (Int :<+> Int)) :=> Sink'"                                                                         )  type family WithGenerator (a :: Type) (filter :: Type) (monadicAction :: Type -> Type) :: Type where-  WithGenerator (Source (Channel inToGen) :>> Generator (Channel genToOut) :>> Sink) filter monadicAction-                                                                    = filter -> WithGenerator (ChanOutIn inToGen genToOut) filter monadicAction-  WithGenerator (ChanIn (a :<+> more)) filter monadicAction         = WriteChannel a -> WithGenerator (ChanIn more) filter monadicAction-  WithGenerator (ChanIn Eof) filter monadicAction                   = monadicAction ()-  WithGenerator (ChanOutIn (a :<+> more) ins) filter monadicAction  = ReadChannel a -> WithGenerator (ChanOutIn more ins) filter monadicAction-  WithGenerator (ChanOutIn Eof ins) filter monadicAction            = WithGenerator (ChanIn ins) filter monadicAction-  WithGenerator dpDefinition _ _                                     = TypeError-                                                                        ( 'Text "Invalid Semantic for Generator Stage"-                                                                          ':$$: 'Text "in the DP Definition '"-                                                                          ':<>: 'ShowType dpDefinition-                                                                          ':<>: 'Text "'"-                                                                          ':$$: 'Text "Language Grammar:"-                                                                          ':$$: 'Text "DP    = Source CHANS :>> Generator CHANS :>> Sink"-                                                                          ':$$: 'Text "CHANS = Channel CH"-                                                                          ':$$: 'Text "CH    = Type | Type :<+> CH"-                                                                          ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :>> Generator (Channel (Int :<+> Int)) :>> Sink'"-                                                                        )-+  WithGenerator (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> Sink) filter monadicAction+                                                                         = filter -> WithGenerator (ChanOutIn inToGen genToOut) filter monadicAction+  WithGenerator (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> FeedbackChannel toSource :=> Sink) filter monadicAction+                                                                         = filter -> WithGenerator (ChanOutIn inToGen (ChanInIn genToOut toSource)) filter monadicAction+  WithGenerator (ChanInIn (a :<+> more) ins) filter monadicAction        = WriteChannel a -> WithGenerator (ChanInIn more ins) filter monadicAction+  WithGenerator (ChanInIn Eof ins) filter monadicAction                  = WithGenerator (ChanIn ins) filter monadicAction+  WithGenerator (ChanIn (a :<+> more)) filter monadicAction              = WriteChannel a -> WithGenerator (ChanIn more) filter monadicAction+  WithGenerator (ChanIn Eof) filter monadicAction                        = monadicAction ()+  WithGenerator (ChanOutIn (a :<+> more) ins) filter monadicAction       = ReadChannel a -> WithGenerator (ChanOutIn more ins) filter monadicAction+  WithGenerator (ChanOutIn Eof (ChanInIn ins ins2)) filter monadicAction = WithGenerator (ChanInIn ins ins2) filter monadicAction+  WithGenerator (ChanOutIn Eof ins) filter monadicAction                 = WithGenerator (ChanIn ins) filter monadicAction+  WithGenerator dpDefinition _ _                                         = TypeError+                                                                             ( 'Text "Invalid Semantic for Generator Stage"+                                                                               ':$$: 'Text "in the DP Definition '"+                                                                               ':<>: 'ShowType dpDefinition+                                                                               ':<>: 'Text "'"+                                                                               ':$$: 'Text "Language Grammar:"+                                                                               ':$$: 'Text "DP    = Source CHANS :=> Generator CHANS :=> Sink"+                                                                               ':$$: 'Text "CHANS = Channel CH"+                                                                               ':$$: 'Text "CH    = Type | Type :<+> CH"+                                                                               ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :=> Generator (Channel (Int :<+> Int)) :=> Sink'"+                                                                             )+      type family WithFilter (dpDefinition :: Type) (param :: Type) (monadicAction :: Type -> Type) :: Type where-  WithFilter (Source (Channel inToGen) :>> Generator (Channel genToOut) :>> Sink) param monadicAction+  WithFilter (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> Sink) param monadicAction                                                     = param -> WithFilter (ChanOutIn inToGen genToOut) param monadicAction-  WithFilter (ChanIn (dpDefinition :<+> more)) param monadicAction         = WriteChannel dpDefinition -> WithFilter (ChanIn more) param monadicAction-  WithFilter (ChanIn Eof) param monadicAction                   = monadicAction ()-  WithFilter (ChanOutIn (dpDefinition :<+> more) ins) param monadicAction  = ReadChannel dpDefinition -> WithFilter (ChanOutIn more ins) param monadicAction+  WithFilter (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> FeedbackChannel toSource  :=> Sink) param monadicAction+                                                    = param -> WithFilter (ChanOutIn inToGen (ChanInIn genToOut toSource)) param monadicAction+  WithFilter (ChanInIn (a :<+> more) ins) param monadicAction       +                                                    = WriteChannel a -> WithFilter (ChanInIn more ins) param monadicAction+  WithFilter (ChanInIn Eof ins) param monadicAction                 +                                                    = WithFilter (ChanIn ins) param monadicAction+  WithFilter (ChanIn (dpDefinition :<+> more)) param monadicAction         +                                                    = WriteChannel dpDefinition -> WithFilter (ChanIn more) param monadicAction+  WithFilter (ChanIn Eof) param monadicAction       = monadicAction ()+  WithFilter (ChanOutIn (dpDefinition :<+> more) ins) param monadicAction  +                                                    = ReadChannel dpDefinition -> WithFilter (ChanOutIn more ins) param monadicAction+  WithFilter (ChanOutIn Eof (ChanInIn ins ins2)) param monadicAction +                                                    = WithFilter (ChanInIn ins ins2) param monadicAction   WithFilter (ChanOutIn Eof ins) param m            = WithFilter (ChanIn ins) param m-  WithFilter dpDefinition _ _                                  = TypeError-                                                ( 'Text "Invalid Semantic Semantic for Generator Stage"-                                                  ':$$: 'Text "in the DP Definition '"-                                                  ':<>: 'ShowType dpDefinition-                                                  ':<>: 'Text "'"-                                                  ':$$: 'Text "Language Grammar:"-                                                  ':$$: 'Text "DP    = Source CHANS :>> Generator CHANS :>> Sink"-                                                  ':$$: 'Text "CHANS = Channel CH"-                                                  ':$$: 'Text "CH    = Type | Type :<+> CH"-                                                  ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :>> Generator (Channel (Int :<+> Int)) :>> Sink'"-                                                )+  WithFilter dpDefinition _ _                       = TypeError+                                                        ( 'Text "Invalid Semantic Semantic for Generator Stage"+                                                          ':$$: 'Text "in the DP Definition '"+                                                          ':<>: 'ShowType dpDefinition+                                                          ':<>: 'Text "'"+                                                          ':$$: 'Text "Language Grammar:"+                                                          ':$$: 'Text "DP    = Source CHANS :=> Generator CHANS :=> Sink"+                                                          ':$$: 'Text "CHANS = Channel CH"+                                                          ':$$: 'Text "CH    = Type | Type :<+> CH"+                                                          ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :=> Generator (Channel (Int :<+> Int)) :=> Sink'"+                                                        )  type family WithSink (dpDefinition :: Type) (monadicAction :: Type -> Type) :: Type where-  WithSink (Source (Channel inToGen) :>> Generator (Channel genToOut) :>> Sink) monadicAction-                                                                = WithSink (ChanOut genToOut) monadicAction+  WithSink (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> Sink) monadicAction+                                                              = WithSink (ChanOut genToOut) monadicAction+  WithSink (Source (Channel inToGen) :=> Generator (Channel genToOut) :=> FeedbackChannel toSource :=> Sink) monadicAction+                                                              = WithSink (ChanOut genToOut) monadicAction   WithSink (ChanOut (dpDefinition :<+> more)) monadicAction   = ReadChannel dpDefinition -> WithSink (ChanOut more) monadicAction   WithSink (ChanOut Eof) monadicAction                        = monadicAction ()   WithSink dpDefinition _                                     = TypeError@@ -156,10 +181,10 @@                                                                       ':<>: 'ShowType dpDefinition                                                                       ':<>: 'Text "'"                                                                       ':$$: 'Text "Language Grammar:"-                                                                      ':$$: 'Text "DP    = Source CHANS :>> Generator CHANS :>> Sink"+                                                                      ':$$: 'Text "DP    = Source CHANS :=> Generator CHANS :=> Sink"                                                                       ':$$: 'Text "CHANS = Channel CH"                                                                       ':$$: 'Text "CH    = Type | Type :<+> CH"-                                                                      ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :>> Generator (Channel (Int :<+> Int)) :>> Sink'"+                                                                      ':$$: 'Text "Example: 'Source (Channel (Int :<+> Int)) :=> Generator (Channel (Int :<+> Int)) :=> Sink'"                                                                     )  @@ -211,7 +236,7 @@ -- | 'DynamicPipeline' data type which contains all the three Stages definitions that have been generated by other combinators like 'withSource', -- 'withGenerator' and 'withSink'. ----- [@dpDefinition ~ 'Source' ('Channel' ..) ':>>' 'Generator' ('Channel' ..) ':>>' 'Sink'@]: /DP/ Type level Flow Definition+-- [@dpDefinition ~ 'Source' ('Channel' ..) ':=>' 'Generator' ('Channel' ..) ':=>' 'Sink'@]: /DP/ Type level Flow Definition -- -- [@filterState@]: State of the 'StateT' 'Monad' that is the local State of the Filter execution --@@ -229,7 +254,7 @@ -- in orther to know how to spawn a new 'Filter' if it is needed, and the 'Stage' of the Generator to allow the user to perform some computation -- in that case. ----- [@dpDefinition ~ 'Source' ('Channel' ..) ':>>' 'Generator' ('Channel' ..) ':>>' 'Sink'@]: /DP/ Type level Flow Definition+-- [@dpDefinition ~ 'Source' ('Channel' ..) ':=>' 'Generator' ('Channel' ..) ':=>' 'Sink'@]: /DP/ Type level Flow Definition -- -- [@filterState@]: State of the 'StateT' 'Monad' that is the local State of the Filter execution --@@ -249,7 +274,7 @@ --  -- * All the 'Filter' execution (a.k.a. @forM_ actors runStage@) executes in a 'StateT' 'Monad' to share an internal state among 'Actor's. ----- [@dpDefinition ~ 'Source' ('Channel' ..) ':>>' 'Generator' ('Channel' ..) ':>>' 'Sink'@]: /DP/ Type level Flow Definition+-- [@dpDefinition ~ 'Source' ('Channel' ..) ':=>' 'Generator' ('Channel' ..) ':=>' 'Sink'@]: /DP/ Type level Flow Definition -- -- [@filterState@]: State of the 'StateT' 'Monad' that is the local State of the Filter execution --@@ -264,7 +289,7 @@  -- | 'Actor' Is a particular 'Stage' computation inside a 'Filter'. ----- [@dpDefinition ~ 'Source' ('Channel' ..) ':>>' 'Generator' ('Channel' ..) ':>>' 'Sink'@]: /DP/ Type level Flow Definition+-- [@dpDefinition ~ 'Source' ('Channel' ..) ':=>' 'Generator' ('Channel' ..) ':=>' 'Sink'@]: /DP/ Type level Flow Definition -- -- [@filterState@]: State of the 'StateT' 'Monad' that is the local State of the Filter execution --@@ -343,7 +368,7 @@  -- | Combinator for Building a 'Source' Stage. It uses an Associated Type Class to deduce the Function Signature required to the user -- taken from /DP/ Type Level Flow Definition --- [@dpDefinition ~ 'Source' ('Channel' ..) ':>>' 'Generator' ('Channel' ..) ':>>' 'Sink'@]: /DP/ Type level Flow Definition+-- [@dpDefinition ~ 'Source' ('Channel' ..) ':=>' 'Generator' ('Channel' ..) ':=>' 'Sink'@]: /DP/ Type level Flow Definition -- -- [@st@]: Existential Scope of 'DP' 'Monad'. {-# INLINE withSource #-}@@ -354,7 +379,7 @@  -- | Combinator for Building a 'Generator' Stage. It uses an Associated Type Class to deduce the Function Signature required to the user -- taken from /DP/ Type Level Flow Definition --- [@dpDefinition ~ 'Source' ('Channel' ..) ':>>' 'Generator' ('Channel' ..) ':>>' 'Sink'@]: /DP/ Type level Flow Definition+-- [@dpDefinition ~ 'Source' ('Channel' ..) ':=>' 'Generator' ('Channel' ..) ':=>' 'Sink'@]: /DP/ Type level Flow Definition -- -- [@filter@]: 'Filter' template type --@@ -367,7 +392,7 @@  -- | Combinator for Building a 'Sink' Stage. It uses an Associated Type Class to deduce the Function Signature required to the user -- taken from /DP/ Type Level Flow Definition --- [@dpDefinition ~ 'Source' ('Channel' ..) ':>>' 'Generator' ('Channel' ..) ':>>' 'Sink'@]: /DP/ Type level Flow Definition+-- [@dpDefinition ~ 'Source' ('Channel' ..) ':=>' 'Generator' ('Channel' ..) ':=>' 'Sink'@]: /DP/ Type level Flow Definition -- -- [@st@]: Existential Scope of 'DP' 'Monad'. {-# INLINE withSink #-}@@ -385,31 +410,36 @@ mkDP' = DynamicPipeline @dpDefinition  -- Hiding DP Constraint for running DP-type DPConstraint dpDefinition filterState st filterParam filter iparams gparams oparams r2 r3 l1 r4 l2 t2 s1 t1 s2 t5 l3 l4 =-  ( MkChans dpDefinition-  , HChan dpDefinition ~ r3 t2-  , Filter dpDefinition filterState filterParam st ~ filter-  , CloseList l3-  , CloseList l4-  , CloseList (HAppendListR l1 l2)-  , iparams ~ WithSource dpDefinition (DP st)-  , gparams ~ WithGenerator dpDefinition filter (DP st)-  , oparams ~ WithSink dpDefinition (DP st)-  , ArityRev iparams (HLength (ExpandSourceToCh dpDefinition))-  , ArityFwd iparams (HLength (ExpandSourceToCh dpDefinition))-  , HCurry' (HLength (ExpandSourceToCh dpDefinition)) iparams l3 (DP st ())-  , ArityRev gparams (HLength (ExpandGenToCh dpDefinition filter))-  , ArityFwd gparams (HLength (ExpandGenToCh dpDefinition filter))-  , HCurry' (HLength (ExpandGenToCh dpDefinition filter)) gparams (filter ': HAppendListR l1 l2) (DP st ())-  , ArityRev oparams (HLength (ExpandSinkToCh dpDefinition))-  , ArityFwd oparams (HLength (ExpandSinkToCh dpDefinition))-  , HCurry' (HLength (ExpandSinkToCh dpDefinition)) oparams l4 (DP st ())-  , AllChans r2 r3 l1 r4 l2 t2 s1 t1 s2 t5 l3 l4)+type DPConstraint dpDefinition filterState st filterParam filter gparams slr slw glr glw silr silw iparams oparams ls lsi +  = ( MkChans dpDefinition+    , HChan dpDefinition ~ ChanRecord slr slw glr glw silr silw+    , HAppendList slr slw+    , HAppendList glr glw+    , HAppendList silr silw+    , Filter dpDefinition filterState filterParam st ~ filter+    , ls ~ HAppendListR slr slw+    , lsi ~ HAppendListR silr silw+    , CloseList ls+    , CloseList lsi+    , CloseList (HAppendListR glr glw)+    , iparams ~ WithSource dpDefinition (DP st)+    , gparams ~ WithGenerator dpDefinition filter (DP st)+    , oparams ~ WithSink dpDefinition (DP st)+    , ArityRev iparams (HLength (ExpandSourceToCh dpDefinition))+    , ArityFwd iparams (HLength (ExpandSourceToCh dpDefinition))+    , HCurry' (HLength (ExpandSourceToCh dpDefinition)) iparams ls (DP st ())+    , ArityRev gparams (HLength (ExpandGenToCh dpDefinition filter))+    , ArityFwd gparams (HLength (ExpandGenToCh dpDefinition filter))+    , HCurry' (HLength (ExpandGenToCh dpDefinition filter)) gparams (filter ': HAppendListR glr glw) (DP st ())+    , ArityRev oparams (HLength (ExpandSinkToCh dpDefinition))+    , ArityFwd oparams (HLength (ExpandSinkToCh dpDefinition))+    , HCurry' (HLength (ExpandSinkToCh dpDefinition)) oparams lsi (DP st ())+    )  {-# INLINE buildDPProg #-}-buildDPProg :: forall dpDefinition filterState st filterParam filter iparams gparams oparams r2 r3 l1 r4 l2 t2 s1 t1 s2 t5 l3 l4.-               DPConstraint dpDefinition filterState st filterParam filter iparams gparams oparams r2 r3 l1 r4 l2 t2 s1 t1 s2 t5 l3 l4-            => DynamicPipeline dpDefinition filterState filterParam st -> DP st ()+buildDPProg :: forall dpDefinition filterState st filterParam filter gparams slr slw glr glw silr silw iparams oparams ls lsi.+            DPConstraint dpDefinition filterState st filterParam filter gparams slr slw glr glw silr silw iparams oparams ls lsi => +            DynamicPipeline dpDefinition filterState filterParam st -> DP st () buildDPProg DynamicPipeline{..} = do   (cIns, cGen, cOut) <- inGenOut <$> withDP (makeChans @dpDefinition)   let genWithFilter   = _gsFilterTemplate generator .*. cGen@@ -419,8 +449,8 @@  -- | Smart constructor for 'DynamicPipeline' Definition {-# INLINE mkDP #-}-mkDP :: forall dpDefinition filterState st filterParam filter iparams gparams oparams r2 r3 l1 r4 l2 t2 s1 t1 s2 t5 l3 l4.-        DPConstraint dpDefinition filterState st filterParam filter iparams gparams oparams r2 r3 l1 r4 l2 t2 s1 t1 s2 t5 l3 l4+mkDP :: forall dpDefinition filterState st filterParam filter gparams slr slw glr glw silr silw iparams oparams ls lsi.+        DPConstraint dpDefinition filterState st filterParam filter gparams slr slw glr glw silr silw iparams oparams ls lsi      => Stage (WithSource dpDefinition (DP st)) -- ^ 'Source' Stage generated by 'withSource' combinator      -> GeneratorStage dpDefinition filterState filterParam st  -- ^ 'Generator' Stage generated by 'withGenerator' combinator      -> Stage (WithSink dpDefinition (DP st))   -- ^ 'Sink' Stage generated by 'withSink' combinator@@ -454,15 +484,14 @@   close = const $ pure ()  -- | 'SpawnFilterConstraint' Constraint type alias-type SpawnFilterConstraint dpDefinition readElem st filterState filterParam l r t l1 b0 l2 l3 b2 b3 l4 =+type SpawnFilterConstraint dpDefinition readElem st filterState filterParam l l1 l2 l3 b2 b3 l4 =                 ( MkChans (ChansFilter dpDefinition)-                , FilterChans r (HList l3) t (HList (ReadChannel readElem : l1))                 , l1 ~ l                 , CloseList (ReadChannel readElem ': l4)                 , HAppendList l l3                 , l4 ~ HAppendListR l l3                 , l2 ~ (readElem ': ReadChannel readElem ': l4)-                , HChan (ChansFilter dpDefinition) ~ r t+                , HChan (ChansFilter dpDefinition) ~ InOutChan (ReadChannel readElem : l) l3                 , WithFilter dpDefinition filterParam (StateT filterState (DP st)) ~ (b2 -> ReadChannel b2 -> b3)                 , HLength (ExpandFilterToCh dpDefinition filterParam) ~ HLength l2                 , HCurry' (HLength l2) (WithFilter dpDefinition filterParam (StateT filterState (DP st))) l2 (StateT filterState (DP st) ())@@ -474,7 +503,7 @@ -- The user will have the capability to select how those filters are going to be spawned, for example on each read element, how to setup -- initial states of 'StateT' Monad on 'Actor' computations in filters, among others. ----- [@dpDefinition ~ 'Source' ('Channel' ..) ':>>' 'Generator' ('Channel' ..) ':>>' 'Sink'@]: /DP/ Type level Flow Definition+-- [@dpDefinition ~ 'Source' ('Channel' ..) ':=>' 'Generator' ('Channel' ..) ':=>' 'Sink'@]: /DP/ Type level Flow Definition -- -- [@readElem@]: Type of the element that is being read from the Selected Channel in the 'Generator' Stage --@@ -537,8 +566,8 @@  -- | Run 'UnFoldFilter' {-# INLINE unfoldF #-}-unfoldF :: forall dpDefinition readElem st filterState filterParam l r t l1 b0 l2 l3 b2 b3 l4.-           SpawnFilterConstraint dpDefinition readElem st filterState filterParam l r t l1 b0 l2 l3 b2 b3 l4+unfoldF :: forall dpDefinition readElem st filterState filterParam l l1 l2 l3 b2 b3 l4.+           SpawnFilterConstraint dpDefinition readElem st filterState filterParam l l1 l2 l3 b2 b3 l4         => UnFoldFilter dpDefinition readElem st filterState filterParam l -- ^ 'UnFoldFilter'         -> DP st (HList l) -- ^Return the list of 'ReadChannel's with the results to be read for the 'Generator' at the end. You can use this to pass the results to 'Sink' unfoldF = loopSpawn@@ -551,7 +580,7 @@       _ufOnElem elem'       if _ufSpawnIf elem'         then do-          (reads', writes' :: HList l3) <- getFilterChannels <$> DP (makeChans @(ChansFilter dpDefinition))+          (reads', writes' :: HList l3) <- getFilterChannels <$> DP (makeChansF @(ChansFilter dpDefinition))           let hlist = elem' .*. _ufReadChannel .*. (_ufRsChannels `hAppendList` writes')           void $ runFilter _ufFilter (_ufInitState elem') hlist (_ufReadChannel .*. (_ufRsChannels `hAppendList` writes'))           return $ uf { _ufReadChannel = hHead reads', _ufRsChannels = hTail reads' }