box 0.0.1.5 → 0.1.0

raw patch · 19 files changed

+382/−301 lines, 19 filesdep +mtldep −protoludePVP ok

version bump matches the API change (PVP)

Dependencies added: mtl

Dependencies removed: protolude

API changes (from Hackage documentation)

Files

• box.cabal +2/−4
• readme.md +41/−2
• src/Box.hs +31/−23
• src/Box/Box.hs +1/−3
• src/Box/Broadcast.hs +0/−2
• src/Box/Committer.hs +2/−4
• src/Box/Connectors.hs +27/−17
• src/Box/Control.hs +73/−77
• src/Box/Emitter.hs +4/−4
• src/Box/IO.hs +56/−49
• src/Box/Plugs.hs +0/−3
• src/Box/Queue.hs +112/−75
• src/Box/Stream.hs +2/−5
• src/Box/Time.hs +2/−5
• src/Box/Transducer.hs +3/−3
• src/Box/Updater.hs +8/−9
• stack.yaml +1/−1
• test/ctest.hs +17/−13
• test/test.hs +0/−2

@@ -1,5 +1,5 @@ name:           box-version:        0.0.1.5+version:        0.1.0 synopsis:       boxes description:    concurrent, effectful boxes category:       project@@ -53,7 +53,6 @@     , lens     , pipes     , profunctors-    , protolude     , streaming     , text     , time@@ -75,7 +74,7 @@     , dejafu     , generic-lens     , lens-    , protolude+    , mtl     , random     , streaming     , text@@ -93,5 +92,4 @@   build-depends:     base >=4.7 && <5     , doctest-    , protolude   default-language: Haskell2010

@@ -1,9 +1,48 @@ [box](https://tonyday567.github.io/box/index.html) [![Build Status](https://travis-ci.org/tonyday567/box.svg)](https://travis-ci.org/tonyday567/box) === -A concurrent, potentially effectful, profunctoril, composable box to put stuff in and take stuff out.+> The box is dark and full of terrors. -It's a pretty solid box.+Take two ubiquitous concepts. There are things that emit stuff. They instantiate a particular thing on demand, somehow and somewhere beyond our concern and frame of reference.  We ask for an `a` and an `a` is there, or not (the thing is a `Maybe a`). The library calls this an Emitter:++```+newtype Emitter m a = Emitter { emit :: m (Maybe a)}+```++And then there is the opposite.  There are things that commit stuff. We offer them a particular thing and they take it away beyond our concerns. It is committed to the void. We offer an `a` and it is committed, or not. The library calls this a Committer:++```+newtype Committer m a = Committer { commit :: a -> m Bool}+```++When you have both things, something that emits and something that commits, and you product them, the library calls this a Box:++```+data Box m c e = Box+  { committer :: Committer m c+  , emitter :: Emitter m e+  }+```++If you choose to see the committer as a conduit into the box, the emitter as a conduit out of the box, and remove concern about what is going on inside the box, then this frame of reference is often called a black box. Or you can choose to think from inside the box. From inside the box, there is again a conduit coming in that emits and a conduit going out that commits. You can take those things being emitted, turn them into something else and emit them.  You are the black box. ++Note how you can get confused with these metaphors.  The wire going into the box is a committer from a point of view outside the metaphorical box but looks like an emitter from the inside. The wire going out is a committer from the inside and an emitter from the outside.++The key to understanding this library is to resist having to choose a single frame of reference and, instead, focus on the types.++A `Box m c e` unifies the functorial wrapper `m` of the emitter and committer, which is what really makes it a functional box. `m` can be IO but is also often Software Transactional Memory (STM) which is the sanest environment for concurrent programming in all of the codingshpere. Boxes can often be hooked together at this level for useful efficiencies.++Boxes have a monoidal instance.  They can be mappended together to form a new box and complexity stays constant.++A Box is also a [profunctor](https://bartoszmilewski.com/2019/03/27/promonads-arrows-and-einstein-notation-for-profunctors/) and, to quote Bartos, a profunctor can be used to glue together two categories. A Box.Transducer, a stateful stream converter with a Category instance, can be used to connect up a box to create a sound compositional building block for arbitrarily complex problems.++Like other paradigms, boxes can be not fun places to get stuck in.  Much of the library are not, in fact, emitters and committers but emitter, committer and box continuations. If you gave me something that takes a box and does something, then I'll give you a something is a natural piece of logic within the library context, and so large parts of the functionality are [continuation-parsing style](https://ro-che.info/articles/2019-06-07-why-use-contt).++Finally, the interface between boxes is a natural place to create concurrency, and Box.Queue provides queues that are guaranteed to not race or block.+++> “Do not define me by my gender or my socio-economic status, Noah Willis. Do not tell me who I am and do not tell me who society thinks I am and then put me in that box and expect me to stay there. Because, I swear to God, I will climb the hell out of that box and I will take that box you've just put me in and I will use that box to smash your face in until you're nothing more than a freckly, bloodied pulp. You got that, sweet cheeks?” ~ Megan Jacobson, Yellow+  recipe ---

@@ -1,7 +1,6 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE NoImplicitPrelude #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeFamilies #-}@@ -18,6 +17,7 @@     -- $emit     -- $transduce     module Box.Box+  , module Box.Broadcast   , module Box.Committer   , module Box.Connectors   , module Box.Cont@@ -28,9 +28,11 @@   , module Box.Stream   , module Box.Time   , module Box.Transducer+  , (&)   ) where  import Box.Box+import Box.Broadcast import Box.Committer import Box.Connectors import Box.Cont@@ -41,14 +43,18 @@ import Box.Stream import Box.Time import Box.Transducer+import Control.Lens ((&))  -- $setup -- >>> :set -XOverloadedStrings -- >>> :set -XGADTs--- >>> import Protolude+-- >>> import Data.Functor.Contravariant -- >>> import Box -- >>> import qualified Streaming.Prelude as S -- >>> import Control.Monad.Conc.Class as C+-- >>> import Data.Text (Text)+-- >>> import qualified Data.Text as Text+-- >>> import Control.Applicative -- >>> let committer' = cStdout 100 -- >>> let emitter' = toEmit (S.each ["hi","bye","q","x"]) -- >>> let box' = Box <$> committer' <*> emitter'@@ -61,20 +67,19 @@ -- something -- True ----- The monoid instance sends each commit to both mappended committers. Delaying effects are introduced to these examples to keep stdout clear of race effects.+-- The monoid instance sends each commit to both mappended committers. Because everything is concurrent, race effects are common on stdout, so we introduce some delaying effects to (hopefully) avoid races. ----- turned off in doctest until the muddled upedness is sorted...--- > let cDelay = cmap (\b -> sleep 0.1 >> pure (Just b)) <$> (liftC <$> cStdout 100)--- > let cImmediate = liftC <$> cStdout 100--- > (etcM () transducer' $ (Box <$> (cImmediate <> cDelay) <*> (liftE <$> emitter'))) >> sleep 1--- echo: hi--- echo: hi--- echo: bye--- echo: bye+-- >>> let cFast = cmap (\b -> sleep 0.01 >> pure (Just b)) . liftC . contramap ("fast: " <>) <$> (cStdout 100)+-- >>> let cSlow = cmap (\b -> sleep 0.1 >> pure (Just b)) . liftC . contramap ("slow: " <>) <$> (cStdout 100)+-- >>> (etcM () transducer' $ (Box <$> (cFast <> cSlow) <*> (liftE <$> emitter'))) >> sleep 1+-- fast: echo: hi+-- slow: echo: hi+-- fast: echo: bye+-- slow: echo: bye -- -- >>> let c = fmap liftC $ cStdout 10 -- >>> let e = fmap liftE $ toEmit (S.each ["hi","bye","q","x"])--- >>> let c' = cmap (\a -> if a=="q" then (sleep 1 >> putStrLn "stolen!" >> sleep 1 >> pure (Nothing)) else (pure (Just a))) <$> c :: Cont IO (Committer IO Text)+-- >>> let c' = cmap (\a -> if a=="q" then (sleep 0.1 >> putStrLn "stolen!" >> sleep 0.1 >> pure (Nothing)) else (pure (Just a))) <$> c :: Cont IO (Committer IO Text) -- >>> fuse (pure . pure) $ Box <$> c' <*> e -- hi -- bye@@ -92,13 +97,15 @@ -- q_right -- x_right --- -- | splitCommit--- FIXME:--- > cs <- splitCommit (cStdout 100)--- > let c2 = contCommit <$> cs--- > (etcM () transducer' $ (Box <$> c2 <*> (liftE <$> emitter'))) >> sleep 1---+-- Splits a committer into two.+-- >>> let cs = splitCommit (liftC <$> cStdout 100)+-- >>> let cc = contCommit <$> cs <*> pure (cmap (\b -> sleep 0.01 >> pure (Just b)) . contramap ("cont: " <>))+-- >>> etcM () transducer' $ (Box <$> cc <*> (liftE <$> emitter'))+-- echo: hi+-- echo: bye+-- cont: echo: hi+-- cont: echo: bye  -- $emit --@@ -114,10 +121,10 @@ -- stole a q! -- x ----- >>> let e1 = fmap show <$> (toEmit $ delayTimed (S.each (zip (fromIntegral <$> [1..10]) ['a'..]))) :: Cont IO (Emitter (C.STM IO) Text)--- >>> let e2 = fmap show <$> (toEmit $ delayTimed (S.each (zip ((\x -> fromIntegral x + 0.1) <$> [1..10]) (reverse ['a'..'z'])))) :: Cont IO (Emitter (C.STM IO) Text)+-- >>> let e1 = fmap (Text.pack . show) <$> (toEmit $ delayTimed (S.each (zip ((0.2*) . fromIntegral <$> [1..10]) ['a'..]))) :: Cont IO (Emitter (C.STM IO) Text)+-- >>> let e2 = fmap (Text.pack . show) <$> (toEmit $ delayTimed (S.each (zip ((0.1+) . (0.2*) . fromIntegral <$> [1..10]) (reverse ['a'..'z'])))) :: Cont IO (Emitter (C.STM IO) Text) -- >>> let e12 = e1 <> e2--- >>> etc () (Transducer identity) $ Box <$> cStdout 6 <*> emerge ((,) <$> e1 <*> e2)+-- >>> etc () (Transducer id) $ Box <$> cStdout 6 <*> emerge ((,) <$> e1 <*> e2) -- 'a' -- 'z' -- 'b'@@ -125,7 +132,7 @@ -- 'c' -- 'x' ----- >>> etc () (Transducer identity) $ Box <$> cStdout 6 <*> (liftA2 (<>) e1 e2)+-- >>> etc () (Transducer id) $ Box <$> cStdout 6 <*> (liftA2 (<>) e1 e2) -- 'a' -- 'z' -- 'b'@@ -143,7 +150,8 @@  -- | broadcasting ----- > (bcast, bcom) <- C.atomically broadcast+-- >>> (bcast, bcom) <- (C.atomically broadcast) :: IO (Broadcaster (C.STM IO) Text, Committer (C.STM IO) Text)+-- -- > (funn, fem) <- C.atomically funnel -- > 

@@ -1,7 +1,6 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE NoImplicitPrelude #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeFamilies #-}@@ -17,11 +16,10 @@   ) where  import Control.Lens hiding ((:>), (.>), (<|), (|>))-import Data.Semigroup hiding (First, getFirst) import Box.Committer import Box.Emitter import Control.Monad.Conc.Class-import Protolude hiding ((.), (<>), STM)+import Control.Applicative  -- | A Box is a product of a Committer m and an Emitter. Think of a box with an incoming wire and an outgoing wire. Now notice that the abstraction is reversable: are you looking at two wires from "inside a box"; a blind erlang grunt communicating with the outside world via the two thin wires, or are you looking from "outside the box"; interacting with a black box object. Either way, it's a box. -- And either way, the committer is contravariant and the emitter covariant so it forms a profunctor.

@@ -1,4 +1,3 @@-{-# LANGUAGE NoImplicitPrelude #-} {-# OPTIONS_GHC -Wall #-}  module Box.Broadcast@@ -14,7 +13,6 @@ import Box.Cont import Box.Emitter import Box.Queue-import Protolude hiding (STM, atomically, (.), (<>)) import Control.Concurrent.Classy.STM as C import Control.Monad.Conc.Class as C 

@@ -2,7 +2,6 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE NoImplicitPrelude #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeFamilies #-}@@ -16,13 +15,12 @@   , handles   ) where -import Control.Category import Control.Lens hiding ((:>), (.>), (<|), (|>)) import Data.Functor.Constant import Data.Functor.Contravariant.Divisible-import Data.Semigroup hiding (First, getFirst)-import Protolude hiding ((.), (<>), STM, atomically) import Control.Monad.Conc.Class as C+import Data.Void (absurd)+import Data.Monoid (First(..))  -- | a Committer a "commits" values of type a. A Sink and a Consumer are some other metaphors for this. --

@@ -1,8 +1,6 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE NoImplicitPrelude #-}-{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeFamilies #-}@@ -20,23 +18,22 @@   , fuseEmit   , fuseEmitM   , fuseCommit+  , fuseCommitM   , emerge   , emergeM   , splitCommit+  , splitCommitSTM   , contCommit   ) where -import Control.Category-import Control.Lens hiding ((:>), (.>), (<|), (|>))-import Data.Semigroup hiding (First, getFirst) import Box.Box import Box.Queue import Box.Committer import Box.Cont import Box.Emitter-import Protolude hiding (STM, (.), (<>)) import Control.Monad.Conc.Class as C import Control.Concurrent.Classy.Async as C+import Control.Monad  -- * primitives -- | fuse an emitter directly to a committer@@ -87,6 +84,10 @@ fuseCommit :: (MonadConc m) => Committer (STM m) a -> Cont m (Committer (STM m) a) fuseCommit c = Cont $ \caction -> queueC caction (`fuseSTM_` c) +-- | fuse a committer to a buffer+fuseCommitM :: (MonadConc m) => Committer m a -> Cont m (Committer m a)+fuseCommitM c = Cont $ \caction -> queueCM caction (`fuse_` c)+ -- | fuse an emitter to a buffer fuseEmit :: (MonadConc m) => Emitter (STM m) a -> Cont m (Emitter (STM m) a) fuseEmit e = Cont $ \eaction -> queueE (fuseSTM_ e) eaction@@ -126,27 +127,36 @@         (queueEM (fuse_ (fst e')) eaction)         (queueEM (fuse_ (snd e')) eaction) --- | merge two committers------ not working+-- | split a committer (STM m) ---splitCommit :: (MonadConc m) =>+splitCommitSTM :: (MonadConc m) =>      Cont m (Committer (STM m) a)   -> Cont m (Either (Committer (STM m) a) (Committer (STM m) a))-splitCommit c =+splitCommitSTM c =   Cont $ \kk ->     with c $ \c' ->-      fst <$>-      C.concurrently+      concurrentlyLeft         (queueC (kk . Left) (`fuseSTM_` c'))         (queueC (kk . Right) (`fuseSTM_` c')) --- | a failed attempt to understand the either continuation style-contCommit :: Either (Committer m Text) (Committer m Text) -> Committer m Text-contCommit ec =+-- | split a committer+--+splitCommit :: (MonadConc m) =>+     Cont m (Committer m a)+  -> Cont m (Either (Committer m a) (Committer m a))+splitCommit c =+  Cont $ \kk ->+    with c $ \c' ->+      concurrentlyLeft+        (queueCM (kk . Left) (`fuse_` c'))+        (queueCM (kk . Right) (`fuse_` c'))++-- | use a split committer+contCommit :: Either (Committer m a) (Committer m b) -> (Committer m a -> Committer m b) -> Committer m b+contCommit ec f =   Committer $ \a ->     case ec of-      Left lc -> commit (contramap ("left " <>) lc) a+      Left lc -> commit (f lc) a       Right rc -> commit rc a  -- | a box modifier that feeds commits back to the emitter

@@ -1,56 +1,63 @@-{-# LANGUAGE NoImplicitPrelude #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MonoLocalBinds #-}+{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RankNTypes #-} {-# OPTIONS_GHC -Wall #-} {-# OPTIONS_GHC -Wno-redundant-constraints #-}  module Box.Control-  ( ControlComm(..)-  , ControlBox-  , ControlConfig(..)-  , defaultControlConfig-  , consoleControlBox-  , parseControlComms-  , controlBox-  , runControlBox-  , testBox-  , timeOut-  ) where+  ( ControlRequest (..),+    ControlResponse (..),+    ControlBox,+    ControlConfig (..),+    defaultControlConfig,+    consoleControlBox,+    parseControlRequest,+    controlBox,+    -- runControlBox,+    testBox,+    timeOut,+  )+where  import Box import Control.Applicative-import Control.Category import Control.Concurrent.Async import Control.Lens hiding ((|>)) import Control.Monad-import Data.Data-import GHC.Generics-import Protolude hiding ((.), STM)-import Text.Read (readMaybe)+import Control.Monad.Conc.Class as C import qualified Data.Attoparsec.Text as A+import Data.Data import qualified Data.Text as Text+import qualified Data.Text.IO as Text+import Data.Text (Text)+import GHC.Generics import qualified Streaming.Prelude as S-import Control.Monad.Conc.Class as C+import Text.Read (readMaybe)+import Data.Functor+import Control.Monad.Trans.Class+import Data.Bool+import Data.Maybe -data ControlComm-  = Ready -- ready for comms-  | Check -- check for existence-  | Died -- died (of its own accord)+data ControlRequest+  = Check -- check for existence   | Stop -- stop (without shutting down)-  | Kill -- stop and quit (& cancel thread)-  | ShutDown -- successfully Killed+  | Quit -- stop, quit & cancel thread   | Start -- start (if not yet started)   | Reset -- stop and start (potentially cancelling a previous instance)+  | Kill -- immediately exit+  deriving (Show, Read, Eq, Data, Typeable, Generic)++data ControlResponse+  = ShutDown -- action died   | On Bool -- are we live?   | Log Text   deriving (Show, Read, Eq, Data, Typeable, Generic) -type ControlBox m = (MonadConc m) => Cont m (Box (STM m) ControlComm ControlComm)+type ControlBox m = (MonadConc m) => Cont m (Box (STM m) ControlResponse ControlRequest)  data ControlConfig   = KeepAlive Double@@ -62,31 +69,39 @@  consoleControlBox :: ControlBox IO consoleControlBox =-  Box <$>-  (contramap show <$>-   (cStdout 1000 :: Cont IO (Committer (STM IO) Text))) <*>-  (emap (pure . either (const Nothing) Just) <$>-   (eParse parseControlComms <$>-    eStdin 1000))+  Box+    <$> ( contramap (Text.pack . show)+            <$> (cStdout 1000 :: Cont IO (Committer (STM IO) Text))+        )+    <*> ( emap (pure . either (const Nothing) Just)+            <$> ( eParse parseControlRequest+                    <$> eStdin 1000+                )+        ) -parseControlComms :: A.Parser ControlComm-parseControlComms =-  A.string "q" $> Stop <|> A.string "s" $> Start <|>-  A.string "x" $> Kill <|> do-    res <- readMaybe . Text.unpack <$> A.takeText-    case res of-      Nothing -> mzero-      Just a -> return a+parseControlRequest :: A.Parser ControlRequest+parseControlRequest =+  A.string "q" $> Stop+    <|> A.string "s" $> Start+    <|> A.string "x" $> Quit+    <|> A.string "c" $> Check+    <|> A.string "r" $> Reset+    <|> do+      res <- readMaybe . Text.unpack <$> A.takeText+      case res of+        Nothing -> mzero+        Just a -> return a  -- | an effect that can be started and stopped--- committer is an existence test--- controlBox :: (MonadConc m) => ControlConfig -> m () -> ControlBox m-controlBox-  :: ControlConfig-     -> IO a -> Box (STM IO) ControlComm ControlComm -> IO Bool-controlBox cfg app (Box c e) = do+controlBox ::+  IO a ->+  Box (STM IO) ControlResponse ControlRequest ->+  IO ()+controlBox app (Box c e) = do   ref' <- C.newIORef Nothing-  go ref'+  _ <- C.atomically (commit c (On False))+  _ <- go ref'+  Text.putStrLn ("after go ref'" :: Text)   where     go ref = do       msg <- C.atomically $ emit e@@ -103,49 +118,30 @@               a <- C.readIORef ref               when (isNothing a) (void $ start ref c)               go ref-            Stop -> cancel' ref >> go ref-            Kill -> cancel' ref >> C.atomically (commit c ShutDown)-            Died ->-              case cfg of-                AllowDeath -> C.atomically $ commit c ShutDown-                KeepAlive x -> do-                  sleep x-                  _ <- C.atomically $ commit c Start-                  go ref+            Stop -> cancel' ref c >> go ref+            Quit -> cancel' ref c >> C.atomically (commit c ShutDown)             Reset -> do               a <- C.readIORef ref-              unless (isNothing a) (cancel' ref)+              unless (isNothing a) (void $ cancel' ref c)               _ <- start ref c               go ref             _ -> go ref     start ref c' = do-      a' <- async (app >> C.atomically (commit c' Died))+      a' <- async (app >> C.atomically (commit c' (On False)))       C.writeIORef ref (Just a')-      C.atomically $ commit c' Ready-    cancel' ref = do+      C.atomically $ commit c' (On True)+    cancel' ref c' = do       mapM_ cancel =<< C.readIORef ref       C.writeIORef ref Nothing--runControlBox :: ControlConfig -> IO () -> IO ()-runControlBox cfg action =-  etc-    ()-    (Transducer $ \s -> s & S.takeWhile (/= ShutDown))-    (boxForgetPlug (void <$> controlBox cfg action))+      C.atomically $ commit c' (On False)  -- | send Start, wait for a Ready signal, run action, wait x secs, then send Quit-testBox :: IO Bool+testBox :: IO () testBox = cb   where     action =-      sequence_ $-      (\x -> putStrLn x >> sleep 1) . (show :: Integer -> Text) <$>-      reverse [0 .. 10]-    cb = with consoleControlBox (controlBox (KeepAlive 3) action)-  -- buff (bounded 1)-  -- ControlStart-  -- ControlReady-  -- ControlQuit+      replicateM_ 3 (sleep 1 >> Text.putStrLn ("beep" :: Text))+    cb = with consoleControlBox (controlBox action)  timeOut :: Double -> ControlBox m timeOut t =

@@ -2,7 +2,6 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE NoImplicitPrelude #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeFamilies #-}@@ -19,13 +18,14 @@   , eParse   ) where -import Control.Category ((.)) import Data.Functor.Constant-import Data.Semigroup hiding (First, getFirst)-import Protolude hiding ((.), (<>), STM, atomically) import qualified Data.Attoparsec.Text as A import qualified Data.Text as Text+import Data.Text (Text) import Control.Monad.Conc.Class as C+import Control.Applicative+import Control.Monad+import Data.Monoid  -- | an `Emitter` "emits" values of type a. A Source & a Producer (of 'a's) are the two other alternative but overloaded metaphors out there. --

@@ -1,7 +1,6 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE NoImplicitPrelude #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeFamilies #-}@@ -10,32 +9,28 @@  -- | IO actions module Box.IO-  ( cStdin_-  , cStdin-  , cStdin'-  , eStdin-  , readStdin-  , eStdout_-  , eStdout-  , eStdoutM-  , eStdout'-  , cStdout-  , showStdout-  , consolePlug-  , emitLines-  , commitLines-  , cCRef-  , cCRefM-  , toListM-  , getCommissions-  , getEmissions-  ) where+  ( cStdin_,+    cStdin,+    cStdin',+    eStdin,+    readStdin,+    eStdout_,+    eStdout,+    eStdoutM,+    eStdout',+    cStdout,+    showStdout,+    consolePlug,+    emitLines,+    commitLines,+    cCRef,+    cCRefM,+    toListM,+    getCommissions,+    getEmissions,+  )+where -import Control.Category-import qualified Control.Foldl as L-import Control.Lens hiding ((:>), (.>), (<|), (|>))-import Data.Semigroup hiding (First, getFirst)-import qualified Data.Text.IO as Text import Box.Box import Box.Committer import Box.Cont@@ -43,20 +38,28 @@ import Box.Plugs import Box.Stream import Box.Transducer-import Protolude hiding ((.), (<>), STM)-import Streaming (Of(..), Stream)-import qualified Streaming.Internal as S-import qualified Streaming.Prelude as S-import qualified Control.Monad.Conc.Class as C import qualified Control.Concurrent.Classy.IORef as C+import qualified Control.Foldl as L+import Control.Lens hiding ((.>), (:>), (<|), (|>))+import Control.Monad+import qualified Control.Monad.Conc.Class as C import Control.Monad.Conc.Class (STM)+import Control.Monad.IO.Class+import Data.Text (Text)+import qualified Data.Text as Text+import qualified Data.Text.IO as Text+import Streaming (Of (..), Stream)+import qualified Streaming.Internal as S+import qualified Streaming.Prelude as S+import System.IO  -- * console+ -- | a single stdin committer action cStdin_ :: Committer (STM IO) Text -> IO () cStdin_ c = do-  a <- getLine-  void $ atomically $ commit c a+  a <- Text.getLine+  void $ C.atomically $ commit c a  -- | a finite stdin committer action cStdin :: Int -> Committer (STM IO) Text -> IO ()@@ -75,40 +78,40 @@ readStdin = emap (pure . either (const Nothing) Just) . eRead <$> eStdin 1000  -- | a single stdout emitter action-eStdout_ :: (Print a) => Emitter (STM IO) a -> IO ()+eStdout_ :: Emitter (STM IO) Text -> IO () eStdout_ e = do-  a <- atomically $ emit e+  a <- C.atomically $ emit e   case a of     Nothing -> pure ()-    Just a' -> putStrLn a'+    Just a' -> Text.putStrLn a'  -- | a single stdout emitter action-eStdoutM_ :: (Print a) => Emitter IO a -> IO ()+eStdoutM_ :: Emitter IO Text -> IO () eStdoutM_ e = do   a <- emit e   case a of     Nothing -> pure ()-    Just a' -> putStrLn a'+    Just a' -> Text.putStrLn a'  -- | a finite stdout emitter action-eStdout :: (Print a) => Int -> Emitter (STM IO) a -> IO ()+eStdout :: Int -> Emitter (STM IO) Text -> IO () eStdout n = replicateM_ n . eStdout_  -- | a finite stdout emitter action-eStdoutM :: (Print a) => Int -> Emitter IO a -> IO ()+eStdoutM :: Int -> Emitter IO Text -> IO () eStdoutM n = replicateM_ n . eStdoutM_  -- | a forever stdout emitter action-eStdout' :: (Print a) => Emitter (STM IO) a -> IO ()+eStdout' :: Emitter (STM IO) Text -> IO () eStdout' = forever . eStdout_  -- | a Cont stdout committer-cStdout :: Print a => Int -> Cont IO (Committer (STM IO) a)+cStdout :: Int -> Cont IO (Committer (STM IO) Text) cStdout n = eStdout n & commitPlug  -- | show to stdout showStdout :: Show a => Cont IO (Committer (STM IO) a)-showStdout = contramap show <$> (cStdout 1000 :: Cont IO (Committer (STM IO) Text))+showStdout = contramap (Text.pack . show) <$> (cStdout 1000 :: Cont IO (Committer (STM IO) Text))  -- | console box -- > etc () (Trans $ \s -> s & S.takeWhile (/="q") & S.map ("echo: " <>)) (console 5)@@ -116,9 +119,10 @@ consolePlug n = boxPlug (eStdout n) (cStdin n)  -- * file operations+ -- | emit lines from a file emitLines :: FilePath -> Cont IO (Emitter (STM IO) Text)-emitLines filePath = Cont (withFile filePath ReadMode) >>= (fromHandle >>> toEmit)+emitLines filePath = Cont (withFile filePath ReadMode) >>= (toEmit . fromHandle)   where     fromHandle :: Handle -> Stream (Of Text) IO ()     fromHandle h =@@ -129,7 +133,7 @@ -- | commit lines to a file commitLines :: FilePath -> Cont IO (Committer (STM IO) Text) commitLines filePath =-  Cont (withFile filePath WriteMode) >>= (toHandle >>> toCommit)+  Cont (withFile filePath WriteMode) >>= (toCommit . toHandle)   where     toHandle h = loop       where@@ -142,12 +146,14 @@               loop rest  -- * concurrent refs+ -- | commit to a list CRef cCRef :: (C.MonadConc m) => m (C.IORef m [b], Cont m (Committer (C.STM m) b), m [b]) cCRef = do   ref <- C.newIORef []-  let c = toCommitFold $-        L.FoldM (\x a -> C.modifyIORef x (a :) >> pure x) (pure ref) (const $ pure ())+  let c =+        toCommitFold $+          L.FoldM (\x a -> C.modifyIORef x (a :) >> pure x) (pure ref) (const $ pure ())   let res = reverse <$> C.readIORef ref   pure (ref, c, res) @@ -155,8 +161,9 @@ cCRefM :: (C.MonadConc m, Monoid a) => m (C.IORef m a, Cont m (Committer (C.STM m) a), m a) cCRefM = do   ref <- C.newIORef mempty-  let c = toCommitFold $-        L.FoldM (\x a -> C.modifyIORef x (a <>) >> pure x) (pure ref) (const $ pure ())+  let c =+        toCommitFold $+          L.FoldM (\x a -> C.modifyIORef x (a <>) >> pure x) (pure ref) (const $ pure ())   let res = C.readIORef ref   pure (ref, c, res) 

@@ -1,7 +1,6 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE NoImplicitPrelude #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeFamilies #-}@@ -19,14 +18,12 @@   , boxForgetPlug   ) where -import Control.Category import Box.Committer import Box.Cont import Box.Box import Box.Queue import Box.Emitter import GHC.Conc-import Protolude hiding ((.), (<>))  -- * plugs -- | hook an emitter action to a queue, creating a committer continuation

@@ -1,7 +1,6 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE NoImplicitPrelude #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeFamilies #-}@@ -19,22 +18,24 @@   , queueE   , queueCM   , queueEM-  , queueCLog-  , queueELog   , waitCancel-  , toBoxLog   , ends+  , withQ+  , withQE+  , withQC+  , toBox+  , concurrentlyLeft+  , concurrentlyRight   ) where  import Box.Box import Box.Committer import Box.Emitter--- import GHC.Conc-import Protolude hiding (STM, check, wait, cancel, atomically, withAsync, concurrently) import Control.Concurrent.Classy.STM as C import Control.Monad.Conc.Class as C import Control.Concurrent.Classy.Async as C import Control.Monad.Catch as C+import Control.Applicative  -- | 'Queue' specifies how messages are queued data Queue a@@ -98,29 +99,6 @@         (Emitter (readCheck sealed o)),         seal) --- | write to a queue, checking the seal-writeCheckLog :: (Show a, MonadIO m, MonadConc m) => TVar (STM m) Bool -> (a -> (STM m) ()) -> a -> m Bool-writeCheckLog sealed i a = do-  b <- atomically $ readTVar sealed-  if b-    then do-      putStrLn ("writeCheckLog sealed" :: Text)-      pure False-    else do-      putStrLn $ "writeCheckLog writing: " <> (show a:: Text)-      atomically $ i a-      pure True---- | read from a queue, and retry if not sealed-readCheckLog :: (Show a, MonadIO m, MonadConc m) => TVar (STM m) Bool -> (STM m) a -> m (Maybe a)-readCheckLog sealed o = do-  o' <- atomically o-  putStrLn $ "readCheckLog reading" <> (show o' :: Text)-  atomically (Just <$> o <|> do-                            b <- readTVar sealed-                            C.check b-                            pure Nothing)- toBoxM :: (MonadConc m) =>   Queue a -> m (Box m a a, m ()) toBoxM q = do@@ -132,19 +110,6 @@         (Emitter (atomically $ readCheck sealed o)),         seal) -toBoxLog :: (Show a, MonadIO m, MonadConc m) =>-  Queue a -> m (Box m a a, m ())-toBoxLog q = do-  (i, o) <- atomically $ ends q-  sealed <- atomically $ newTVarN "sealed" False-  initSeal <- atomically $ readTVar sealed-  putStrLn $ "toBoxLog: seal: " <> (show initSeal :: Text)-  let seal = atomically $ writeTVar sealed True-  pure (Box-        (Committer (writeCheckLog sealed i))-        (Emitter (readCheckLog sealed o)),-        seal)- -- | wait for the first action, and then cancel the second waitCancel :: (MonadConc m) => m b -> m a -> m b waitCancel a b =@@ -154,6 +119,20 @@       cancel b'       pure a'' +-- | run two actions concurrently, but wait and return on the left result.+concurrentlyLeft :: MonadConc m => m a -> m b -> m a+concurrentlyLeft left right =+  withAsync left $ \a ->+  withAsync right $ \_ ->+  wait a++-- | run two actions concurrently, but wait and return on the right result.+concurrentlyRight :: MonadConc m => m a -> m b -> m b+concurrentlyRight left right =+  withAsync left $ \_ ->+  withAsync right $ \b ->+  wait b+ -- | connect a committer and emitter action via spawning a queue, and wait for both to complete. withQ :: (MonadConc m) =>      Queue a@@ -170,37 +149,53 @@          (cio (committer box) `C.finally` atomically seal)          (eio (emitter box) `C.finally` atomically seal)) --- | connect a committer and emitter action via spawning a queue, and wait for both to complete.-withQM :: (MonadConc m) =>+-- | connect a committer and emitter action via spawning a queue, and wait for committer to complete.+withQC :: (MonadConc m) =>      Queue a-  -> (Queue a -> m (Box m a a, m ()))-  -> (Committer m a -> m l)-  -> (Emitter m a -> m r)-  -> m (l, r)-withQM q spawner cio eio =+  -> (Queue a -> (STM m) (Box (STM m) a a, (STM m) ()))+  -> (Committer (STM m) a -> m l)+  -> (Emitter (STM m) a -> m r)+  -> m l+withQC q spawner cio eio =   C.bracket-    (spawner q)-    snd+    (atomically $ spawner q)+    (\(_, seal) -> atomically seal)     (\(box, seal) ->-       concurrently-         (cio (committer box) `C.finally` seal)-         (eio (emitter box) `C.finally` seal))+       concurrentlyLeft+         (cio (committer box) `C.finally` atomically seal)+         (eio (emitter box) `C.finally` atomically seal)) +-- | connect a committer and emitter action via spawning a queue, and wait for emitter to complete.+withQE :: (MonadConc m) =>+     Queue a+  -> (Queue a -> (STM m) (Box (STM m) a a, (STM m) ()))+  -> (Committer (STM m) a -> m l)+  -> (Emitter (STM m) a -> m r)+  -> m r+withQE q spawner cio eio =+  C.bracket+    (atomically $ spawner q)+    (\(_, seal) -> atomically seal)+    (\(box, seal) ->+       concurrentlyRight+         (cio (committer box) `C.finally` atomically seal)+         (eio (emitter box) `C.finally` atomically seal))+ -- | connect a committer and emitter action via spawning a queue, and wait for both to complete.-withQMLog :: (MonadConc m, MonadIO m) =>+withQM :: (MonadConc m) =>      Queue a   -> (Queue a -> m (Box m a a, m ()))   -> (Committer m a -> m l)   -> (Emitter m a -> m r)   -> m (l, r)-withQMLog q spawner cio eio =+withQM q spawner cio eio =   C.bracket     (spawner q)     snd     (\(box, seal) ->        concurrently-         (cio (committer box) `C.finally` (putStrLn ("committer sealed" :: Text) >> seal))-         (eio (emitter box) `C.finally` (putStrLn ("emitter sealed" :: Text) >> seal)))+         (cio (committer box) `C.finally` seal)+         (eio (emitter box) `C.finally` seal))  -- | create an unbounded queue queue ::@@ -226,7 +221,6 @@   m l queueC cm em = fst <$> withQ Unbounded toBox cm em - -- | create an unbounded queue, returning the emitter result queueCM ::   (MonadConc m) =>@@ -243,20 +237,63 @@   m r queueEM cm em = snd <$> withQM Unbounded toBoxM cm em --- | create an unbounded queue, returning the emitter result-queueELog ::-  (Show a, MonadConc m, MonadIO m) =>-  (Committer m a -> m l) ->-  (Emitter m a -> m r) ->-  m r-queueELog cm em = snd <$> withQMLog Unbounded toBoxLog-  (\c -> putStrLn ("cm acted" :: Text) >> cm c)-  (\c -> putStrLn ("em acted" :: Text) >> em c) --- | create an unbounded queue, returning the committer result-queueCLog ::-  (Show a, MonadConc m, MonadIO m) =>-  (Committer m a -> m l) ->-  (Emitter m a -> m r) ->-  m l-queueCLog cm em = fst <$> withQM Unbounded toBoxLog cm em+{- |++The one-in-the-chamber problem++This is the referential transparency refactoring I did to solve the one-in-the-chamber problem.  An etc process wasn't closing down when it should, until the committer fired once more:++-- etc () (Transducer $ \s -> s & S.takeWhile (/="q")) (Box <$> cStdout 2 <*> eStdin 2)++On entering a 'q' in stdin, this code piece requires another input from stdin before it shuts down.++-}++-- > etc () (Transducer $ \s -> s & S.takeWhile (/="q")) (Box <$> cStdout 2 <*> eStdin 2)+-- etc substitution+-- > with (Box <$> cStdout 2 <*> eStdin 2) $ \(Box c e) -> (e & toStream & transduce (Transducer $ \s -> s & S.takeWhile (/="q")) & fromStream) c & flip execStateT ()+-- no state & transduction unwrapping+-- > with (Box <$> cStdout 2 <*> eStdin 2) $ \(Box c e) -> (e & toStream & S.takeWhile (/="q") & fromStream) c+-- subbing the IO's+-- > with (Box <$> (eStdout 2 & commitPlug) <*> (cStdin 2 & emitPlug)) $ \(Box c e) -> (e & toStream & transduce (Transducer $ \s -> s & S.takeWhile (/="q")) & fromStream) c+-- unplugging+-- > with (Box <$> (Cont $ \cio -> queueC cio (eStdout 2)) <*> (Cont $ \eio -> queueE (cStdin 2) eio)) $ \(Box c e) -> (e & toStream & S.takeWhile (/="q") & fromStream) c+-- fmapping the Box+-- > Cont (\r_ -> (Cont $ \cio -> queueC cio (eStdout 2)) `with` \x -> r_ (Box x))+-- twisting the with simplifying the Cont+-- > Cont (\r_ -> queueC (r_ . Box) (eStdout 2))+-- spaceship time!+-- > Cont (\r_ -> (Cont (\e -> queueC (e . Box) (eStdout 2))) `with` \f -> (Cont $ \eio -> queueE (cStdin 2) eio) `with` \x -> r_ (f x))+-- flipping the withs+-- > Cont (\r_ -> with (Cont (\e -> queueC (e . Box) (eStdout 2))) (\f -> with (Cont $ \eio -> queueE (cStdin 2) eio) (r_ . f)))+-- swallowing the withs+-- > with Cont (\r_ -> queueC ((\f -> queueE (cStdin 2) (r_ . f)) . Box) (eStdout 2))++-- subbing back in mainline+-- > with (Cont (\r_ -> queueC ((\f -> queueE (cStdin 2) (r_ . f)) . Box) (eStdout 2))) (\(Box c e) -> (e & toStream & S.takeWhile (/="q") & fromStream) c)+-- > queueC ((\f -> queueE (cStdin 2) ((\(Box c e) -> (e & toStream & S.takeWhile (/="q") & fromStream) c) . f)) . Box) (eStdout 2)+-- subbing queues+-- > fmap fst (withQ Unbounded toBox ((\f -> queueE (cStdin 2) ((\(Box c e) -> (e & toStream & S.takeWhile (/="q") & fromStream) c) . f)) . Box) (eStdout 2))+-- subbing stdin and stdout+-- > fmap fst (withQ Unbounded toBox ((\f -> fmap snd (withQ Unbounded toBox (\c -> cStdin_ c *> cStdin_ c) ((\(Box c e) -> (e & toStream & S.takeWhile (/="q") & fromStream) c) . f))) . Box) (\e -> eStdout_ e *> eStdout_ e))+-- removes the second eStdout_ (still requires another stdin input before it closes up)+-- fmap fst (withQ Unbounded toBox ((\f -> fmap snd (withQ Unbounded toBox (\c -> cStdin_ c *> cStdin_ c) ((\(Box c e) -> (e & toStream & S.takeWhile (/="q") & fromStream) c) . f))) . Box) eStdout_)+-- remove surperfluous fsts and snds+-- > withQ Unbounded toBox ((\f -> (withQ Unbounded toBox (\c -> cStdin_ c *> cStdin_ c) ((\(Box c e) -> (e & toStream & S.takeWhile (/="q") & fromStream) c) . f))) . Box) eStdout_+-- IO (((),()), ())+-- an intuitive unwrapping of the f+-- > withQ Unbounded toBox (\c -> (withQ Unbounded toBox (\c' -> cStdin_ c' *> cStdin_ c') ((\e -> (e & toStream & S.takeWhile (/="q") & fromStream) c)))) eStdout_++{-++And here was the problem is much easier to see. The withQ's were waiting on both sides of the queue.++I replaced `snd <$> withQ` with `withQE`++-}++-- subbing withQE fixes!+-- withQ Unbounded toBox (\c -> (withQE Unbounded toBox (\c' -> cStdin_ c' *> cStdin_ c') ((\e -> (fromStream . S.takeWhile (/="q") . toStream $ e) c)))) eStdout_++

@@ -1,7 +1,6 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE NoImplicitPrelude #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeFamilies #-}@@ -23,8 +22,6 @@   , fromStreamM   ) where -import Control.Category- import Box.Committer import Box.Cont import Box.Emitter@@ -33,7 +30,7 @@ import qualified Control.Foldl as L import qualified Streaming.Prelude as S import Control.Monad.Conc.Class as C-import Protolude hiding ((<>), (.), STM, check, wait, cancel, atomically, withAsync, concurrently)+import Control.Monad  -- * streaming -- | create a committer from a stream consumer@@ -63,7 +60,7 @@ -- todo: look at biases queueStream ::      (MonadConc m) => Stream (Of a) m () -> Cont m (Stream (Of a) m ())-queueStream i = Cont $ \o -> queueE (fromStream i) (toStream >>> o)+queueStream i = Cont $ \o -> queueE (fromStream i) (o . toStream)  -- | turn an emitter into a stream toStream :: (MonadConc m) => Emitter (STM m) a -> Stream (Of a) m ()

@@ -1,6 +1,5 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE NoImplicitPrelude #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# OPTIONS_GHC -fno-warn-name-shadowing #-}@@ -17,17 +16,15 @@   , emitStamp   ) where -import Control.Applicative-import Control.Monad-import Data.Maybe import Data.Time import Box.Cont import Box.Emitter import Box.Stream-import Protolude hiding (threadDelay, STM) import qualified Streaming.Prelude as S import qualified Streaming as S import Control.Monad.Conc.Class as C+import Control.Monad.IO.Class+import Control.Monad.Trans.Class  -- | sleep for x seconds sleep :: (MonadConc m) => Double -> m ()

@@ -1,7 +1,6 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE NoImplicitPrelude #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeFamilies #-}@@ -17,7 +16,8 @@   , asPipe   ) where -import Control.Category+import Prelude hiding ((.), id)+import Control.Category (Category(..)) import Control.Lens hiding ((:>), (.>), (<|), (|>)) import Control.Monad.Base (MonadBase, liftBase) import Box.Box@@ -27,10 +27,10 @@ import Box.Stream import qualified Pipes import qualified Pipes.Prelude as Pipes-import Protolude hiding ((.), (<>)) import Streaming (Of(..), Stream) import qualified Streaming.Prelude as S import Control.Monad.Conc.Class as C+import Control.Monad.Trans.State.Lazy  -- | transduction -- [wiki](https://en.wikipedia.org/wiki/Transducer) says: "A transducer is a device that converts energy from one form to another." Translated to context, this Transducer converts a stream of type a to a stream of a different type.

@@ -1,4 +1,3 @@-{-# LANGUAGE NoImplicitPrelude #-} {-# LANGUAGE ExistentialQuantification #-} {-# LANGUAGE RankNTypes #-} {-# OPTIONS_GHC -Wall #-}@@ -14,14 +13,14 @@ import Control.Applicative (Applicative((<*>), pure)) import Control.Foldl (Fold(..), FoldM(..)) import qualified Control.Foldl as Foldl-import Data.IORef (newIORef, readIORef, writeIORef) import Box-import Protolude+import Control.Monad.Conc.Class as C+import qualified GHC.Conc  -- | An updater of a value a, where the updating process consists of an IO fold over an emitter data Updater a =   forall e. Updater (FoldM IO e a)-                    (Cont IO (Emitter STM e))+                    (Cont IO (Emitter GHC.Conc.STM e))  instance Functor Updater where   fmap f (Updater fold' e) = Updater (fmap f fold') e@@ -56,7 +55,7 @@       eT = fmap (fmap Left) eL <> fmap (fmap Right) eR  -- | Create an `Updatable` value using a pure `Fold`-updater :: Fold e a -> Cont IO (Emitter STM e) -> Updater a+updater :: Fold e a -> Cont IO (Emitter GHC.Conc.STM e) -> Updater a updater fold' = Updater (Foldl.generalize fold')  -- | run an action on each update@@ -79,18 +78,18 @@       handler b       return x' -updates :: Updater a -> Cont IO (Emitter STM a)+updates :: Updater a -> Cont IO (Emitter GHC.Conc.STM a) updates (Updater (FoldM step begin done) e) = Cont $ \e' -> queueE cio e'   where     ioref c = do       x <- begin       a <- done x       _ <- atomically $ commit c a-      newIORef x+      C.newIORef x     cio c =       with e $ \e' -> do         ioref' <- ioref c-        x <- readIORef ioref'+        x <- C.readIORef ioref'         e'' <- atomically $ emit e'         case e'' of           Nothing -> pure ()@@ -98,4 +97,4 @@             x' <- step x e'''             a <- done x'             _ <- atomically $ commit c a-            writeIORef ioref' x'+            C.writeIORef ioref' x'

@@ -1,4 +1,4 @@-resolver: lts-14.11+resolver: lts-14.13  packages:   - .

@@ -1,7 +1,6 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE NoImplicitPrelude #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeFamilies #-}@@ -18,7 +17,7 @@ -- module Main where -import Control.Category (id)+import Prelude import Control.Monad.Conc.Class as C import Control.Concurrent.Classy.STM as C import Box.Box@@ -31,8 +30,10 @@ import Box.Queue import Box.Stream import Box.Transducer-import Data.String-import Protolude hiding (STM)+import Control.Monad.IO.Class+import Data.Text (Text)+import qualified Data.Text as Text+import qualified Data.Text.IO as Text import Test.DejaFu import Test.DejaFu.Types import qualified Streaming.Prelude as S@@ -42,6 +43,9 @@ import qualified Control.Monad.Trans.State as Trans import Data.Generics.Labels () import Data.Generics.Product+import GHC.Generics+import Control.Monad.State.Lazy+import Control.Applicative  -- | the test box is a pure list emitter into an IORef appending list testBox :: (MonadConc m) => Int -> m (Cont m (Box (STM m) Int Int), m [Int])@@ -157,7 +161,7 @@   fuse (pure . pure) (Box <$> (liftC <$> c1) <*> e1)   fuse (pure . pure) (Box <$> (liftC <$> c2) <*> e2)   eres <- readIORef eref-  putStrLn $ "eref: " <> (show eres :: Text)+  liftIO $ Text.putStrLn $ "eref: " <> Text.pack (show eres)   (,) <$> r1 <*> r2  -- | a broadcaster @@ -168,16 +172,16 @@ -- | create a (broadcaster, committer) broadcast' :: (Show a, MonadConc m, MonadIO m) => m (Broadcaster' m a, Committer m a) broadcast' = do-  ref <- C.atomically $ newTVar mempty+  ref <- C.atomically $ C.newTVar mempty   let com = Committer $ \a -> do-        putStrLn $ "broadcaster': " <> (show a :: Text)-        c <- C.atomically $ readTVar ref+        liftIO $ Text.putStrLn $ "broadcaster': " <> Text.pack (show a)+        c <- C.atomically $ C.readTVar ref         commit c a   pure (Broadcaster' ref, com)  -- | subscribe to a broadcaster-subscribe' :: (Show a, MonadIO m, MonadConc m) => Broadcaster' m a -> Cont m (Emitter m a)-subscribe' (Broadcaster' tvar) = Cont $ \e -> queueELog cio e+subscribe' :: (MonadIO m, MonadConc m) => Broadcaster' m a -> Cont m (Emitter m a)+subscribe' (Broadcaster' tvar) = Cont $ \e -> queueEM cio e   where     cio c = C.atomically $ modifyTVar' tvar (mappend c) @@ -188,9 +192,9 @@   where     go = do       a <- emit e-      putStrLn $ "fuse_' emit: " <> (show a :: Text)+      liftIO $ Text.putStrLn $ "fuse_' emit: " <> Text.pack (show a)       c' <- maybe (pure False) (commit c) a-      putStrLn $ "fuse_' commit: " <> (show c' :: Text)+      liftIO $ Text.putStrLn $ "fuse_' commit: " <> Text.pack (show c')       when c' go  fuse' :: (Show b, MonadIO m) => (a -> m (Maybe b)) -> Cont m (Box m b a) -> m ()@@ -252,7 +256,7 @@ counter n =   pure $     Emitter $ do-        a <- Protolude.get+        a <- Control.Monad.State.Lazy.get         case a < n of           False -> pure Nothing           True -> do

@@ -1,9 +1,7 @@-{-# LANGUAGE NoImplicitPrelude #-} {-# OPTIONS_GHC -Wall #-}  module Main where -import Protolude import Test.DocTest  main :: IO ()