box 0.4.0 → 0.5.0

raw patch · 22 files changed

+1137/−2070 lines, 22 filesdep +comonaddep +mmorphdep +numhaskdep −asyncdep −manageddep −pipesdep ~attoparsecdep ~concurrencydep ~contravariantsetup-changedPVP ok

version bump matches the API change (PVP)

Dependencies added: comonad, mmorph, numhask, optparse-generic

Dependencies removed: async, managed, pipes, protolude, typed-process

Dependency ranges changed: attoparsec, concurrency, contravariant, dejafu, doctest, exceptions, foldl, generic-lens, lens, mtl, profunctors, random, streaming, text, time, transformers, transformers-base, websockets

API changes (from Hackage documentation)

Files

• Setup.hs +0/−2
• app/concurrency-tests.hs +76/−335
• app/websocket-tests.hs +158/−77
• box.cabal +115/−73
• readme.md +11/−9
• src/Box.hs +213/−121
• src/Box/Box.hs +44/−12
• src/Box/Broadcast.hs +0/−62
• src/Box/Committer.hs +46/−32
• src/Box/Connectors.hs +105/−143
• src/Box/Cont.hs +29/−9
• src/Box/Control.hs +0/−438
• src/Box/Emitter.hs +104/−50
• src/Box/IO.hs +76/−167
• src/Box/Plugs.hs +0/−52
• src/Box/Queue.hs +39/−219
• src/Box/Stream.hs +0/−88
• src/Box/Time.hs +63/−43
• src/Box/Transducer.hs +49/−28
• src/Box/Updater.hs +0/−105
• stack.yaml +3/−2
• test/test.hs +6/−3

@@ -1,2 +0,0 @@-import Distribution.Simple-main = defaultMain

@@ -1,365 +1,106 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedLabels #-}+{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RankNTypes #-}+{-# LANGUAGE RebindableSyntax #-} {-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-} {-# LANGUAGE TypeFamilies #-}-{-# LANGUAGE OverloadedLabels #-} {-# OPTIONS_GHC -Wall #-}-{-# OPTIONS_GHC -fno-warn-type-defaults #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE DeriveGeneric #-}-{-# LANGUAGE DataKinds #-}-{-# LANGUAGE TypeApplications #-} {-# OPTIONS_GHC -fno-warn-missing-signatures #-}+{-# OPTIONS_GHC -fno-warn-type-defaults #-}  -- | dejavu testing--- module Main where -import Prelude+import Box+import Control.Lens import Control.Monad.Conc.Class as C-import Control.Concurrent.Classy.STM as C-import Box.Box-import Box.Committer-import Box.Emitter-import Box.Broadcast-import Box.Connectors-import Box.Cont-import Box.IO-import Box.Queue-import Box.Stream-import Box.Transducer-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 NumHask.Prelude hiding (STM) import qualified Streaming.Prelude as S import System.Random-import Control.Lens hiding ((:>), (.>), (<|), (|>))-import Control.Concurrent.Classy.Async as C-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])-testBox n = do-  (_, c, res) <- cCRef-  let e = toEmit (S.take n $ S.each [0..])-  pure (Box <$> c <*> e, res)+import Test.DejaFu hiding (get)+import Test.DejaFu.Types --- | fuse-exFuse :: (MonadConc m) => Int -> m [Int]-exFuse n = do-  (b, res) <- testBox n-  fuse (pure . pure) (liftB <$> b)+tERef :: (MonadConc m) => Emitter m a -> m [a]+tERef e = do+  (c, res) <- cRef+  glue c e   res -exFuseSTM :: (MonadConc m) => Int -> m [Int]-exFuseSTM n = do-  (b, res) <- testBox n-  fuseSTM (pure . pure) b+tEState :: (Monad m) => Emitter m a -> m [a]+tEState e = flip execStateT [] $ glue stateC (hoist lift e)++tToListE :: (MonadConc m) => Int -> m [Int]+tToListE n =+  toListE <$.> fromListE [1 .. n]++tFromListE :: (MonadConc m) => Int -> m [Int]+tFromListE n = do+  (c, res) <- cRef+  let e = fromListE [0 .. (n - 1)]+  fuse (pure . pure) <$.> (Box <$> pure c <*> e)   res -exEtc :: (MonadConc m) => Int -> m [Int]-exEtc n = do-  (b, res) <- testBox n-  etc () (Transducer id) b+tToList_ :: (MonadConc m) => Int -> m [Int]+tToList_ n =+  toList_ <$.> fromListE [1 .. n]++-- failing+tFromList_ :: (MonadConc m) => Int -> m [Int]+tFromList_ n = do+  (c, res) <- cRef+  fromList_ [1 .. n] c   res --- | one emitter and 2 committers - STM fusion-e1c2 :: (MonadConc m) => Cont m (Emitter (STM m) b) -> m ([b],[b])-e1c2 e = do-  (_,c1,r1) <- cCRef-  (_,c2,r2) <- cCRef-  fuseSTM (pure . pure) (Box <$> c1 <*> e)-  fuseSTM (pure . pure) (Box <$> c2 <*> e)-  (,) <$> r1 <*> r2+tFromList_' :: (MonadConc m) => Int -> m [Int]+tFromList_' n = reverse <$> (flip execStateT [] $ fromList_ [1 .. n] stateC) -exe1c2 :: (MonadConc m) => Int -> m ([Int],[Int])-exe1c2 n = e1c2 (toEmit (S.take n $ S.each [0..]))+tPureState :: Int -> [Int]+tPureState n =+  runIdentity $ fmap (reverse . fst) $ flip execStateT ([], [1 .. n]) $ glue (hoist (zoom _1) stateC) (hoist (zoom _2) stateE) --- | one emitter and 2 committers - IO fusion-e1c2IO :: (MonadConc m) => Cont m (Emitter (STM m) b) -> m ([b],[b])-e1c2IO e = do-  (_,c1,r1) <- cCRef-  (_,c2,r2) <- cCRef-  fuse (pure . pure) (liftB <$> (Box <$> c1 <*> e))-  fuse (pure . pure) (liftB <$> (Box <$> c2 <*> e))-  (,) <$> r1 <*> r2+-- tPureBoxF :: (Monad m) => (Box m Int Int -> m ()) -> Int -> m [Int]+-- > tPureBoxF (etc' () (Transducer id))+-- > tPureBoxF (fuse (pure . pure))+-- > tPureBoxF (\(Box c e) -> glue c e)+tPureBoxF f n =+  fmap (reverse . fst) $ flip execStateT ([], [1 .. n]) $ f (Box (hoist (zoom _1) stateC) (hoist (zoom _2) stateE)) -exe1c2IO :: (MonadConc m) => Int -> m ([Int],[Int])-exe1c2IO n = e1c2IO (toEmit (S.take n $ S.each [0..]))+-- tForkEmit <$.> (fromListE [1..4])+-- tForkEmit <$.> (toEmitter (S.take 4 $ S.each [1..]))+tForkEmit :: (MonadConc m) => Emitter m b -> m ([b], [b])+tForkEmit e = do+  (c1, r1) <- cRef+  (c2, r2) <- cRef+  let e' = forkEmit e c1+  glue c2 e'+  (,) <$> r1 <*> r2  -- | test when the deterministic takes too long (which is almost always)-t :: (MonadIO n, Eq b, Show b, MonadDejaFu n) =>-     String -> ConcT n b -> n Bool-t l c = dejafuWay (randomly (mkStdGen 42) 1000) defaultMemType l alwaysSame c+t ::+  (MonadIO n, Eq b, Show b, MonadDejaFu n) =>+  ConcT n b ->+  n Bool+t c = dejafuWay (randomly (mkStdGen 42) 1000) defaultMemType "" alwaysSame c  main :: IO () main = do-  let n = 2-  sequence_ $ autocheck <$> [exFuse n, exFuseSTM n, exEtc n]-  void $ t "e1c2" (exe1c2 10)-  void $ t "e1c2 IO" (exe1c2IO 10)--exc :: (MonadConc m) => Int -> m ([Int],[Int])-exc n = do-  ref <- newIORef 0-  let e = Emitter $ do-        a <- readIORef ref-        if a < n-          then do-            writeIORef ref (a+1)-            pure (Just a)-          else pure Nothing-  (_,c1,r1) <- cCRef-  (_,c2,r2) <- cCRef-  fuse (pure . pure) (Box <$> (liftC <$> c1) <*> pure e)-  fuse (pure . pure) (Box <$> (liftC <$> c2) <*> pure e)-  (,) <$> r1 <*> r2--eCounter :: (C.MonadConc m) => Int -> Int -> m (Emitter m Int, IORef m Int)-eCounter start n = do-  ref <- newIORef start-  pure (-    Emitter $ do-        a <- readIORef ref-        if a < n-          then do-          writeIORef ref (a+1)-          pure (Just a)-        else pure Nothing, ref)--eCounter' :: (C.MonadConc m) => Int -> Int -> m (Cont m (Emitter m Int), IORef m Int)-eCounter' start n = do-  ref <- newIORef start-  pure ( fuseEmitM $ -    Emitter $ do-        a <- readIORef ref-        if a < n-          then do-          writeIORef ref (a+1)-          pure (Just a)-        else pure Nothing, ref)--exc' :: (MonadIO m, MonadConc m) => Int -> m ([Int],[Int])-exc' n = do-  (b, c) <- broadcast'-  (ec, eref) <- eCounter 0 n -  let e1 = subscribe' b-  let e2 = subscribe' b-  fuse' (pure . pure) (pure $ Box c ec)-  (_,c1,r1) <- cCRef-  (_,c2,r2) <- cCRef-  fuse (pure . pure) (Box <$> (liftC <$> c1) <*> e1)-  fuse (pure . pure) (Box <$> (liftC <$> c2) <*> e2)-  eres <- readIORef eref-  liftIO $ Text.putStrLn $ "eref: " <> Text.pack (show eres)-  (,) <$> r1 <*> r2---- | a broadcaster -newtype Broadcaster' m a = Broadcaster'-  { unBroadcast :: TVar (STM m) (Committer m a)-  }---- | create a (broadcaster, committer)-broadcast' :: (Show a, MonadConc m, MonadIO m) => m (Broadcaster' m a, Committer m a)-broadcast' = do-  ref <- C.atomically $ C.newTVar mempty-  let com = Committer $ \a -> do-        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' :: (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)---- * primitives--- | fuse an emitter directly to a committer-fuse_' :: (Show a, MonadIO m) => Emitter m a -> Committer m a -> m ()-fuse_' e c = go-  where-    go = do-      a <- emit e-      liftIO $ Text.putStrLn $ "fuse_' emit: " <> Text.pack (show a)-      c' <- maybe (pure False) (commit c) a-      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 ()-fuse' f box = with box $ \(Box c e) -> fuse_' (emap f e) c---- exEmerge :: (MonadIO m, MonadConc m) => Int -> Int -> Int -> Int -> m [Int]-exEmerge :: MonadConc m => Int -> Int -> Int -> Int -> m ([Int], Int, Int)-exEmerge st1 st2 n1 n2 = do-  (e1, eref1) <- eCounter st1 n1-  (e2, eref2) <- eCounter st2 n2-  (_,c1,r1) <- cCRef-  fuse (pure . pure) $ Box <$> (liftC <$> c1) <*> emergeM (pure (e1, e2))-  (,,) <$> r1 <*> readIORef eref1 <*> readIORef eref2--exEmergeM :: MonadConc m => Int -> Int -> Int -> Int -> m ([Int], Int, Int)-exEmergeM st1 st2 n1 n2 = do-  (e1, eref1) <- eCounter' st1 n1-  (e2, eref2) <- eCounter' st2 n2-  (_,c1,r1) <- cCRef-  fuse (pure . pure) $ Box <$> (liftC <$> c1) <*> emergeM ((,) <$> e1 <*> e2)-  (,,) <$> r1 <*> readIORef eref1 <*> readIORef eref2---temerge :: IO Bool-temerge = dejafuWay-    (randomly (mkStdGen 42) 1000)-    defaultMemType-    "test emergeM"-    alwaysSame-    (exEmergeM 0 0 2 2)--exCSplit :: MonadConc m => Int -> Int -> m [Int]-exCSplit st1 n1 = do-  (e1, _) <- eCounter' st1 n1-  (_,c1,r1) <- cCRef-  fuse (pure . pure) $ Box <$> (liftC <$> liftA2 (<>) c1 c1) <*> e1-  r1--contCommit' :: Either (Committer m Int) (Committer m Int) -> Committer m Int-contCommit' ec =-  Committer $ \a ->-    case ec of-      Left lc -> commit (contramap (100+) lc) a-      Right rc -> commit rc a--splitCommitM :: (MonadConc m) =>-     Cont m (Committer m a)-  -> Cont m (Either (Committer m a) (Committer m a))-splitCommitM c =-  Cont $ \kk ->-    with c $ \c' ->-      fst <$>-      C.concurrently-        (queueCM' (kk . Left) (`fuse_` c'))-        (queueCM' (kk . Right) (`fuse_` c'))--counter :: (MonadState Int m) => Int -> StateT Int m (Emitter m Int)-counter n =-  pure $-    Emitter $ do-        a <- Control.Monad.State.Lazy.get-        case a < n of-          False -> pure Nothing-          True -> do-            put $ a + 1-            pure (Just a)--counterT :: (Num a, Ord a, Monad m) => a -> Emitter (StateT a m) a-counterT n =-    Emitter $ do-        a <- Trans.get-        case a < n of-          False -> pure Nothing-          True -> do-            Trans.put $ a + 1-            pure (Just a)---rememberer :: (MonadState [Int] m) => StateT [Int] m (Committer m Int)-rememberer =-  pure $-    Committer $ \a -> do-        modify (a:)-        pure True--remembererT :: (Monad m) => Committer (StateT [a] m) a-remembererT =-    Committer $ \a -> do-        Trans.modify (a:)-        pure True---data StateExs = StateExs { count :: Int, result :: [Int]} deriving (Show, Eq, Generic)--boxCount ::-  (MonadConc m, MonadState StateExs m) =>-  Int ->-  m (Box m Int Int)-boxCount n = Box <$> pure rememberer' <*> pure counter' where-  counter' =-    Emitter $ do-        a <- use #count-        case a < n of-          False -> pure Nothing-          True -> do-            #count += 1-            pure (Just a)-  rememberer' =-    Committer $ \a -> do-        #result %= (a:)-        pure True--countEmitter :: (Ord a, Num a, MonadState s m, Data.Generics.Product.HasField "count" s s a a) => a -> Emitter m a-countEmitter n = Emitter $ do-  a <- use (field @"count")-  case a < n of-    False -> pure Nothing-    True -> do-      field @"count" += 1-      pure (Just a)--resultCommitter :: (MonadState s m, Data.Generics.Product.HasField "result" s s [a] [a]) => Committer m a-resultCommitter = Committer $ \a -> do-  field @"result" %= (a:)-  pure True---- boxCount' :: (MonadState StateExs m, MonadConc m) => Int -> Box m Int Int--- boxCount' n = Box (zoom #result resultCommitter) (zoom #count (countEmitter n))--exs :: (MonadConc m) => Int -> m [Int]-exs n = do-  (StateExs _ res) <- execStateT-    (fuse (pure . pure) (pure $ Box resultCommitter (countEmitter n)))-    (StateExs 0 [])-  pure (reverse res)--fuse'' :: (Show b, MonadIO m) => (a -> m (Maybe b)) -> Cont m (Box m b a) -> m ()-fuse'' f box = with box $ \(Box c e) -> fuse_' (emap f e) c----- | emitter hooked into broadcasting is broken ...--- > exbr 2--- ([],[])----broadcasting :: MonadConc m => Cont m (Emitter (STM m) b) -> m ([b], [b])-broadcasting e = do-  (b, c) <- C.atomically broadcast-  let e1 = subscribe b-  let e2 = subscribe b-  fuseSTM (pure . pure) (Box <$> pure c <*> e)-  (_,c1,r1) <- cCRef-  (_,c2,r2) <- cCRef-  fuseSTM (pure . pure) (Box <$> c1 <*> e1)-  fuseSTM (pure . pure) (Box <$> c2 <*> e2)-  (,) <$> r1 <*> r2--exbrPure :: (MonadConc m) => Int -> m ([Int],[Int])-exbrPure n = broadcasting (toEmit (S.take n $ S.each [0..]))---- > exbrIO 2--- 1--- 2--- ([],[])----exbrIO :: Int -> IO ([Text],[Text])-exbrIO n = broadcasting (eStdin n)--+  let n = 4+  sequence_ $+    autocheck+      <$> [ tToListE n,+            tFromListE n,+            tToList_ n,+            tFromList_' n,+            pure (tPureState n),+            tPureBoxF (etc () (Transducer id)) n,+            tPureBoxF (fuse (pure . pure)) n,+            tPureBoxF (\(Box c e) -> glue c e) n,+            (\(a, b) -> a <> b) <$> (tForkEmit <$.> (fromListE [1 .. n])),+            (\(a, b) -> a <> b) <$> (tForkEmit <$.> (toEmitter (S.take 4 $ S.each [1 ..])))+          ]

@@ -1,136 +1,217 @@+{-# LANGUAGE DataKinds #-} {-# LANGUAGE DeriveGeneric #-}-{-# LANGUAGE NoImplicitPrelude #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE OverloadedLabels #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE NoImplicitPrelude #-} {-# OPTIONS_GHC -Wall #-}+{-# OPTIONS_GHC -Wno-unused-do-bind #-}  {- -FIXME:--Models a websocket connection:--The client socket reads (Either ControlRequest Text) from stdin.--The server socket processes Left ControlRequests and echoes Right text messages back to the client.--The client prints received messages to stdout.+Models a websocket connection  -}  module Main where  import Box-import Box.Control-import Control.Lens-import Control.Monad.Managed+import qualified Control.Concurrent.Classy.Async as C+import Control.Lens hiding (Unwrapped, Wrapped)+import Control.Monad.Catch+import Control.Monad.Conc.Class as C import Data.Generics.Labels ()-import qualified Data.Text as Text-import GHC.Generics import qualified Network.WebSockets as WS-import Protolude hiding (STM)+import NumHask.Prelude hiding (STM, bracket)+import Options.Generic  data ConfigSocket   = ConfigSocket       { host :: Text,-        port :: Int+        port :: Int,+        path :: Text       }   deriving (Show, Eq, Generic)  defaultConfigSocket :: ConfigSocket-defaultConfigSocket = ConfigSocket "127.0.0.1" 9160+defaultConfigSocket = ConfigSocket "127.0.0.1" 9160 "/" -client :: ConfigSocket -> WS.ClientApp () -> IO ()-client c = WS.runClient (Text.unpack $ c ^. #host) (c ^. #port) "/"+client :: (MonadIO m) => ConfigSocket -> WS.ClientApp () -> m ()+client c app = liftIO $ WS.runClient (unpack $ c ^. #host) (c ^. #port) (unpack $ c ^. #path) app -server :: ConfigSocket -> WS.ServerApp -> IO ()-server c = WS.runServer (Text.unpack $ c ^. #host) (c ^. #port)+server :: (MonadIO m) => ConfigSocket -> WS.ServerApp -> m ()+server c app = liftIO $ WS.runServer (unpack $ c ^. #host) (c ^. #port) app -mconn :: WS.PendingConnection -> Managed WS.Connection-mconn p =-  managed $-    bracket-      (WS.acceptRequest p)-      (\conn -> WS.sendClose conn ("Bye from mconn!" :: Text))+con :: (MonadMask m, MonadIO m) => WS.PendingConnection -> Cont m WS.Connection+con p = Cont $ \action ->+  bracket+    (liftIO $ WS.acceptRequest p)+    (\conn -> liftIO $ WS.sendClose conn ("Bye from con!" :: Text))+    action +clientApp ::+  (MonadIO m, MonadConc m) =>+  Box m (Either Text Text) Text ->+  WS.Connection ->+  m ()+clientApp (Box c e) conn =+  void $+    C.race+      (receiver c conn)+      (sender (Box mempty e) conn)++serverApp ::+  WS.PendingConnection ->+  IO ()+serverApp p =+  with+    (con p)+    ( responder+        (\x -> bool (Right $ "echo:" <> x) (Left "quit") (x == "q"))+        mempty+    )++serverIO :: IO ()+serverIO = server defaultConfigSocket serverApp++clientIO :: IO ()+clientIO =+  (client defaultConfigSocket . clientApp)+    (Box (contramap show toStdout) fromStdin)++data SocketType = Client | Responder | TestRun deriving (Eq, Read, Show, Generic)++instance ParseField SocketType++instance ParseRecord SocketType++instance ParseFields SocketType++data Opts w+  = Opts+      { apptype :: w ::: SocketType <?> "type of websocket app"+      }+  deriving (Generic)++instance ParseRecord (Opts Wrapped)++main :: IO ()+main = do+  o :: Opts Unwrapped <- unwrapRecord "example websocket apps"+  r :: Text <- case apptype o of+    Client -> show <$> clientIO+    Responder -> show <$> q serverIO+    TestRun -> show <$> testRun+  putStrLn r+ -- | default websocket receiver+-- Lefts are info/debug receiver ::-  Committer IO (Either ControlResponse Text) ->+  (MonadIO m) =>+  Committer m (Either Text Text) ->   WS.Connection ->-  IO Bool+  m Bool receiver c conn = go   where     go = do-      msg <- WS.receive conn+      msg <- liftIO $ WS.receive conn       case msg of         WS.ControlMessage (WS.Close w b) ->           commit             c             ( Left-                ( Info $-                    "received close message: " <> show w <> " " <> show b+                ( "receiver: received: close: " <> show w <> " " <> show b                 )             )         WS.ControlMessage _ -> go-        WS.DataMessage _ _ _ msg' -> commit c (Right (WS.fromDataMessage msg')) >> go+        WS.DataMessage _ _ _ msg' -> do+          commit c $ Left $ "receiver: received: " <> (WS.fromDataMessage msg' :: Text)+          _ <- commit c (Right (WS.fromDataMessage msg'))+          go  -- | default websocket sender sender ::-  WS.WebSocketsData a =>-  Emitter IO a ->+  (MonadIO m, WS.WebSocketsData a, Show a) =>+  Box m Text a ->   WS.Connection ->-  IO ()-sender e conn = forever $ do+  m ()+sender (Box c e) conn = forever $ do   msg <- emit e   case msg of     Nothing -> pure ()-    Just msg' -> WS.sendTextData conn msg'--clientApp ::-  Box IO (Either ControlResponse Text) Text ->-  WS.ClientApp ()-clientApp (Box c e) conn =-  void $-    concurrently-      (receiver c conn)-      (sender e conn)---- | single uncontrolled client-clientBox ::-  ConfigSocket ->-  IO ()-clientBox cfg =-  Box.with (liftB <$> (Box <$> showStdout <*> readStdin)) (client cfg . clientApp)+    Just msg' -> do+      commit c $ "sender: sending: " <> (show msg' :: Text)+      liftIO $ WS.sendTextData conn msg' --- | a receiver that immediately sends a response+-- | a receiver that responds based on received Text.+-- lefts are quit signals. Rights are response text. responder ::-  (Text -> Either ControlResponse Text) ->+  (MonadIO m) =>+  (Text -> Either Text Text) ->+  Committer m Text ->   WS.Connection ->-  IO ()-responder f conn = go+  m ()+responder f c conn = go   where     go = do-      msg <- WS.receive conn+      msg <- liftIO $ WS.receive conn       case msg of         WS.ControlMessage (WS.Close _ _) -> do-          WS.sendClose conn ("returning close signal" :: Text)-          go+          commit c "responder: normal close"+          liftIO $ WS.sendClose conn ("received close signal: responder closed." :: Text)         WS.ControlMessage _ -> go-        WS.DataMessage _ _ _ msg' -> WS.sendTextData conn (show $ f $ WS.fromDataMessage msg' :: Text)+        WS.DataMessage _ _ _ msg' -> do+          case (f $ WS.fromDataMessage msg') of+            Left _ -> do+              commit c "responder: sender initiated close"+              liftIO $ WS.sendClose conn ("received close signal: responder closed." :: Text)+            Right r -> do+              commit c ("responder: sending" <> r)+              liftIO $ WS.sendTextData conn r+              go -serverApp ::-  WS.PendingConnection ->-  IO ()-serverApp p = Control.Monad.Managed.with (mconn p) (responder Right)+q :: IO a -> IO (Either () a)+q f = race (cancelQ fromStdin) f --- | controlled server-serverBox ::-  ConfigSocket ->-  IO ()-serverBox cfg =-  Box.with-    (Box <$> showStdout <*> readStdin)-    (controlBox defaultControlConfig (server cfg serverApp))+cancelQ :: Emitter IO Text -> IO ()+cancelQ e = do+  e' <- emit e+  case e' of+    Just "q" -> pure ()+    _ -> do+      putStrLn ("nothing happens" :: Text)+      cancelQ e -main :: IO ()-main = pure ()+-- | test of clientApp via a cRef committer and a canned list of Text+tClient :: [Text] -> IO [Either Text Text]+tClient xs = do+  (c, r) <- cRef+  client+    defaultConfigSocket+    ( \conn ->+        (\b -> clientApp b conn)+          <$.> ( Box+                   <$> pure c+                   <*> fromListE (xs <> ["q"])+               )+    )+  r++tClientIO :: [Text] -> IO ()+tClientIO xs =+  (client defaultConfigSocket . clientApp)+    <$.> (Box (contramap show toStdout) <$> (fromListE (xs <> ["q"])))++-- | main test run of client-server functionality+-- the code starts a server in a thread, starts the client in the main thread, and cancels the server on completion.+-- > testRun+-- [Left "receiver: received: echo:1",Right "echo:1",Left "receiver: received: echo:2",Right "echo:2",Left "receiver: received: echo:3",Right "echo:3",Left "receiver: received: close: 1000 \"received close signal: responder closed.\""]+testRun :: IO [Either Text Text]+testRun = do+  a <- async (server defaultConfigSocket serverApp)+  r <- tClient (show <$> [1 .. 3 :: Int])+  cancel a+  pure r

@@ -1,6 +1,6 @@-cabal-version: 3.0+cabal-version:  2.4 name:           box-version:        0.4.0+version:        0.5.0 synopsis:       boxes description:    concurrent, effectful boxes category:       project@@ -13,8 +13,8 @@ license-file:   LICENSE build-type:     Simple extra-source-files:-    stack.yaml-    readme.md+  stack.yaml+  readme.md  source-repository head   type: git@@ -22,94 +22,136 @@  library   exposed-modules:-      Box-      Box.Box-      Box.Broadcast-      Box.Committer-      Box.Connectors-      Box.Cont-      Box.Control-      Box.Emitter-      Box.IO-      Box.Plugs-      Box.Queue-      Box.Stream-      Box.Time-      Box.Transducer-      Box.Updater+    Box+    Box.Box+    Box.Committer+    Box.Connectors+    Box.Cont+    Box.Emitter+    Box.IO+    Box.Queue+    Box.Time+    Box.Transducer   hs-source-dirs:-      src-  ghc-options: -Wall -Wcompat -Wincomplete-record-updates -Wincomplete-uni-patterns -Wredundant-constraints -funbox-strict-fields+    src+  default-extensions:+    NegativeLiterals+    NoImplicitPrelude+    OverloadedStrings+    UnicodeSyntax+  ghc-options:+    -Wall+    -Wcompat+    -Wincomplete-record-updates+    -Wincomplete-uni-patterns+    -Wredundant-constraints+    -funbox-strict-fields   build-depends:-      async-    , attoparsec-    , base >=4.7 && <5-    , concurrency-    , contravariant-    , exceptions-    , foldl-    , lens-    , pipes-    , profunctors-    , protolude >= 0.3.0 && < 0.4-    , streaming-    , text-    , time-    , transformers-    , transformers-base-    , typed-process+    attoparsec >= 0.13,+    base >=4.7 && <5,+    comonad >= 5.0,+    concurrency >= 1.11,+    contravariant >= 1.5,+    exceptions >= 0.10,+    foldl >= 1.4,+    lens >= 4.19,+    mmorph >= 1.1,+    numhask >= 0.6,+    profunctors >= 5.5,+    streaming >= 0.2,+    text >= 1.2,+    time >= 1.9,+    transformers >= 0.5,+    transformers-base >= 0.4,   default-language: Haskell2010  executable concurrency-tests   main-is: concurrency-tests.hs   hs-source-dirs:-      app-  ghc-options: -funbox-strict-fields -fforce-recomp -threaded -rtsopts -with-rtsopts=-N+    app   build-depends:-    base >=4.7 && <5-    , box-    , concurrency-    , dejafu-    , generic-lens-    , lens-    , mtl-    , protolude-    , random-    , streaming-    , text-    , transformers+    base >=4.7 && <5,+    box,+    concurrency >= 1.11,+    dejafu >= 2.3,+    generic-lens >= 2.0,+    lens >= 4.19,+    numhask >= 0.6,+    random >= 1.1,+    streaming >= 0.2,+    text >= 1.2,+    transformers >= 0.5   default-language: Haskell2010-+  default-extensions:+    NegativeLiterals+    NoImplicitPrelude+    OverloadedStrings+    UnicodeSyntax+  ghc-options:+    -Wall+    -Wcompat+    -Wincomplete-record-updates+    -Wincomplete-uni-patterns+    -Wredundant-constraints+    -funbox-strict-fields+    -fforce-recomp+    -threaded+    -rtsopts+    -with-rtsopts=-N executable websocket-tests   main-is: websocket-tests.hs   hs-source-dirs:-      app+    app   ghc-options: -funbox-strict-fields -fforce-recomp -threaded -rtsopts -with-rtsopts=-N   build-depends:-    base >=4.7 && <5-    , box-    , concurrency-    , dejafu-    , generic-lens-    , lens-    , managed-    , mtl-    , protolude-    , random-    , streaming-    , text-    , transformers-    , websockets+    base >=4.7 && <5,+    box,+    concurrency >= 1.11,+    dejafu >= 2.3,+    exceptions >= 0.10,+    generic-lens >= 2.0,+    lens >= 4.19,+    mtl >= 2.2,+    numhask >= 0.6,+    optparse-generic >= 1.3.0 && < 1.4,+    streaming >= 0.2,+    text >= 1.2,+    transformers >= 0.5,+    websockets >= 0.12   default-language: Haskell2010+  default-extensions:+    NegativeLiterals+    NoImplicitPrelude+    OverloadedStrings+    UnicodeSyntax+  ghc-options:+    -Wall+    -Wcompat+    -Wincomplete-record-updates+    -Wincomplete-uni-patterns+    -Wredundant-constraints+    -funbox-strict-fields  test-suite test   type: exitcode-stdio-1.0   main-is: test.hs   hs-source-dirs:-      test-  ghc-options: -Wall -Wcompat -Wincomplete-record-updates -Wincomplete-uni-patterns -Wredundant-constraints+    test   build-depends:-    base >=4.7 && <5-    , doctest+    base >=4.7 && <5,+    box,+    doctest >= 0.16,+    numhask >= 0.6,   default-language: Haskell2010-+  default-extensions:+    NegativeLiterals+    NoImplicitPrelude+    OverloadedStrings+    UnicodeSyntax+  ghc-options:+    -Wall+    -Wcompat+    -Wincomplete-record-updates+    -Wincomplete-uni-patterns+    -Wredundant-constraints+    -funbox-strict-fields

@@ -3,20 +3,23 @@  > The box is dark and full of terrors. -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:+Take two ubiquitous concepts. There are things that emit stuff. They instantiate a particular thing on demand, somehow and somewhere beyond our concern and viewbox.  We ask an Emitter for an `a` and an `a` is there, or not (best to assume a `Maybe a`). The library calls this an Emitter:  ``` newtype Emitter m a = Emitter { emit :: m (Maybe a)} ```+It emits over a type constructor, m, and this is often monadic in nature, and often IO. -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:+Then there is the opposite.  There are things that commit stuff. We offer a Committer a particular thing and they take it away beyond our concerns. It is committed to the void. We offer an `a` and it is reported back whether the committment was succesful.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:+These two types are duals, across a wide spectrum of what that means. If you do something with a committer, like write to a socket, you can often immediately write an emitter the dual of this, like read from a socket. +When you have both things, something that emits and something that commits, over the same carrier, you have a Box:+ ``` data Box m c e = Box   { committer :: Committer m c@@ -24,15 +27,15 @@   } ``` -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. +A Box tends to flip polarity, like a Möbius strip. and sometimes like a Dali. +You can see a committer as a wire into the box, and an emitter as a wire out of the box. You can then think of the box as opaque, and make the tasks on the outside of the box that much easier, removing concern about what is going on inside the box. But then coding tends to get you inside the box, and point-of-view shifts. From inside the box, stuff,, a's,, are coming from what we had thought of as the committer. It is emitting from our new viewbox, and vice versa. You are the black box, and your box matters. + 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 m c e` unifies the functorial wrapper `m` of the emitter and committer, which is what really makes it a functional box. Boxes can often be hooked together at this level for useful efficiencies.  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. @@ -40,7 +43,6 @@  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  @@ -48,5 +50,5 @@ ---  ```-stack build --test --exec "$(stack path --local-install-root)/bin/concurrency-tests" --file-watch+stack exec --test concurrency-tests --file-watch ```

@@ -1,165 +1,257 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NoImplicitPrelude #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeFamilies #-} {-# OPTIONS_GHC -Wall #-} --- | Boxes that `emit`, `transduce` & `commit`+-- | Effectful, profunctor boxes designed for concurrency. -- -- This library follows the ideas and code from [pipes-concurrency](https://hackage.haskell.org/package/pipes-concurrency) and [mvc](https://hackage.haskell.org/package/mvc) but with some polymorphic tweaks and definitively more pretentious names. module Box   ( -- $setup+    -- $continuations+    -- $boxes     -- $commit     -- $emit-    -- $transduce+    -- $state+    -- $finite     module Box.Box,-    module Box.Broadcast,     module Box.Committer,     module Box.Connectors,     module Box.Cont,     module Box.Emitter,     module Box.IO,-    module Box.Plugs,     module Box.Queue,-    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 import Box.Emitter import Box.IO-import Box.Plugs import Box.Queue-import Box.Stream import Box.Time import Box.Transducer-import Control.Lens ((&)) --- $setup--- >>> :set -XOverloadedStrings--- >>> :set -XGADTs--- >>> 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'--- >>> let transducer' = Transducer $ \s -> s & S.takeWhile (/="q") & S.map ("echo: " <>)+{- $setup+>>> :set -XOverloadedStrings+>>> :set -XGADTs+>>> :set -XNoImplicitPrelude+>>> :set -XFlexibleContexts+>>> import NumHask.Prelude+>>> import qualified Prelude as P+>>> import Data.Functor.Contravariant+>>> import Box+>>> import qualified Streaming.Prelude as S+>>> import Control.Monad.Conc.Class as C+>>> import Control.Lens+-} --- toEmit . S.each--- --- | emitting--- > toListE . fromListE :: [a] -> IO [a]--- >>>  toListE . fromListE $ [1..3::Int]--- [1,2,3]------ > fuse (pure . Just) $ Box <$> (liftC <$> showStdout) <*> (fromListE [1..3::Int])+{- $continuations +Continuations are very common in the API with 'Cont' as an inhouse type. +>>> :t fromListE [1..3::Int]+fromListE [1..3::Int] :: MonadConc m => Cont m (Emitter m Int) --- $commit--- committing------ >>> with (cStdout 100) $ \c -> C.atomically (commit c "something")--- something--- True------ 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.------ >>> 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 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--- stolen!--- x------ prism handler------ >>> import Control.Lens (_Right)--- >>> let e2 = (fmap (\x -> Right (x <> "_right")) <$> e) <> (fmap (\x -> Left (x <> "_left")) <$> e)--- >>> let cright = handles _Right <$> c--- >>> fuse (pure . pure) $ Box <$> cright <*> e2--- hi_right--- bye_right--- q_right--- x_right------ | splitCommit--- 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+The applicative is usually the easiest way to think about and combine continuations with their unadorned counterparts. --- $emit------ >>> with (S.each [0..] & toEmit) (C.atomically . emit) >>= print--- Just 0------ >>> let c = fmap liftC $ cStdout 10--- >>> let e = fmap liftE $ toEmit (S.each ["hi","bye","q","x"])--- >>> let e' = emap (\a -> if a=="q" then (sleep 0.1 >> putStrLn "stole a q!" >> sleep 0.1 >> pure (Nothing)) else (pure (Just a))) <$> e :: Cont IO (Emitter IO Text)--- >>> fuse (pure . pure) $ Box <$> c <*> e'--- hi--- bye--- stole a q!--- x------ >>> 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 id) $ Box <$> cStdout 6 <*> emerge ((,) <$> e1 <*> e2)--- 'a'--- 'z'--- 'b'--- 'y'--- 'c'--- 'x'------ >>> etc () (Transducer id) $ Box <$> cStdout 6 <*> (liftA2 (<>) e1 e2)--- 'a'--- 'z'--- 'b'--- 'y'--- 'c'--- 'x'+>>> let box' = Box <$> pure toStdout <*> fromListE ["a", "b" :: Text]+>>> :t box'+box' :: Cont IO (Box IO Text Text) --- $transduce------ >>> etc () transducer' box'--- echo: hi--- echo: bye+-} --- | broadcasting------ >>> (bcast, bcom) <- (C.atomically broadcast) :: IO (Broadcaster (C.STM IO) Text, Committer (C.STM IO) Text)------ > (funn, fem) <- C.atomically funnel--- >+{- $boxes++The basic ways of connecting up a box are all related as follows:++> glue c e == glueb (Box c e)+> glueb == fuse (pure . pure) == etc () (Transducer id)++>>> fromToList_ [1..3] glueb+[1,2,3]++>>> fromToList_ [1..3] (fuse (pure . pure))+[1,2,3]++>>> fromToList_ [1..3] (etc () (Transducer id))+[1,2,3]++1. glue: direct fusion of committer and emitter++>>> runCont $ glue <$> pure toStdout <*> fromListE (show <$> [1..3])+1+2+3++Variations to the above code include:++Use of continuation applicative operators:++- the '(<*.>)' operator is short hand for runCont $ xyz '(<*>)' zy.++- the '(<$.>)' operator is short hand for runCont $ xyz '(<$>)' zy.++> glue <$> pure toStdout <*.> fromListE (show <$> [1..3])+> glue toStdout <$.> fromListE (show <$> [1..3])++Changing the type in the Emitter (The double fmap is cutting through the Cont and Emitter layers):++> glue toStdout <$.> fmap (fmap show) (fromListE [1..3])++Changing the type in the committer (which is Contrvariant so needs to be a contramap):++> glue (contramap show toStdout) <$.> fromListE [1..3]++Using the box version of glue:++> glueb <$.> (Box <$> pure toStdout <*> (fmap show <$> fromListE [1..3]))++2. fusion of a box, with an (a -> m (Maybe b)) function to allow for mapping, filtering and simple effects.++>>> let box' = Box <$> pure toStdout <*> fromListE (show <$> [1..3])+>>> fuse (\a -> bool (pure $ Just $ "echo: " <> a) (pure Nothing) (a==("2"::Text))) <$.> box'+echo: 1+echo: 3++3. Transduction++For complicated branching and layering, the library provides an API into the streaming library, via 'etc'.++>>> :t etc+etc :: Monad m => s -> Transducer s a b -> Box m b a -> m s++A 'Transducer' is a pure, stateful stream to stream transformation that is highly composable, abstraction leakage-proof and has a category instance.++'etc' stands for emit -> transduce -> commit, and can be considered a distant relative of the model-view-controller abstraction.++>>> let transducer' = Transducer $ \s -> s & S.takeWhile (/="3") & S.chain (const (modify (+1))) & S.map ("echo: " <>)+>>> etc 0 transducer' <$.> box'+echo: 1+echo: 2+2++-}++{- $commit++>>> commit toStdout "I'm committed!"+I'm committed!+True++Use mapC to modify a Committer and introduce effects.++>>> let c = mapC (\a -> if a==2 then (sleep 0.1 >> putStrLn "stole a 2!" >> sleep 0.1 >> pure (Nothing)) else (pure (Just a))) (contramap (show :: Int -> Text) toStdout)+>>> glueb <$.> (Box <$> pure c <*> fromListE [1..3])+1+stole a 2!+3++The monoid instance of Committer sends each commit to both mappended committers. Because effects are also mappended together, the committed result is not always what is expected.++>>> let cFast = mapC (\b -> pure (Just b)) . contramap ("fast: " <>) $ toStdout+>>> let cSlow = mapC (\b -> sleep 0.1 >> pure (Just b)) . contramap ("slow: " <>) $ toStdout+>>> (glueb <$.> (Box <$> pure (cFast <> cSlow) <*> fromListE (show <$> [1..3]))) <* sleep 1+fast: 1+slow: 1+fast: 2+slow: 2+fast: 3+slow: 3++To approximate what is intuitively expected, use 'concurrentC'.++>>> runCont $ (fromList_ (show <$> [1..3]) <$> (concurrentC cFast cSlow)) <> pure (sleep 1)+fast: 1+fast: 2+fast: 3+slow: 1+slow: 2+slow: 3++This is all non-deterministic, hence the necessity for messy delays and heuristic avoidance of console races.++-}++{- $emit++>>> ("I'm emitted!" :: Text) & Just & pure & Emitter & emit >>= print+Just "I'm emitted!"++>>> with (fromListE [1]) (\e' -> (emit e' & fmap show :: IO Text) >>= putStrLn & replicate 3 & sequence_)+Just 1+Nothing+Nothing++>>> toListE <$.> (fromListE [1..3])+[1,2,3]++The monoid instance is left-biased.++>>> toListE <$.> (fromListE [1..3] <> fromListE [7..9])+[1,2,3,7,8,9]++Use concurrentE to get some nondeterministic balance.++> let es = (join $ concurrentE <$> (fromListE [1..3]) <*> (fromListE [7..9]))+> glue (contramap show toStdout) <$.> es+1+2+7+3+8+9++-}++{- $state++State committers and emitters are related as follows:++>>> runIdentity $ fmap (reverse . fst) $ flip execStateT ([],[1..4]) $ glue (hoist (zoom _1) stateC) (hoist (zoom _2) stateE)+[1,2,3,4]++For some reason, related to a lack of an MFunctor instance for Cont, but exactly not yet categorically pinned to a wall, the following compiles but is wrong.++>>> flip runStateT [] $ runCont $ glue <$> pure stateC <*> fromListE [1..4]+((),[])++-}++{- $finite++Most committers and emitters will run forever until:++- the glued or fused other-side returns.+- the Transducer, stream or monadic action returns.++Finite ends (collective noun for emitters and committers) can be created with 'sink' and 'source' eg++>>> glue <$> contramap (show :: Int -> Text) <$> (sink 5 putStrLn) <*.> fromListE [1..]+1+2+3+4+5++Two infinite ends will tend to run infinitely.++> glue <$> pure (contramap show toStdout) <*.> fromListE [1..]++1+2+...+💁+∞++-}+

@@ -10,17 +10,20 @@ -- | A box is something that commits and emits module Box.Box   ( Box (..),-    liftB,     bmap,+    hoistb,+    glue,+    glueb,+    fuse,+    dotb,   ) where -import Prelude import Box.Committer import Box.Emitter-import Control.Applicative-import Control.Lens hiding ((.>), (:>), (<|), (|>))-import Control.Monad.Conc.Class+import Data.Functor.Contravariant+import Data.Profunctor+import NumHask.Prelude  -- | 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.@@ -32,6 +35,10 @@         emitter :: Emitter m e       } +-- | Wrong signature for the MFunctor class+hoistb :: Monad m => (forall a. m a -> n a) -> Box m c e -> Box n c e+hoistb nat (Box c e) = Box (hoist nat c) (hoist nat e)+ instance (Functor m) => Profunctor (Box m) where   dimap f g (Box c e) = Box (contramap f c) (fmap g e) @@ -39,15 +46,40 @@   (<>) (Box c e) (Box c' e') = Box (c <> c') (e <> e')  instance (Alternative m, Monad m) => Monoid (Box m c e) where-   mempty = Box mempty mempty-   mappend = (<>) --- | lift a box from STM-liftB :: (MonadConc m) => Box (STM m) a b -> Box m a b-liftB (Box c e) = Box (liftC c) (liftE e)- -- | a profunctor dimapMaybe bmap :: (Monad m) => (a' -> m (Maybe a)) -> (b -> m (Maybe b')) -> Box m a b -> Box m a' b'-bmap fc fe (Box c e) = Box (cmap fc c) (emap fe e)+bmap fc fe (Box c e) = Box (mapC fc c) (mapE fe e)++{-+instance Category (Box Identity) where+  id = Box ??? ???+  (.) (Box c e) (Box c' e') = runIdentity $ glue c e' >> pure (Box c' e)+-}++-- | composition of monadic boxes+dotb :: (Monad m) => Box m a b -> Box m b c -> m (Box m a c)+dotb (Box c e) (Box c' e') = glue c' e *> pure (Box c e')++-- | Connect an emitter directly to a committer of the same type.+--+-- The monadic action returns when the committer finishes.+glue :: (Monad m) => Committer m a -> Emitter m a -> m ()+glue c e = go+  where+    go = do+      a <- emit e+      c' <- maybe (pure False) (commit c) a+      when c' go++-- | Short-circuit a homophonuos box.+glueb :: (Monad m) => Box m a a -> m ()+glueb (Box c e) = glue c e++-- | fuse a box+--+-- > fuse (pure . pure) == glueb == etc () (Transducer id)+fuse :: (Monad m) => (a -> m (Maybe b)) -> Box m b a -> m ()+fuse f (Box c e) = glue c (mapE f e)

@@ -1,62 +0,0 @@-{-# OPTIONS_GHC -Wall #-}---- | This module is experimental and may not work.-module Box.Broadcast-  ( Broadcaster (..),-    broadcast,-    subscribe,-    Funneler (..),-    funnel,-    widen,-  )-where--import Prelude-import Box.Committer-import Box.Cont-import Box.Emitter-import Box.Queue-import Control.Concurrent.Classy.STM as C-import Control.Monad.Conc.Class as C---- | a broadcaster-newtype Broadcaster m a-  = Broadcaster-      { unBroadcast :: TVar m (Committer m a)-      }---- | create a (broadcaster, committer)-broadcast :: (MonadSTM stm) => stm (Broadcaster stm a, Committer stm a)-broadcast = do-  ref <- newTVar mempty-  let com = Committer $ \a -> do-        c <- readTVar ref-        commit c a-  return (Broadcaster ref, com)---- | subscribe to a broadcaster-subscribe :: (MonadConc m) => Broadcaster (STM m) a -> Cont m (Emitter (STM m) a)-subscribe (Broadcaster tvar) = Cont $ \e -> queueE' cio e-  where-    cio c = atomically $ modifyTVar' tvar (mappend c)---- | a funneler-newtype Funneler m a-  = Funneler-      { unFunnel :: TVar m (Emitter m a)-      }---- | create a (funneler, emitter)-funnel :: (MonadSTM stm) => stm (Funneler stm a, Emitter stm a)-funnel = do-  ref <- newTVar mempty-  let em =-        Emitter $ do-          e <- readTVar ref-          emit e-  pure (Funneler ref, em)---- | widen to a funneler-widen :: (MonadConc m) => Funneler (STM m) a -> Cont m (Committer (STM m) a)-widen (Funneler tvar) =-  Cont $ \c -> queueC' c $ \e -> atomically $ modifyTVar' tvar (mappend e)

@@ -5,24 +5,24 @@ {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeFamilies #-} {-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE NoImplicitPrelude #-} {-# OPTIONS_GHC -Wall #-}  -- | `commit` module Box.Committer   ( Committer (..),-    liftC,-    cmap,-    handles,+    drain,+    mapC,+    premapC,+    postmapC,+    stateC,+    listC,   ) where -import Prelude-import Control.Lens hiding ((.>), (:>), (<|), (|>))-import Control.Monad.Conc.Class as C-import Data.Functor.Constant+import Data.Functor.Contravariant import Data.Functor.Contravariant.Divisible-import Data.Monoid (First (..))-import Data.Void (absurd)+import NumHask.Prelude  -- | a Committer a "commits" values of type a. A Sink and a Consumer are some other metaphors for this. --@@ -32,11 +32,13 @@       { commit :: a -> m Bool       } +instance MFunctor Committer where+  hoist nat (Committer c) = Committer $ nat . c+ instance (Applicative m) => Semigroup (Committer m a) where   (<>) i1 i2 = Committer (\a -> (||) <$> commit i1 a <*> commit i2 a)  instance (Applicative m) => Monoid (Committer m a) where-   mempty = Committer (\_ -> pure False)    mappend = (<>)@@ -45,7 +47,6 @@   contramap f (Committer a) = Committer (a . f)  instance (Applicative m) => Divisible (Committer m) where-   conquer = Committer (\_ -> pure False)    divide f i1 i2 =@@ -54,7 +55,6 @@         (b, c) -> (||) <$> commit i1 b <*> commit i2 c  instance (Applicative m) => Decidable (Committer m) where-   lose f = Committer (absurd . f)    choose f i1 i2 =@@ -63,13 +63,15 @@         Left b -> commit i1 b         Right c -> commit i2 c --- | lift a committer from STM-liftC :: (MonadConc m) => Committer (STM m) a -> Committer m a-liftC c = Committer $ atomically . commit c+-- | Do nothing with values that are committed.+--+-- This is useful for keeping the commit end of a box or pipeline open.+drain :: (Applicative m) => Committer m a+drain = Committer (\_ -> pure True)  -- | This is a contramapMaybe, if such a thing existed, as the contravariant version of a mapMaybe.  See [witherable](https://hackage.haskell.org/package/witherable)-cmap :: (Monad m) => (b -> m (Maybe a)) -> Committer m a -> Committer m b-cmap f c = Committer go+mapC :: (Monad m) => (b -> m (Maybe a)) -> Committer m a -> Committer m b+mapC f c = Committer go   where     go b = do       fb <- f b@@ -77,20 +79,32 @@         Nothing -> pure True         Just fb' -> commit c fb' --- | prism handler-handles ::+-- | adds a monadic action to the committer+premapC ::+  (Applicative m) =>+  (Committer m a -> m ()) ->+  Committer m a ->+  Committer m a+premapC f c = Committer $ \a -> f c *> commit c a++-- | adds a post-commit monadic action to the committer+postmapC ::   (Monad m) =>-  -- |-  ((b -> Constant (First b) b) -> (a -> Constant (First b) a)) ->-  -- |-  Committer m b ->+  (Committer m a -> m ()) ->+  Committer m a ->   Committer m a-handles k (Committer commit_) =-  Committer-    ( \a ->-        case match a of-          Nothing -> return False-          Just b -> commit_ b-    )-  where-    match = getFirst . getConstant . k (Constant . First . Just)+postmapC f c = Committer $ \a -> do+  r <- commit c a+  f c+  pure r++-- | commit to a StateT list+stateC :: (Monad m) => Committer (StateT [a] m) a+stateC = Committer $ \a -> do+  modify (a :)+  pure True++-- | list committer+listC :: (Monad m) => Committer m a -> Committer m [a]+listC c = Committer $ \as ->+  any id <$> (sequence $ commit c <$> as)

@@ -8,74 +8,99 @@  -- | various ways to connect things up module Box.Connectors-  ( fuse_,-    fuseSTM_,-    fuse,-    fuseSTM,+  ( fromListE,+    fromList_,+    toList_,+    fromToList_,+    emitQ,+    commitQ,+    sink,+    source,     forkEmit,     feedback,-    feedbackE,-    fuseEmit,-    fuseEmitM,-    fuseCommit,-    fuseCommitM,-    emerge,-    emergeM,-    splitCommit,-    splitCommitSTM,-    contCommit,-    fromListE,-    toListE,+    queueCommitter,+    queueEmitter,+    concurrentE,+    concurrentC,   ) where -import Prelude import Box.Box import Box.Committer import Box.Cont import Box.Emitter import Box.Queue import Control.Concurrent.Classy.Async as C-import Control.Monad-import Control.Monad.Conc.Class as C+import Control.Lens+import Control.Monad.Conc.Class (MonadConc)+import NumHask.Prelude hiding (STM, atomically) --- * primitives+-- | Turn a list into an 'Emitter' continuation via a 'Queue'+fromListE :: (MonadConc m) => [a] -> Cont m (Emitter m a)+fromListE xs = Cont $ queueE (eListC (Emitter . pure . Just <$> xs)) --- | fuse an emitter directly to a committer-fuse_ :: (Monad m) => Emitter m a -> Committer m a -> m ()-fuse_ e c = go-  where-    go = do-      a <- emit e-      c' <- maybe (pure False) (commit c) a-      when c' go+eListC :: (Monad m) => [Emitter m a] -> Committer m a -> m ()+eListC [] _ = pure ()+eListC (e : es) c = do+  x <- emit e+  case x of+    Nothing -> pure ()+    Just x' -> commit c x' *> eListC es c --- | slightly more efficient version-fuseSTM_ :: (MonadConc m) => Emitter (STM m) a -> Committer (STM m) a -> m ()-fuseSTM_ e c = go-  where-    go = do-      b <--        C.atomically $ do-          a <- emit e-          maybe (pure False) (commit c) a-      when b go+-- | fromList_ directly supplies to a committer action+--+-- FIXME: fromList_ combined with cRef is failing dejavu concurrency testing...+fromList_ :: Monad m => [a] -> Committer m a -> m ()+fromList_ xs c = flip evalStateT xs $ glue (hoist lift c) stateE --- | fuse a box+-- | toList_ directly receives from an emitter ----- > (fuse (pure . Just) $ liftB <$> (Box <$> cStdout 2 <*> emitter')) >> sleep 1--- hi--- bye+-- TODO: check isomorphism ----- > etc () (Transducer id) == fuse (pure . pure) . fmap liftB-fuse :: (Monad m) => (a -> m (Maybe b)) -> Cont m (Box m b a) -> m ()-fuse f box = with box $ \(Box c e) -> fuse_ (emap f e) c+-- > toList_ == toListE+toList_ :: (Monad m) => Emitter m a -> m [a]+toList_ e = reverse <$> flip execStateT [] (glue stateC (hoist lift e)) --- | fuse a box with an STM mapMaybe action-fuseSTM :: (MonadConc m) => (a -> (STM m) (Maybe b)) -> Cont m (Box (STM m) b a) -> m ()-fuseSTM f box = with box $ \(Box c e) -> fuseSTM_ (emap f e) c+-- | take a list, emit it through a box, and output the committed result.+--+-- The pure nature of this computation is highly useful for testing,+-- especially where parts of the box under investigation has non-deterministic attributes.+fromToList_ :: (Monad m) => [a] -> (Box (StateT ([b], [a]) m) b a -> StateT ([b], [a]) m r) -> m [b]+fromToList_ xs f = do+  (res, _) <- flip execStateT ([], xs) $ f (Box (hoist (zoom _1) stateC) (hoist (zoom _2) stateE))+  pure (reverse res) --- | fuse-branch an emitter+-- | hook a committer action to a queue, creating an emitter continuation+emitQ :: (MonadConc m) => (Committer m a -> m r) -> Cont m (Emitter m a)+emitQ cio = Cont $ \eio -> queueE cio eio++-- | hook a committer action to a queue, creating an emitter continuation+commitQ :: (MonadConc m) => (Emitter m a -> m r) -> Cont m (Committer m a)+commitQ eio = Cont $ \cio -> queueC cio eio++-- | singleton sink+sink1 :: (Monad m) => (a -> m ()) -> Emitter m a -> m ()+sink1 f e = do+  a <- emit e+  case a of+    Nothing -> pure ()+    Just a' -> f a'++-- | finite sink+sink :: (MonadConc m) => Int -> (a -> m ()) -> Cont m (Committer m a)+sink n f = commitQ $ replicateM_ n . sink1 f++-- | singleton source+source1 :: (Monad m) => m a -> Committer m a -> m ()+source1 a c = do+  a' <- a+  void $ commit c a'++-- | finite source+source :: (MonadConc m) => Int -> m a -> Cont m (Emitter m a)+source n f = emitQ $ replicateM_ n . source1 f++-- | glues an emitter to a committer, then resupplies the emitter forkEmit :: (Monad m) => Emitter m a -> Committer m a -> Emitter m a forkEmit e c =   Emitter $ do@@ -83,82 +108,51 @@     maybe (pure ()) (void <$> commit c) a     pure a --- * buffer hookups- -- | fuse a committer to a buffer-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+queueCommitter :: (MonadConc m) => Committer m a -> Cont m (Committer m a)+queueCommitter c = Cont $ \caction -> queueC caction (glue c)  -- | fuse an emitter to a buffer-fuseEmitM :: (MonadConc m) => Emitter m a -> Cont m (Emitter m a)-fuseEmitM e = Cont $ \eaction -> queueEM' (fuse_ e) eaction+queueEmitter :: (MonadConc m) => Emitter m a -> Cont m (Emitter m a)+queueEmitter e = Cont $ \eaction -> queueE (`glue` e) eaction --- | merge two emitters+-- | concurrently run two emitters ----- This differs from `liftA2 (<>)` in that the monoidal (and alternative) instance of an Emitter is left-biased (The left emitter exhausts before the right one is begun). This merge is concurrent.-emerge ::-  (MonadConc m) =>-  Cont m (Emitter (STM m) a, Emitter (STM m) a) ->-  Cont m (Emitter (STM m) a)-emerge e =-  Cont $ \eaction ->-    with e $ \e' ->-      fst-        <$> C.concurrently-          (queueE' (fuseSTM_ (fst e')) eaction)-          (queueE' (fuseSTM_ (snd e')) eaction)---- | monadic version-emergeM ::+-- This differs from mappend in that the monoidal (and alternative) instance of an Emitter is left-biased (The left emitter exhausts before the right one is begun). This is non-deterministically concurrent.+concurrentE ::   (MonadConc m) =>-  Cont m (Emitter m a, Emitter m a) ->+  Emitter m a ->+  Emitter m a ->   Cont m (Emitter m a)-emergeM e =+concurrentE e e' =   Cont $ \eaction ->-    with e $ \e' ->-      fst-        <$> C.concurrently-          (queueEM' (fuse_ (fst e')) eaction)-          (queueEM' (fuse_ (snd e')) eaction)+    fst+      <$> C.concurrently+        (queueE (`glue` e) eaction)+        (queueE (`glue` e') eaction) --- | split a committer (STM m)-splitCommitSTM ::-  (MonadConc m) =>-  Cont m (Committer (STM m) a) ->-  Cont m (Either (Committer (STM m) a) (Committer (STM m) a))-splitCommitSTM c =-  Cont $ \kk ->-    with c $ \c' ->-      concurrentlyLeft-        (queueC' (kk . Left) (`fuseSTM_` c'))-        (queueC' (kk . Right) (`fuseSTM_` c'))+-- | run two committers concurrently+concurrentC :: (MonadConc m) => Committer m a -> Committer m a -> Cont m (Committer m a)+concurrentC c c' = mergeC <$> eitherC c c' --- | split a committer-splitCommit ::+eitherC ::   (MonadConc m) =>-  Cont m (Committer m a) ->+  Committer m a ->+  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'))+eitherC cl cr =+  Cont $+    \kk ->+      fst+        <$> C.concurrently+          (queueC (kk . Left) (glue cl))+          (queueC (kk . Right) (glue cr)) --- | use a split committer-contCommit :: Either (Committer m a) (Committer m b) -> (Committer m a -> Committer m b) -> Committer m b-contCommit ec f =+mergeC :: Either (Committer m a) (Committer m a) -> Committer m a+mergeC ec =   Committer $ \a ->     case ec of-      Left lc -> commit (f lc) a+      Left lc -> commit lc a       Right rc -> commit rc a  -- | a box modifier that feeds commits back to the emitter@@ -170,37 +164,5 @@ feedback f box =   Cont $ \bio ->     with box $ \(Box c e) -> do-      fuse_ (emap f e) c+      glue c (mapE f e)       bio (Box c e)---- | an emitter post-processor that cons transformed emissions back into the emitter-feedbackE ::-  (MonadConc m) =>-  (a -> m (Maybe a)) ->-  Emitter m a ->-  Cont m (Emitter m a)-feedbackE f e =-  emergeM ((,) <$> pure e <*> fuseEmitM (emap f e))---- | turn a list into an emitter-fromListE :: [a] -> Cont IO (Emitter IO a)-fromListE xs = Cont $ queueEM' (eListC (Emitter . pure . Just <$> xs))--eListC :: (Monad m) => [Emitter m a] -> Committer m a -> m ()-eListC [] _ = pure ()-eListC (e:es) c = do-  x <- emit e-  case x of-    Nothing -> pure ()-    Just x' -> commit c x' >> eListC es c---- | turn a list into an emitter-toListE :: (MonadConc m) => Cont m (Emitter m a) -> m [a]-toListE ce = with ce (go [])-  where-    go xs e = do-      x <- emit e-      case x of-        Nothing -> pure (reverse xs)-        Just x' -> go (x':xs) e-

@@ -7,13 +7,15 @@ -- | A continuation type. module Box.Cont   ( Cont (..),+    runCont,+    (<$.>),+    (<*.>),     Cont_ (..),+    runCont_,   ) where -import Prelude-import Control.Applicative-import Control.Monad.IO.Class (MonadIO (liftIO))+import NumHask.Prelude hiding (STM, atomically)  -- | A continuation similar to ` Control.Monad.ContT` but where the result type is swallowed by an existential newtype Cont m a@@ -25,13 +27,11 @@   fmap f mx = Cont (\return_ -> mx `with` \x -> return_ (f x))  instance Applicative (Cont m) where-   pure r = Cont (\return_ -> return_ r)    mf <*> mx = Cont (\return_ -> mf `with` \f -> mx `with` \x -> return_ (f x))  instance Monad (Cont m) where-   return r = Cont (\return_ -> return_ r)    ma >>= f = Cont (\return_ -> ma `with` \a -> f a `with` \b -> return_ b)@@ -48,11 +48,14 @@   (<>) = liftA2 (<>)  instance (Functor m, Semigroup a, Monoid a) => Monoid (Cont m a) where-   mempty = pure mempty    mappend = (<>) +-- | finally run a continuation+runCont :: Cont m (m r) -> m r+runCont x = with x id+ -- | sometimes you have no choice but to void it up newtype Cont_ m a   = Cont_@@ -63,13 +66,11 @@   fmap f mx = Cont_ (\return_ -> mx `with_` \x -> return_ (f x))  instance Applicative (Cont_ m) where-   pure r = Cont_ (\return_ -> return_ r)    mf <*> mx = Cont_ (\return_ -> mf `with_` \f -> mx `with_` \x -> return_ (f x))  instance Monad (Cont_ m) where-   return r = Cont_ (\return_ -> return_ r)    ma >>= f = Cont_ (\return_ -> ma `with_` \a -> f a `with_` \b -> return_ b)@@ -86,7 +87,26 @@   (<>) = liftA2 (<>)  instance (Functor m, Semigroup a, Monoid a) => Monoid (Cont_ m a) where-   mempty = pure mempty    mappend = (<>)++-- | finally run a Cont_+runCont_ :: Cont_ m (m ()) -> m ()+runCont_ x = with_ x id++infixr 3 <$.>++-- | fmap over a continuation and then run the result.+--+-- The . position is towards the continuation+(<$.>) :: (a -> m r) -> Cont m a -> m r+(<$.>) f a = runCont (fmap f a)++infixr 3 <*.>++-- | fmap over a continuation and then run the result.+--+-- The . position is towards the continuation+(<*.>) :: Cont m (a -> m r) -> Cont m a -> m r+(<*.>) f a = runCont (f <*> a)

@@ -1,438 +0,0 @@-{-# LANGUAGE DeriveDataTypeable #-}-{-# LANGUAGE DeriveGeneric #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE MonoLocalBinds #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE RankNTypes #-}-{-# OPTIONS_GHC -Wall #-}-{-# OPTIONS_GHC -Wno-redundant-constraints #-}---- | An example of a Box for the command line.-module Box.Control-  ( ControlRequest (..),-    ControlResponse (..),-    Toggle (..),-    ControlBox,-    ControlBox_,-    ControlConfig (..),-    defaultControlConfig,-    consoleControlBox,-    consoleControlBox_,-    parseControlRequest,-    controlBox,-    controlBoxProcess,-    controlConsole,-    testBoxManual,-    testBoxAuto,-    beep,-    timeOut,-    timedRequests,-    testCatControl,-  )-where--import Box-import Control.Applicative-import Control.Concurrent.Async-import Control.Concurrent.Classy.IORef as C-import Control.Concurrent.Classy.STM.TVar as C-import Control.Lens hiding ((|>))-import Control.Monad-import Control.Monad.Conc.Class as C-import Control.Monad.STM.Class as C-import Control.Monad.Trans.Class-import qualified Data.Attoparsec.Text as A-import Data.Bool-import Data.Data-import Data.Functor-import Data.Maybe-import qualified Data.Text as Text-import Data.Text (Text)-import qualified Data.Text.IO as Text-import GHC.Generics-import qualified Streaming.Prelude as S-import System.IO-import System.Process.Typed-import Prelude---- | request ADT-data ControlRequest-  = Check -- check for existence-  | Start -- start (if not yet started) idempotent-  | Stop -- cancel (without shutting down) idempotent-  | Reset -- stop and start (potentially cancelling a previous instance)-  | Quit -- stop & shutdown-  deriving (Show, Read, Eq, Data, Typeable, Generic)---- | Parse command line requests-parseControlRequest :: A.Parser a -> A.Parser (Either ControlRequest a)-parseControlRequest pa =-  A.string "check" $> Left Check-    <|> A.string "start" $> Left Start-    <|> A.string "quit" $> Left Stop-    <|> A.string "reset" $> Left Reset-    <|> A.string "shutdown" $> Left Quit-    <|> (Right <$> pa)--data Toggle = On | Off deriving (Show, Read, Eq, Generic)---- | response ADT-data ControlResponse-  = ShuttingDown -- shutdown-  | Status (Toggle, Int) -- on/off and number restarts left-  | Info Text-  deriving (Show, Read, Eq, Generic)---- | A 'Box' that only communicates via 'ControlRequest' and 'ControlResponse'-type ControlBox_ m = (MonadConc m) => Cont m (Box (STM m) ControlResponse ControlRequest)---- | A 'Box' that communicates via 'ControlRequest' and 'ControlResponse' or an underlying typed-channel-type ControlBox a b m = (MonadConc m) => Cont m (Box (STM m) (Either ControlResponse a) (Either ControlRequest b))---- |-data ControlConfig-  = ControlConfig-      { -- | maximum number of starts allowed-        starts :: Int,-        -- | whether to start automatically-        autoStart :: Bool,-        -- | whether to rerun with a delay if the action dies-        autoRestart :: Maybe Double,-        -- | logging debug Info-        debug :: Bool-      }-  deriving (Show, Eq, Ord)---- | Default is one start, manual start and no autorestart.-defaultControlConfig :: ControlConfig-defaultControlConfig = ControlConfig 1 False Nothing False---- | a command-line control box.-consoleControlBox :: ControlBox Text Text IO-consoleControlBox =-  Box-    <$> ( contramap (Text.pack . show)-            <$> (cStdout 1000 :: Cont IO (Committer (STM IO) Text))-        )-    <*> ( emap (pure . either (const Nothing) Just)-            <$> ( eParse (parseControlRequest A.takeText)-                    <$> eStdin 1000-                )-        )---- | a command-line control box.-consoleControlBox_ :: ControlBox_ IO-consoleControlBox_ =-  bmap (pure . Just . Left) (pure . either Just (const Nothing))-    <$> consoleControlBox--data ControlBoxState a = CBS {actionThread :: Maybe (Async ()), restartsLeft :: Int}---- | an effect that can be started, stopped and restarted (a limited number of times)-controlBox ::-  ControlConfig ->-  IO a ->-  Box (STM IO) ControlResponse ControlRequest ->-  IO ()-controlBox (ControlConfig restarts' autostart autorestart debug') app (Box c e) = do-  info "controlBox"-  ref <- C.newIORef (CBS Nothing restarts')-  shut <- atomically $ newTVar False-  when autostart (info "autostart" >> start ref shut)-  info "race_"-  race_-    (go ref shut)-    (shutCheck shut)-  cancelThread ref-  info "controlBox end"-  where-    cancelThread r = do-      info "cancelThread"-      (CBS a n) <- readIORef r-      maybe (info "no thread found" >> pure ()) (\x -> cancel x >> info "thread cancelled") a-      writeIORef r (CBS Nothing n)-    shutCheck s = do-      info "shutCheck"-      atomically $ check =<< readTVar s-      info "shutCheck signal received"-    status r = do-      info "status"-      s <- C.readIORef r-      C.atomically-        ( void $-            commit-              c-              (Status (bool Off On (isJust (actionThread s)), restartsLeft s))-        )-    loopApp r s app' = do-      info "loopApp"-      _ <- app'-      info "post app'"-      checkRestarts r s-      info "maybe restarting"-      maybe (pure ()) (\t -> sleep t >> dec r >> loopApp r s app') autorestart-    dec r = do-      info "dec"-      cfg@(CBS _ n) <- readIORef r-      writeIORef r (cfg {restartsLeft = n - 1})-    start r s = do-      info "start"-      (CBS a _) <- readIORef r-      when (isNothing a) $ do-        a' <--          async-            ( do-                dec r-                loopApp r s app-                cfg <- readIORef r-                writeIORef r (cfg {actionThread = Nothing})-            )-        link a'-        cfg <- readIORef r-        writeIORef r (cfg {actionThread = Just a'})-    stop r s = do-      info "stop"-      cancelThread r-      checkRestarts r s-    info t = bool (pure ()) (void $ commit (liftC c) $ Info t) debug'-    shutdown = do-      info "shutDown"-      void $ commit (liftC c) ShuttingDown-    checkRestarts r s = do-      info "check restarts"-      (CBS _ n) <- C.readIORef r-      bool-        ( do-            atomically $ writeTVar s True-            shutdown-        )-        (pure ())-        (n > 0)-    go r s = do-      info "go"-      status r-      msg <- C.atomically $ emit e-      case msg of-        Nothing -> go r s-        Just msg' ->-          case msg' of-            Check ->-              go r s-            Start -> do-              start r s-              go r s-            Stop -> do-              stop r s-              go r s-            Quit -> stop r s >> shutdown-            Reset -> stop r s >> start r s >> go r s---- control box process-data CBP = CBP {listenThread :: Maybe (Async ()), process :: Maybe (Process Handle Handle ()), restarts :: Int}---- | an effect that can be started, stopped and restarted (a limited number of times)-controlBoxProcess ::-  ControlConfig ->-  ProcessConfig Handle Handle () ->-  Box (STM IO) (Either ControlResponse Text) (Either ControlRequest Text) ->-  IO ()-controlBoxProcess (ControlConfig restarts' autostart _ debug') pc (Box c e) = do-  info "controlBoxProcess"-  ref <- C.newIORef (CBP Nothing Nothing restarts')-  shut <- atomically $ C.newTVar False-  when autostart (info "autostart" >> start ref shut)-  info "race_"-  race_-    (go ref shut)-    (shutCheck shut)-  cancelThread ref-  info "controlBoxProcess end"-  where-    cancelThread r = do-      info "cancelThread"-      a <- readIORef r-      maybe (info "no listener on cancelThread") (\x -> cancel x >> info "listener cancelled") (listenThread a)-      maybe (info "no process on cancelThread") (\x -> stopProcess x >> info "process cancelled") (process a)-      writeIORef r (CBP Nothing Nothing (restarts a))-    shutCheck s = do-      info "shutCheck"-      atomically $ check =<< readTVar s-      info "shutCheck signal received"-    status r = do-      info "status"-      a <- C.readIORef r-      C.atomically-        ( void $-            commit-              c-              (Left $ Status (bool Off On (isJust (process a)), restarts a))-        )-    loopApp r _ = do-      info "loopApp"-      p' <- startProcess pc-      a <- readIORef r-      when (isJust (process a)) (info "eeek, a process ref has been overwritten")-      when (isJust (listenThread a)) (info "eeek, a listener ref has been overwritten")-      info "process is up"-      wo <- async (lloop0 (getStdout p'))-      writeIORef r (CBP (Just wo) (Just p') (restarts a))-      info "listener is up"-      link wo-    lloop0 o = do-      b <- hIsEOF o-      when (not b) (checkOutH o >> lloop0 o)-    checkOutH o = do-      info "waiting for process output"-      t <- Text.hGetLine o-      info ("received: " <> t)-      C.atomically $ void $ commit (contramap Right c) t-    dec r = do-      info "dec"-      a <- readIORef r-      writeIORef r (a {restarts = restarts a - 1})-    start r s = do-      info "start"-      a <- readIORef r-      when (isNothing (process a)) $ do-        dec r-        loopApp r s-    stop r s = do-      info "stop"-      cancelThread r-      checkRestarts r s-    info t = bool (pure ()) (void $ commit (liftC c) $ Left (Info t)) debug'-    shutdown = do-      info "shutDown"-      void $ commit (liftC c) (Left ShuttingDown)-    checkRestarts r s = do-      info "check restarts"-      n <- restarts <$> C.readIORef r-      bool-        ( do-            atomically $ writeTVar s True-            shutdown-        )-        (pure ())-        (n > 0)-    writeIn r t = do-      info ("writeIn: " <> t)-      p <- process <$> C.readIORef r-      maybe-        (info "no stdin available")-        (\i -> hPutStrLn (getStdin i) (Text.unpack t) >> hFlush (getStdin i))-        p-    go r s = do-      info "go"-      status r-      msg <- C.atomically $ emit e-      case msg of-        Nothing -> go r s-        Just msg' ->-          case msg' of-            Left Check ->-              go r s-            Left Start -> do-              start r s-              go r s-            Left Stop -> do-              stop r s-              go r s-            Left Quit -> stop r s >> shutdown-            Left Reset -> stop r s >> start r s >> go r s-            Right t -> writeIn r t >> go r s--controlConsole ::-  Cont IO (Box (STM IO) (Either ControlResponse Text) (Either ControlRequest Text))-controlConsole =-  Box-    <$> ( contramap (either (("Response: " <>) . Text.pack . show) id)-            <$> (cStdout 1000 :: Cont IO (Committer (STM IO) Text))-        )-    <*> ( fmap (either (Right . ("parse error: " <>)) id)-            . eParse (parseControlRequest A.takeText) <$> eStdin 1000-        )---- | action for testing-beep :: Int -> Int -> Double -> IO ()-beep m x s = when (x <= m) (sleep s >> Text.putStrLn ("beep " <> Text.pack (show x)) >> beep m (x + 1) s)---- | A box with a self-destruct timer.-timeOut :: Double -> ControlBox m a b-timeOut t =-  Box <$> mempty <*> ((lift (sleep t) >> S.yield (Left Quit)) & toEmit)---- | a canned ControlRequest emitter with delays-timedRequests ::-  (MonadConc m) =>-  [(ControlRequest, Double)] ->-  Cont m (Emitter (STM m) ControlRequest)-timedRequests xs = toEmit $ foldr (>>) (pure ()) $ (\(a, t) -> lift (sleep t) >> S.yield a) <$> xs---- | manual testing--- > testBoxManual (ControlConfig 1 True (Just 0.5) False) 2.3 (beep 3 1 0.5)--- Status (On,0)--- beep 1--- beep 2--- beep 3--- Left ShutDown-testBoxManual :: ControlConfig -> Double -> IO () -> IO ()-testBoxManual cfg t effect =-  with-    ( bmap (pure . Just . Left) (pure . either Just (const Nothing))-        <$> consoleControlBox <> timeOut t-    )-    (controlBox cfg effect)---- | auto testing--- FIXME: Doesn't work with doctest--- > testBoxAuto (ControlConfig 5 True (Just 0.2) False) 5 [(Check, 0.1), (Start,0.1), (Stop,1), (Start, 0.1), (Check, 0.1), (Reset,0.1)] (beep 2 1 1)--- Left (Status (On,5))--- Left (Status (On,4))--- Left (Status (On,4))--- beep 1--- Left (Status (Off,4))--- Left (Status (On,4))--- Left (Status (On,3))--- Left (Status (On,2))--- beep 1--- beep 2--- beep 1--- Left ShuttingDown------ testBoxAuto (ControlConfig 1 True (Just 0.5) False) 3 [(Reset,1.1), (Quit, 1)] (beep 3 1 1)--- Left (Status (On,1))--- beep 1--- Left ShuttingDown--- Left (Status (On,-1))-testBoxAuto :: ControlConfig -> Double -> [(ControlRequest, Double)] -> IO () -> IO ()-testBoxAuto cfg t xs effect =-  with-    ( bmap (pure . Just . Left) (pure . either Just (const Nothing))-        <$> ( consoleControlBox-                <> timeOut t-                <> (Box <$> mempty <*> (fmap Left <$> timedRequests xs))-            )-    )-    (controlBox cfg effect)--cannedCat :: ProcessConfig Handle Handle ()-cannedCat =-  setStdin createPipe-    $ setStdout createPipe-    $ setStderr-      closed-      "cat"---- > testCatControl defaultControlConfig--- Left (Status (Off,1))--- s---Left (Status (On,0))---hello cat---Left (Status (On,0))---Right "hello cat"---x---Left ShuttingDown---Left ShuttingDown-testCatControl :: ControlConfig -> IO ()-testCatControl cfg = with controlConsole (controlBoxProcess cfg cannedCat)

@@ -1,5 +1,6 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE LambdaCase #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-}@@ -11,23 +12,22 @@ -- | `emit` module Box.Emitter   ( Emitter (..),-    liftE,-    emap,-    keeps,-    eRead,-    eParse,+    mapE,+    readE,+    readE_,+    parseE,+    parseE_,+    premapE,+    postmapE,+    postmapM,+    toListE,+    unlistE,+    stateE,   ) where -import Prelude-import Control.Applicative-import Control.Monad-import Control.Monad.Conc.Class as C import qualified Data.Attoparsec.Text as A-import Data.Functor.Constant-import Data.Monoid-import qualified Data.Text as Text-import Data.Text (Text)+import NumHask.Prelude  -- | an `Emitter` "emits" values of type a. A Source & a Producer (of a's) are the two other alternative but overloaded metaphors out there. --@@ -37,17 +37,18 @@       { emit :: m (Maybe a)       } +instance MFunctor Emitter where+  hoist nat (Emitter e) = Emitter (nat e)+ instance (Functor m) => Functor (Emitter m) where   fmap f m = Emitter (fmap (fmap f) (emit m))  instance (Applicative m) => Applicative (Emitter m) where-   pure r = Emitter (pure (pure r))    mf <*> mx = Emitter ((<*>) <$> emit mf <*> emit mx)  instance (Monad m) => Monad (Emitter m) where-   return r = Emitter (return (return r))    m >>= f =@@ -58,7 +59,6 @@         Just a -> emit (f a)  instance (Monad m, Alternative m) => Alternative (Emitter m) where-   empty = Emitter (pure Nothing)    x <|> y =@@ -69,7 +69,6 @@         Just a -> pure (Just a)  instance (Alternative m, Monad m) => MonadPlus (Emitter m) where-   mzero = empty    mplus = (<|>)@@ -78,18 +77,13 @@   (<>) = (<|>)  instance (Alternative m, Monad m) => Monoid (Emitter m a) where-   mempty = empty    mappend = (<>) --- | lift an STM emitter-liftE :: (MonadConc m) => Emitter (STM m) a -> Emitter m a-liftE = Emitter . atomically . emit- -- | like a monadic mapMaybe. (See [witherable](https://hackage.haskell.org/package/witherable))-emap :: (Monad m) => (a -> m (Maybe b)) -> Emitter m a -> Emitter m b-emap f e = Emitter go+mapE :: (Monad m) => (a -> m (Maybe b)) -> Emitter m a -> Emitter m b+mapE f e = Emitter go   where     go = do       a <- emit e@@ -101,38 +95,98 @@             Nothing -> go             Just fa' -> pure (Just fa') --- | prism handler-keeps ::-  (Monad m) =>-  -- |-  ((b -> Constant (First b) b) -> (a -> Constant (First b) a)) ->-  -- |-  Emitter m a ->-  Emitter m b-keeps k (Emitter emit_) = Emitter emit_'-  where-    emit_' = do-      ma <- emit_-      case ma of-        Nothing -> return Nothing-        Just a ->-          case match a of-            Nothing -> emit_'-            Just b -> return (Just b)-    match = getFirst . getConstant . k (Constant . First . Just)+-- | parse emitter which returns the original text on failure+parseE :: (Functor m) => A.Parser a -> Emitter m Text -> Emitter m (Either Text a)+parseE parser e = (\t -> either (const $ Left t) Right (A.parseOnly parser t)) <$> e --- | attoparsec parse emitter-eParse :: (Functor m) => A.Parser a -> Emitter m Text -> Emitter m (Either Text a)-eParse parser e = either (Left . Text.pack) Right . A.parseOnly parser <$> e+-- | no error-reporting parsing+parseE_ :: (Monad m) => A.Parser a -> Emitter m Text -> Emitter m a+parseE_ parser = mapE (pure . (either (const Nothing) Just)) . parseE parser --- | read parse emitter-eRead ::+-- | read parse emitter, returning the original string on error+readE ::   (Functor m, Read a) =>   Emitter m Text ->   Emitter m (Either Text a)-eRead = fmap $ parsed . Text.unpack+readE = fmap $ parsed . unpack   where     parsed str =       case reads str of         [(a, "")] -> Right a-        _ -> Left (Text.pack str)+        _ -> Left (pack str)++-- | no error-reporting reading+readE_ ::+  (Monad m, Read a) =>+  Emitter m Text ->+  Emitter m a+readE_ = mapE (pure . (either (const Nothing) Just)) . readE++-- | adds a pre-emit monadic action to the emitter+premapE ::+  (Applicative m) =>+  (Emitter m a -> m ()) ->+  Emitter m a ->+  Emitter m a+premapE f e = Emitter $ f e *> emit e++-- | adds a post-emit monadic action to the emitter+postmapE ::+  (Monad m) =>+  (Emitter m a -> m ()) ->+  Emitter m a ->+  Emitter m a+postmapE f e = Emitter $ do+  r <- emit e+  f e+  pure r++-- | add a post-emit monadic action on the emitted value (if there was any)+postmapM ::+  (Monad m) =>+  (a -> m ()) ->+  Emitter m a ->+  Emitter m a+postmapM f e = Emitter $ do+  r <- emit e+  case r of+    Nothing -> pure Nothing+    Just r' -> do+      f r'+      pure (Just r')++-- | turn an emitter into a list+toListE :: (Monad m) => Emitter m a -> m [a]+toListE e = go [] e+  where+    go xs e' = do+      x <- emit e'+      case x of+        Nothing -> pure (reverse xs)+        Just x' -> go (x' : xs) e'++-- | emit from a StateT list+--+-- This compiles but is an infinite "a" emitter:+--+-- let e1 = hoist (flip evalStateT ["a", "b"::Text]) stateE :: Emitter IO Text+stateE :: (Monad m) => Emitter (StateT [a] m) a+stateE = Emitter $ do+  xs' <- get+  case xs' of+    [] -> pure Nothing+    (x : xs'') -> do+      put xs''+      pure $ Just x++-- | convert a list emitter to a Stateful element emitter+unlistE :: (Monad m) => Emitter m [a] -> Emitter (StateT [a] m) a+unlistE es = mapE unlistS (hoist lift es)+  where+    unlistS xs = do+      rs <- get+      case rs <> xs of+        [] -> pure Nothing+        (x : xs') -> do+          put xs'+          pure (Just x)

@@ -1,216 +1,125 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE NoImplicitPrelude #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE NoImplicitPrelude #-} {-# OPTIONS_GHC -Wall #-} {-# OPTIONS_GHC -fno-warn-type-defaults #-}  -- | IO actions module Box.IO-  ( cStdin_,-    cStdin,-    cStdin',-    eStdin,+  ( fromStdin,+    toStdout,+    fromStdinN,+    toStdoutN,     readStdin,-    eStdout_,-    eStdout,-    eStdoutM,-    eStdout',-    cStdout,-    cStdout',-    eStdin',     showStdout,-    getLine,-    putLine,-    consolePlug,-    emitLines,-    commitLines,-    cCRef,-    cCRefM,-    toListM,-    getCommissions,-    getEmissions,-    fileEmitter,-    fileCommitter,-    appendCommitter,+    handleE,+    handleC,+    cRef,+    eRef,+    fileE,+    fileWriteC,+    fileAppendC,   ) where -import Box.Box import Box.Committer+import Box.Connectors import Box.Cont import Box.Emitter-import Box.Plugs-import Box.Stream-import Box.Transducer 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 qualified Data.Text as Text-import qualified Data.Text.IO as Text-import qualified Streaming.Prelude as S-import System.IO hiding (getLine)-import Protolude hiding (getLine, STM)+import Data.Text.IO (hGetLine)+import NumHask.Prelude hiding (STM) +-- $setup+-- >>> :set -XOverloadedStrings+ -- * console --- | a single stdin committer action-cStdin_ :: Committer (STM IO) Text -> IO ()-cStdin_ c = do-  a <- Text.getLine-  void $ C.atomically $ commit c a+-- | emit Text from stdin inputs+--+-- >>> :t emit fromStdin+-- emit fromStdin :: IO (Maybe Text)+fromStdin :: Emitter IO Text+fromStdin = Emitter $ Just <$> NumHask.Prelude.getLine --- | a finite stdin committer action-cStdin :: Int -> Committer (STM IO) Text -> IO ()-cStdin n c = replicateM_ n (cStdin_ c)+-- | commit to stdout+--+-- >>> commit toStdout ("I'm committed!" :: Text)+-- I'm committed!+-- True+toStdout :: Committer IO Text+toStdout = Committer $ \a -> putStrLn a >> pure True --- | a forever stdin committer action-cStdin' :: Committer (STM IO) Text -> IO ()-cStdin' = forever . cStdin_+-- | finite console emitter+fromStdinN :: Int -> Cont IO (Emitter IO Text)+fromStdinN n = source n NumHask.Prelude.getLine --- | a Cont stdin emitter-eStdin :: Int -> Cont IO (Emitter (STM IO) Text)-eStdin n = cStdin n & emitPlug+-- | finite console committer+toStdoutN :: Int -> Cont IO (Committer IO Text)+toStdoutN n = sink n putStrLn  -- | read from console, throwing away read errors-readStdin :: Read a => Cont IO (Emitter (STM IO) a)-readStdin = emap (pure . either (const Nothing) Just) . eRead <$> eStdin 1000---- | a single stdout emitter action-eStdout_ :: Emitter (STM IO) Text -> IO ()-eStdout_ e = do-  a <- C.atomically $ emit e-  case a of-    Nothing -> pure ()-    Just a' -> Text.putStrLn a'---- | a single stdout emitter action-eStdoutM_ :: Emitter IO Text -> IO ()-eStdoutM_ e = do-  a <- emit e-  case a of-    Nothing -> pure ()-    Just a' -> Text.putStrLn a'---- | a finite stdout emitter action-eStdout :: Int -> Emitter (STM IO) Text -> IO ()-eStdout n = replicateM_ n . eStdout_---- | a finite stdout emitter action-eStdoutM :: Int -> Emitter IO Text -> IO ()-eStdoutM n = replicateM_ n . eStdoutM_---- | a forever stdout emitter action-eStdout' :: Emitter (STM IO) Text -> IO ()-eStdout' = forever . eStdout_---- | a Cont stdout committer-cStdout :: Int -> Cont IO (Committer (STM IO) Text)-cStdout n = eStdout n & commitPlug---- | a Cont stdout committer-cStdout' :: Cont IO (Committer (STM IO) Text)-cStdout' = eStdout' & commitPlug---- | a Cont stdin emitter-eStdin' :: Cont IO (Emitter (STM IO) Text)-eStdin' = cStdin' & emitPlug+readStdin :: Read a => Emitter IO a+readStdin = mapE (pure . either (const Nothing) Just) . readE $ fromStdin  -- | show to stdout-showStdout :: Show a => Cont IO (Committer (STM IO) a)-showStdout = contramap (Text.pack . show) <$> (cStdout 1000 :: Cont IO (Committer (STM IO) Text))---- | Get a single line from a handle.-getLine_ :: Handle -> Committer (STM IO) Text -> IO ()-getLine_ h c = do-  a <- Text.hGetLine h-  void $ C.atomically $ commit c a---- | Get lines from a handle forever.-getLine :: Handle -> Committer (STM IO) Text -> IO ()-getLine h = forever . getLine_ h---- | Put a single line to a handle.-putLine_ :: Handle -> Emitter (STM IO) Text -> IO ()-putLine_ h e = do-  a <- C.atomically $ emit e-  case a of-    Nothing -> pure ()-    Just a' -> Text.hPutStrLn h a'---- | a forever getLine committer action-putLine :: Handle -> Emitter (STM IO) Text -> IO ()-putLine h = forever . putLine_ h-+showStdout :: Show a => Committer IO a+showStdout = contramap show toStdout --- | console box--- > etc () (Trans $ \s -> s & S.takeWhile (/="q") & S.map ("echo: " <>)) (console 5)-consolePlug :: Int -> Cont IO (Box (STM IO) Text Text)-consolePlug n = boxPlug (eStdout n) (cStdin n)+-- * handle operations --- * file operations -- | Emits lines of Text from a handle.-emitLines :: Handle -> Emitter IO Text-emitLines h = Emitter $ do-  l :: (Either IOException Text) <- try (Text.hGetLine h)+handleE :: Handle -> Emitter IO Text+handleE h = Emitter $ do+  l :: (Either IOException Text) <- try (hGetLine h)   pure $ case l of     Left _ -> Nothing     Right a -> bool (Just a) Nothing (a == "")  -- | Commit lines of Text to a handle.-commitLines :: Handle -> Committer IO Text-commitLines h = Committer $ \a -> do-  Text.hPutStrLn h a+handleC :: Handle -> Committer IO Text+handleC h = Committer $ \a -> do+  hPutStrLn h a   pure True -fileEmitter :: FilePath -> Cont IO (Emitter IO Text)-fileEmitter fp = Cont $ \eio -> withFile fp ReadMode (eio . emitLines)+-- | Emits lines of Text from a file.+fileE :: FilePath -> Cont IO (Emitter IO Text)+fileE fp = Cont $ \eio -> withFile fp ReadMode (eio . handleE) -fileCommitter :: FilePath -> Cont IO (Committer IO Text)-fileCommitter fp = Cont $ \cio -> withFile fp WriteMode (cio . commitLines)+-- | Commits lines of Text to a file.+fileWriteC :: FilePath -> Cont IO (Committer IO Text)+fileWriteC fp = Cont $ \cio -> withFile fp WriteMode (cio . handleC) -appendCommitter :: FilePath -> Cont IO (Committer IO Text)-appendCommitter fp = Cont $ \cio -> withFile fp AppendMode (cio . commitLines)+-- | Commits lines of Text, appending to a file.+fileAppendC :: FilePath -> Cont IO (Committer IO Text)+fileAppendC fp = Cont $ \cio -> withFile fp AppendMode (cio . handleC) --- * 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+-- | commit to a list IORef+cRef :: (C.MonadConc m) => m (Committer m a, m [a])+cRef = do   ref <- C.newIORef []-  let c =-        toCommitFold $-          L.FoldM (\x a -> C.modifyIORef x (a :) >> pure x) (pure ref) (const $ pure ())+  let c = Committer $ \a -> do+        C.modifyIORef ref (a :)+        pure True   let res = reverse <$> C.readIORef ref-  pure (ref, c, res)---- | commit to a monoidal CRef-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 res = C.readIORef ref-  pure (ref, c, res)---- | fold an emitter through a transduction, committing to a list-toListM :: (C.MonadConc m) => Cont m (Emitter (C.STM m) a) -> s -> Transducer s a b -> m ([b], s)-toListM e s t = do-  (_, c, res) <- cCRef-  r <- etc s t (Box <$> c <*> e)-  (,) <$> res <*> pure r---- | get all commissions as a list-getCommissions :: (C.MonadConc m) => Cont m (Emitter (C.STM m) a) -> s -> Transducer s a b -> m [b]-getCommissions e s t = fst <$> toListM e s t+  pure (c, res) --- | get all emissions-getEmissions :: (C.MonadConc m) => Int -> Cont m (Emitter (C.STM m) a) -> m [a]-getEmissions n e = fst <$> toListM e () (Transducer $ S.take n)+-- | emit from a list IORef+eRef :: (C.MonadConc m) => [a] -> m (Emitter m a)+eRef xs = do+  ref <- C.newIORef xs+  let e = Emitter $ do+        as <- C.readIORef ref+        case as of+          [] -> pure Nothing+          (x : xs') -> do+            C.writeIORef ref xs'+            pure $ Just x+  pure e

@@ -1,52 +0,0 @@-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE RankNTypes #-}-{-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE TypeFamilies #-}-{-# OPTIONS_GHC -Wall #-}-{-# OPTIONS_GHC -fno-warn-type-defaults #-}---- | plugs--- box continuations-module Box.Plugs-  ( commitPlug,-    emitPlug,-    emitPlugM,-    boxPlug,-    boxForgetPlug,-  )-where--import Prelude-import Box.Box-import Box.Committer-import Box.Cont-import Box.Emitter-import Box.Queue-import GHC.Conc---- * plugs---- | hook an emitter action to a queue, creating a committer continuation-commitPlug :: (Emitter STM a -> IO ()) -> Cont IO (Committer STM a)-commitPlug eio = Cont $ \cio -> queueC cio eio---- | hook a committer action to a queue, creating an emitter continuation-emitPlug :: (Committer STM a -> IO r) -> Cont IO (Emitter STM a)-emitPlug cio = Cont $ \eio -> queueE' cio eio---- | hook a committer action to a queue, creating an emitter continuation-emitPlugM :: (Committer IO a -> IO r) -> Cont IO (Emitter IO a)-emitPlugM cio = Cont $ \eio -> queueEM' cio eio---- | create a double-queued box plug-boxPlug ::-  (Emitter STM a -> IO ()) ->-  (Committer STM b -> IO ()) ->-  Cont IO (Box STM a b)-boxPlug eio cio = Box <$> commitPlug eio <*> emitPlug cio---- | create a box plug from a box action.  Caution: implicitly, this (has to) forget interactions between emitter and committer in the one action (and it does so silently).  These forgotten interactions are typically those that create races-boxForgetPlug :: (Box STM b a -> IO ()) -> Cont IO (Box STM a b)-boxForgetPlug bio = boxPlug (bio . Box mempty) (bio . (`Box` mempty))

@@ -12,35 +12,31 @@ -- Follows [pipes-concurrency](https://hackage.haskell.org/package/pipes-concurrency) module Box.Queue   ( Queue (..),-    queue,     queueC,-    queueC',     queueE,-    queueE',-    queueCM,-    queueCM',-    queueEM,-    queueEM',     waitCancel,     ends,-    withQ,     withQE,     withQC,     toBox,+    toBoxM,+    liftB,     concurrentlyLeft,     concurrentlyRight,+    fromAction,+    fuseActions,   ) where -import Prelude import Box.Box import Box.Committer+import Box.Cont import Box.Emitter-import Control.Applicative import Control.Concurrent.Classy.Async as C import Control.Concurrent.Classy.STM as C import Control.Monad.Catch as C import Control.Monad.Conc.Class as C+import NumHask.Prelude hiding (STM, atomically)  -- | 'Queue' specifies how messages are queued data Queue a@@ -111,125 +107,65 @@       seal     ) +-- | turn a queue into a box (and a seal), and lift from stm to the underlying monad. toBoxM ::   (MonadConc m) =>   Queue a ->   m (Box m a a, m ()) toBoxM q = do-  (i, o) <- atomically $ ends q-  sealed <- atomically $ newTVarN "sealed" False-  let seal = atomically $ writeTVar sealed True-  pure-    ( Box-        (Committer (atomically . writeCheck sealed i))-        (Emitter (atomically $ readCheck sealed o)),-      seal-    )+  (b, s) <- atomically $ toBox q+  pure (liftB b, atomically s)  -- | wait for the first action, and then cancel the second waitCancel :: (MonadConc m) => m b -> m a -> m b waitCancel a b =-  withAsync a $ \a' ->-    withAsync b $ \b' -> do-      a'' <- wait a'-      cancel b'+  C.withAsync a $ \a' ->+    C.withAsync b $ \b' -> do+      a'' <- C.wait a'+      C.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+  C.withAsync left $ \a ->+    C.withAsync right $ \_ ->+      C.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+  C.withAsync left $ \_ ->+    C.withAsync right $ \b ->+      C.wait b  -- | connect a committer and emitter action via spawning a queue, and wait for both to complete.-withQ ::-  (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 (l, r)-withQ q spawner cio eio =-  C.bracket-    (atomically $ spawner q)-    (\(_, seal) -> atomically seal)-    ( \(box, seal) ->-        concurrently-          (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 committer to complete. withQC ::   (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 l-withQC q spawner cio eio =-  C.bracket-    (atomically $ spawner q)-    (\(_, seal) -> atomically seal)-    ( \(box, 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.-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)-withQM q spawner cio eio =+  m l+withQC q spawner cio eio =   C.bracket     (spawner q)     snd     ( \(box, seal) ->-        concurrently+        concurrentlyLeft           (cio (committer box) `C.finally` seal)           (eio (emitter box) `C.finally` seal)     )  -- | connect a committer and emitter action via spawning a queue, and wait for both to complete.-withQEM ::+withQE ::   (MonadConc m) =>   Queue a ->   (Queue a -> m (Box m a a, m ())) ->   (Committer m a -> m l) ->   (Emitter m a -> m r) ->   m r-withQEM q spawner cio eio =+withQE q spawner cio eio =   C.bracket     (spawner q)     snd@@ -239,149 +175,33 @@           (eio (emitter box) `C.finally` seal)     ) --- | connect a committer and emitter action via spawning a queue, and wait for both to complete.-withQCM ::-  (MonadConc m) =>-  Queue a ->-  (Queue a -> m (Box m a a, m ())) ->-  (Committer m a -> m l) ->-  (Emitter m a -> m r) ->-  m l-withQCM q spawner cio eio =-  C.bracket-    (spawner q)-    snd-    ( \(box, seal) ->-        concurrentlyLeft-          (cio (committer box) `C.finally` seal)-          (eio (emitter box) `C.finally` seal)-    )----- | create an unbounded queue-queue ::-  (MonadConc m) =>-  (Committer (STM m) a -> m l) ->-  (Emitter (STM m) a -> m r) ->-  m (l, r)-queue = withQ Unbounded toBox- -- | create an unbounded queue, returning the emitter result-queueE ::-  (MonadConc m) =>-  (Committer (STM m) a -> m l) ->-  (Emitter (STM m) a -> m r) ->-  m r-queueE cm em = snd <$> withQ Unbounded toBox cm em--queueE' ::-  (MonadConc m) =>-  (Committer (STM m) a -> m l) ->-  (Emitter (STM m) a -> m r) ->-  m r-queueE' cm em = withQE Unbounded toBox cm em---- | create an unbounded queue, returning the committer result queueC ::   (MonadConc m) =>-  (Committer (STM m) a -> m l) ->-  (Emitter (STM m) a -> m r) ->-  m l-queueC cm em = fst <$> withQ Unbounded toBox cm em---- | create an unbounded queue, returning the committer result-queueC' ::-  (MonadConc m) =>-  (Committer (STM m) a -> m l) ->-  (Emitter (STM m) a -> m r) ->-  m l-queueC' cm em = withQC Unbounded toBox cm em---- | create an unbounded queue, returning the emitter result-queueCM ::-  (MonadConc m) =>   (Committer m a -> m l) ->   (Emitter m a -> m r) ->   m l-queueCM cm em = fst <$> withQM Unbounded toBoxM cm em---- | create an unbounded queue, returning the emitter result-queueCM' ::-  (MonadConc m) =>-  (Committer m a -> m l) ->-  (Emitter m a -> m r) ->-  m l-queueCM' cm em = withQCM Unbounded toBoxM cm em---- | create an unbounded queue, returning the emitter result-queueEM ::-  (MonadConc m) =>-  (Committer m a -> m l) ->-  (Emitter m a -> m r) ->-  m r-queueEM cm em = snd <$> withQM Unbounded toBoxM cm em+queueC cm em = withQC Unbounded toBoxM cm em  -- | create an unbounded queue, returning the emitter result-queueEM' ::+queueE ::   (MonadConc m) =>   (Committer m a -> m l) ->   (Emitter m a -> m r) ->   m r-queueEM' cm em = withQEM Unbounded toBoxM 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 - not ok here--- > 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 - okhere--- > 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.+queueE cm em = withQE Unbounded toBoxM cm em -I replaced `snd <$> withQ` with `withQE`+-- | lift a box from STM+liftB :: (MonadConc m) => Box (STM m) a b -> Box m a b+liftB (Box c e) = Box (hoist atomically c) (hoist atomically e) --}+-- | turn a box action into a box continuation+fromAction :: (MonadConc m) => (Box m a b -> m r) -> Cont m (Box m b a)+fromAction baction = Cont $ fuseActions baction --- subbing withQE fixes!--- withQ Unbounded toBox (\c -> (withQE Unbounded toBox (\c' -> cStdin_ c' *> cStdin_ c') ((\e -> (fromStream . S.takeWhile (/="q") . toStream $ e) c)))) eStdout_+-- | connect up two box actions via two queues+fuseActions :: (MonadConc m) => (Box m a b -> m r) -> (Box m b a -> m r') -> m r'+fuseActions abm bam = do+  (Box ca ea, _) <- toBoxM Unbounded+  (Box cb eb, _) <- toBoxM Unbounded+  concurrentlyRight (abm (Box ca eb)) (bam (Box cb ea))

@@ -1,88 +0,0 @@-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE RankNTypes #-}-{-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE TypeFamilies #-}-{-# OPTIONS_GHC -Wall #-}-{-# OPTIONS_GHC -fno-warn-type-defaults #-}---- | Streaming functionality-module Box.Stream-  ( toStream,-    fromStream,-    toCommit,-    toCommitFold,-    toCommitSink,-    toEmit,-    queueStream,-    toStreamM,-    fromStreamM,-  )-where--import Prelude-import Box.Committer-import Box.Cont-import Box.Emitter-import Box.Queue-import qualified Control.Foldl as L-import Control.Monad-import Control.Monad.Conc.Class as C-import Streaming (Of (..), Stream)-import qualified Streaming.Prelude as S---- * streaming---- | create a committer from a stream consumer-toCommit :: (MonadConc m) => (Stream (Of a) m () -> m r) -> Cont m (Committer (STM m) a)-toCommit f =-  Cont (\c -> queueC' c (\(Emitter o) -> f . toStream . Emitter $ o))---- | create a committer from a fold-toCommitFold :: (MonadConc m) => L.FoldM m a () -> Cont m (Committer (STM m) a)-toCommitFold f = toCommit (fmap S.snd' . L.impurely S.foldM f)---- | create a committer from a sink-toCommitSink :: (MonadConc m) => (a -> m ()) -> Cont m (Committer (STM m) a)-toCommitSink sink = toCommitFold (L.FoldM step begin done)-  where-    step x a = do-      sink a-      pure x-    begin = pure ()-    done = pure---- | create an emitter from a stream-toEmit :: (MonadConc m) => Stream (Of a) m () -> Cont m (Emitter (STM m) a)-toEmit s = Cont (queueE' (fromStream s))---- | insert a queue into a stream (left biased collapse)--- todo: look at biases-queueStream ::-  (MonadConc m) => Stream (Of a) m () -> Cont m (Stream (Of a) m ())-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 ()-toStream e = toStreamM (liftE e)---- | turn an emitter into a stream-toStreamM :: (MonadConc m) => Emitter m a -> Stream (Of a) m ()-toStreamM e = S.untilRight getNext-  where-    getNext = maybe (Right ()) Left <$> emit e---- | turn a stream into a committer-fromStream :: (MonadConc m) => Stream (Of b) m () -> Committer (STM m) b -> m ()-fromStream s c = fromStreamM s (liftC c)---- | turn a stream into a committer-fromStreamM :: (MonadConc m) => Stream (Of b) m () -> Committer m b -> m ()-fromStreamM s c = go s-  where-    go str = do-      eNxt <- S.next str -- uncons requires r ~ ()-      forM_ eNxt $ \(a, str') -> do-        continue <- commit c a-        when continue (go str')

@@ -9,61 +9,51 @@ -- | timing effects module Box.Time   ( sleep,-    keepOpen,-    delayTimed,+    sleepUntil,     Stamped (..),     stampNow,-    emitStamp,+    stampE,+    emitOn,+    playback,+    simulate,   ) where -import Prelude import Box.Cont import Box.Emitter-import Box.Stream import Control.Monad.Conc.Class as C-import Control.Monad.IO.Class-import Control.Monad.Trans.Class import Data.Time-import qualified Streaming as S-import qualified Streaming.Prelude as S+import NumHask.Prelude hiding (STM, atomically)+import qualified Prelude as P  -- | sleep for x seconds sleep :: (MonadConc m) => Double -> m ()-sleep x = threadDelay (floor $ x * 1e6)+sleep x = C.threadDelay (fromIntegral (floor $ x * 1e6 :: Integer)) --- | keeping a box open sometimes needs a long running emitter-keepOpen :: (MonadConc m) => Cont m (Emitter (STM m) a)-keepOpen = toEmit $ lift $ threadDelay (365 * 24 * 60 * 60 * 10 ^ 6)+-- | sleep until a certain time (in the future)+sleepUntil :: UTCTime -> IO ()+sleepUntil u = do+  t0 <- getCurrentTime+  sleep (toDouble $ diffUTCTime u t0) --- | a stream with suggested delays.  DiffTime is the length of time to wait since the start of the stream--- > delayTimed (S.each (zip (fromIntegral <$> [1..10]) [1..10])) |> S.print-delayTimed ::-  (MonadConc m, MonadIO m) =>-  S.Stream (S.Of (NominalDiffTime, a)) m () ->-  S.Stream (S.Of a) m ()-delayTimed s = do-  t0 <- liftIO getCurrentTime-  go (S.hoistUnexposed lift s) t0-  where-    go s t0 = do-      n <- S.uncons s-      case n of-        Nothing -> pure ()-        Just ((t1, a'), s') -> do-          lift $ delayTo (addUTCTime t1 t0)-          S.yield a'-          go s' t0-    delayTo t = do-      now <- liftIO getCurrentTime-      let gap = max 0 $ diffUTCTime t now-      -- sleep gap-      threadDelay (truncate (gap * 1000000))+toDouble :: NominalDiffTime -> Double+toDouble t =+  (/ 1000000000000.0) $+    fromIntegral (P.floor $ t P.* 1000000000000 :: Integer) --- | A value with a timestamp annotation.+fromDouble :: Double -> NominalDiffTime+fromDouble x =+  let d0 = ModifiedJulianDay 0+      days = floor (x / toDouble nominalDay)+      secs = x - fromIntegral days * toDouble nominalDay+      t0 = UTCTime d0 (picosecondsToDiffTime 0)+      t1 = UTCTime (addDays days d0) (picosecondsToDiffTime $ floor (secs / 1.0e-12))+   in diffUTCTime t1 t0++-- | A value with a UTCTime annotation. data Stamped a   = Stamped-      { timestamp :: UTCTime,+      { stamp :: UTCTime,         value :: a       }   deriving (Eq, Show, Read)@@ -75,9 +65,39 @@   pure $ Stamped t a  -- | adding a time stamp--- todo: how to do this properly?-emitStamp ::+stampE ::   (MonadConc m, MonadIO m) =>-  Cont m (Emitter m a) ->-  Cont m (Emitter m (Stamped a))-emitStamp e = emap (fmap Just . stampNow) <$> e+  Emitter m a ->+  Emitter m (Stamped a)+stampE e = mapE (fmap Just . stampNow) e++-- | wait until Stamped time before emitting+emitOn ::+  Emitter IO (Stamped a) ->+  Emitter IO a+emitOn e =+  mapE+    ( \(Stamped u a) -> do+        sleepUntil u+        pure $ Just a+    )+    e++-- | reset the emitter stamps to by in sync with the current time and adjust the speed+-- >>> let e1 = fromListE (zipWith (\x a -> Stamped (addUTCTime (fromDouble x) t) a) [0..5] [0..5])+playback :: Double -> Emitter IO (Stamped a) -> IO (Emitter IO (Stamped a))+playback speed e = do+  r <- emit e+  case r of+    Nothing -> pure mempty+    Just (Stamped u0 _) -> do+      t0 <- getCurrentTime+      let ua = diffUTCTime t0 u0+      let delta u = addUTCTime ua $ addUTCTime (fromDouble ((toDouble $ diffUTCTime u u0) * speed)) u0+      pure (mapE (\(Stamped u a) -> pure (Just (Stamped (delta u) a))) e)++-- | simulate a delay from a (Stamped a) Emitter relative to the first timestamp+simulate :: Double -> Emitter IO (Stamped a) -> Cont IO (Emitter IO a)+simulate speed e = Cont $ \eaction -> do+  e' <- playback speed e+  eaction (emitOn e')

@@ -11,8 +11,12 @@ module Box.Transducer   ( Transducer (..),     etc,-    etcM,-    asPipe,+    toCommitter,+    toEmitter,+    toStream,+    fromStream,+    foldCommitter,+    sinkCommitter,   ) where @@ -20,17 +24,13 @@ import Box.Committer import Box.Cont import Box.Emitter-import Box.Stream-import Control.Category (Category (..))+import Box.Queue+import qualified Control.Foldl as L import Control.Lens hiding ((.>), (:>), (<|), (|>))-import Control.Monad.Base (MonadBase, liftBase) import Control.Monad.Conc.Class as C-import Control.Monad.Trans.State.Lazy-import qualified Pipes-import qualified Pipes.Prelude as Pipes+import NumHask.Prelude hiding (STM, atomically) import Streaming (Of (..), Stream) import qualified Streaming.Prelude as S-import Prelude hiding ((.), id)  -- | 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.@@ -44,33 +44,54 @@       }  instance Category (Transducer s) where-   (Transducer t1) . (Transducer t2) = Transducer (t1 . t2)    id = Transducer id --- | convert a Pipe to a Transducer-asPipe ::-  (Monad m) =>-  Pipes.Pipe a b (StateT s m) () ->-  (Stream (Of a) (StateT s m) () -> Stream (Of b) (StateT s m) ())-asPipe p s = ((s & Pipes.unfoldr S.next) Pipes.>-> p) & S.unfoldr Pipes.next- -- | emit - transduce - commit -- -- with etc, you're in the box, and inside the box, there are no effects: just a stream of a's, pure functions and state tracking. It's a nice way to code, and very friendly for the compiler. When the committing and emitting is done, the box collapses to state. -- -- The combination of an input tape, an output tape, and a state-based stream computation lends itself to the etc computation as a [finite-state transducer](https://en.wikipedia.org/wiki/Finite-state_transducer) or mealy machine.-etc :: (MonadConc m) => s -> Transducer s a b -> Cont m (Box (C.STM m) b a) -> m s-etc st t box =-  with box $ \(Box c e) ->-    (e & toStream & transduce t & fromStream) c & flip execStateT st+etc :: Monad m => s -> Transducer s a b -> Box m b a -> m s+etc st t (Box c e) =+  (e & hoist lift & toStream & transduce t & fromStream) (hoist lift c) & flip execStateT st --- | Monadic version of etc.-etcM :: (MonadConc m, MonadBase m m) => s -> Transducer s a b -> Cont m (Box m b a) -> m s-etcM st t box =-  with box $ \(Box c e) ->-    (liftE' e & toStreamM & transduce t & fromStreamM) (liftC' c) & flip execStateT st+-- | create an committer from a stream continuation+toCommitter :: (MonadConc m) => (Stream (Of a) m () -> m r) -> Cont m (Committer m a)+toCommitter f =+  Cont (\c -> queueC c (\(Emitter o) -> f . toStream . Emitter $ o))++-- | create an emitter from a stream+toEmitter :: (MonadConc m) => Stream (Of b) m () -> Cont m (Emitter m b)+toEmitter s = Cont (queueE (fromStream s))++-- | turn an emitter into a stream+toStream :: (Monad m) => Emitter m a -> Stream (Of a) m ()+toStream e = S.untilRight getNext   where-    liftC' c = Committer $ liftBase . commit c-    liftE' = Emitter . liftBase . emit+    getNext = maybe (Right ()) Left <$> emit e++-- | turn a stream into a committer+fromStream :: (Monad m) => Stream (Of b) m () -> Committer m b -> m ()+fromStream s c = go s+  where+    go str = do+      eNxt <- S.next str -- uncons requires r ~ ()+      forM_ eNxt $ \(a, str') -> do+        continue <- commit c a+        when continue (go str')++-- | create a committer from a fold+foldCommitter :: (MonadConc m) => L.FoldM m a () -> Cont m (Committer m a)+foldCommitter f = toCommitter (fmap S.snd' . L.impurely S.foldM f)++-- | create a committer from a sink+sinkCommitter :: (MonadConc m) => (a -> m ()) -> Cont m (Committer m a)+sinkCommitter sink = foldCommitter (L.FoldM step begin done)+  where+    step x a = do+      sink a+      pure x+    begin = pure ()+    done = pure

@@ -1,105 +0,0 @@-{-# LANGUAGE ExistentialQuantification #-}-{-# LANGUAGE RankNTypes #-}-{-# OPTIONS_GHC -Wall #-}---- | based on https://github.com/Gabriel439/Haskell-MVC-Updates-Library-module Box.Updater-  ( Updater (..),-    updater,-    listen,-    updates,-  )-where--import Prelude-import Box-import Control.Foldl (Fold (..), FoldM (..))-import qualified Control.Foldl as Foldl-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 GHC.Conc.STM e))--instance Functor Updater where-  fmap f (Updater fold' e) = Updater (fmap f fold') e--{--> onLeft (f <*> x) = onLeft f <*> onLeft x->-> onLeft (pure r) = pure r--}-onLeft :: Monad m => FoldM m a b -> FoldM m (Either a x) b-onLeft (FoldM step begin done) = FoldM step' begin done-  where-    step' x (Left a) = step x a-    step' x _ = return x--{--> onRight (f <*> x) = onRight f <*> onRight x->-> onRight (pure r) = pure r--}-onRight :: Monad m => FoldM m a b -> FoldM m (Either x a) b-onRight (FoldM step begin done) = FoldM step' begin done-  where-    step' x (Right a) = step x a-    step' x _ = return x--instance Applicative Updater where--  pure a = Updater (pure a) mempty--  (Updater foldL eL) <*> (Updater foldR eR) = Updater foldT eT-    where-      foldT = onLeft foldL <*> onRight foldR-      eT = fmap (fmap Left) eL <> fmap (fmap Right) eR---- | Create an 'Updater' value using a pure 'Fold'-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--- > listen mempty = id--- >--- > listen (f <> g) = listen g . listen f-listen :: (a -> IO ()) -> Updater a -> Updater a-listen handler (Updater (FoldM step begin done) mController) =-  Updater (FoldM step' begin' done) mController-  where-    begin' = do-      x <- begin-      b <- done x-      handler b-      return x-    step' x a = do-      x' <- step x a-      b <- done x'-      handler b-      return x'---- | Convert an 'Updater' to an Emitter continuation.-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-      C.newIORef x-    cio c =-      with e $ \e' -> do-        ioref' <- ioref c-        x <- C.readIORef ioref'-        e'' <- atomically $ emit e'-        case e'' of-          Nothing -> pure ()-          Just e''' -> do-            x' <- step x e'''-            a <- done x'-            _ <- atomically $ commit c a-            C.writeIORef ioref' x'

@@ -1,7 +1,8 @@-resolver: lts-15.13+resolver: nightly-2020-06-25  packages:   - .  extra-deps:-  - protolude-0.3.0+  - numhask-0.6.0+

@@ -1,12 +1,15 @@+{-# LANGUAGE NoImplicitPrelude #-} {-# OPTIONS_GHC -Wall #-}  module Main where -import Prelude+import NumHask.Prelude import Test.DocTest  main :: IO () main = doctest-  [ "src/Box.hs"-    -- "src/Box/Control.hs"+  [ "src/Box.hs",+    -- "src/Box/Committer.hs",+    "src/Box/IO.hs"+    -- "app/websocket-tests.hs"   ]