box 0.2.0 → 0.3.0

raw patch · 19 files changed

+1038/−474 lines, 19 filesdep +manageddep +protoludedep +typed-processnew-component:exe:concurrency-testsnew-component:exe:websocket-testsPVP ok

version bump matches the API change (PVP)

Dependencies added: managed, protolude, typed-process, websockets

API changes (from Hackage documentation)

Files

• app/concurrency-tests.hs +365/−0
• app/websocket-tests.hs +136/−0
• box.cabal +31/−4
• src/Box/Box.hs +1/−0
• src/Box/Broadcast.hs +3/−2
• src/Box/Committer.hs +1/−0
• src/Box/Connectors.hs +13/−12
• src/Box/Cont.hs +1/−2
• src/Box/Control.hs +355/−74
• src/Box/Emitter.hs +1/−0
• src/Box/IO.hs +37/−1
• src/Box/Plugs.hs +3/−2
• src/Box/Queue.hs +78/−4
• src/Box/Stream.hs +4/−3
• src/Box/Time.hs +1/−0
• src/Box/Updater.hs +2/−2
• stack.yaml +1/−1
• test/ctest.hs +0/−366
• test/test.hs +5/−1

@@ -0,0 +1,365 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# 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 #-}++-- | dejavu testing+--+module Main where++import Prelude+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 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)++-- | fuse+exFuse :: (MonadConc m) => Int -> m [Int]+exFuse n = do+  (b, res) <- testBox n+  fuse (pure . pure) (liftB <$> b)+  res++exFuseSTM :: (MonadConc m) => Int -> m [Int]+exFuseSTM n = do+  (b, res) <- testBox n+  fuseSTM (pure . pure) b+  res++exEtc :: (MonadConc m) => Int -> m [Int]+exEtc n = do+  (b, res) <- testBox n+  etc () (Transducer id) b+  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++exe1c2 :: (MonadConc m) => Int -> m ([Int],[Int])+exe1c2 n = e1c2 (toEmit (S.take n $ S.each [0..]))++-- | 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++exe1c2IO :: (MonadConc m) => Int -> m ([Int],[Int])+exe1c2IO n = e1c2IO (toEmit (S.take n $ S.each [0..]))++-- | 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++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)++

@@ -0,0 +1,136 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE NoImplicitPrelude #-}+{-# LANGUAGE OverloadedLabels #-}+{-# LANGUAGE OverloadedStrings #-}+{-# OPTIONS_GHC -Wall #-}++{-++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.++-}++module Main where++import Box+import Box.Control+import Control.Lens+import Control.Monad.Managed+import Data.Generics.Labels ()+import qualified Data.Text as Text+import GHC.Generics+import qualified Network.WebSockets as WS+import Protolude hiding (STM)++data ConfigSocket+  = ConfigSocket+      { host :: Text,+        port :: Int+      }+  deriving (Show, Eq, Generic)++defaultConfigSocket :: ConfigSocket+defaultConfigSocket = ConfigSocket "127.0.0.1" 9160++client :: ConfigSocket -> WS.ClientApp () -> IO ()+client c = WS.runClient (Text.unpack $ c ^. #host) (c ^. #port) "/"++server :: ConfigSocket -> WS.ServerApp -> IO ()+server c = WS.runServer (Text.unpack $ c ^. #host) (c ^. #port)++mconn :: WS.PendingConnection -> Managed WS.Connection+mconn p =+  managed $+    bracket+      (WS.acceptRequest p)+      (\conn -> WS.sendClose conn ("Bye from mconn!" :: Text))++-- | default websocket receiver+receiver ::+  Committer IO (Either ControlResponse Text) ->+  WS.Connection ->+  IO Bool+receiver c conn = go+  where+    go = do+      msg <- WS.receive conn+      case msg of+        WS.ControlMessage (WS.Close w b) ->+          commit+            c+            ( Left+                ( Info $+                    "received close message: " <> show w <> " " <> show b+                )+            )+        WS.ControlMessage _ -> go+        WS.DataMessage _ _ _ msg' -> commit c (Right (WS.fromDataMessage msg')) >> go++-- | default websocket sender+sender ::+  WS.WebSocketsData a =>+  Emitter IO a ->+  WS.Connection ->+  IO ()+sender 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)++-- | a receiver that immediately sends a response+responder ::+  (Text -> Either ControlResponse Text) ->+  WS.Connection ->+  IO ()+responder f conn = go+  where+    go = do+      msg <- WS.receive conn+      case msg of+        WS.ControlMessage (WS.Close _ _) -> do+          WS.sendClose conn ("returning close signal" :: Text)+          go+        WS.ControlMessage _ -> go+        WS.DataMessage _ _ _ msg' -> WS.sendTextData conn (show $ f $ WS.fromDataMessage msg' :: Text)++serverApp ::+  WS.PendingConnection ->+  IO ()+serverApp p = Control.Monad.Managed.with (mconn p) (responder Right)++-- | controlled server+serverBox ::+  ConfigSocket ->+  IO ()+serverBox cfg =+  Box.with+    (Box <$> showStdout <*> readStdin)+    (controlBox defaultControlConfig (server cfg serverApp))++main :: IO ()+main = pure ()

@@ -1,5 +1,5 @@ name:           box-version:        0.2.0+version:        0.3.0 synopsis:       boxes description:    concurrent, effectful boxes category:       project@@ -58,14 +58,15 @@     , time     , transformers     , transformers-base+    , typed-process   default-language: Haskell2010 -executable box-test-  main-is: ctest.hs+executable concurrency-tests+  main-is: concurrency-tests.hs   other-modules:       Paths_box   hs-source-dirs:-      test+      app   ghc-options: -funbox-strict-fields -fforce-recomp -threaded -rtsopts -with-rtsopts=-N   build-depends:     base >=4.7 && <5@@ -81,6 +82,31 @@     , transformers   default-language: Haskell2010 +executable websocket-tests+  main-is: websocket-tests.hs+  other-modules:+      Paths_box+  hs-source-dirs:+      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+  default-language: Haskell2010++ test-suite test   type: exitcode-stdio-1.0   main-is: test.hs@@ -93,3 +119,4 @@     base >=4.7 && <5     , doctest   default-language: Haskell2010+

@@ -15,6 +15,7 @@   ) where +import Prelude import Box.Committer import Box.Emitter import Control.Applicative

@@ -11,6 +11,7 @@   ) where +import Prelude import Box.Committer import Box.Cont import Box.Emitter@@ -35,7 +36,7 @@  -- | 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+subscribe (Broadcaster tvar) = Cont $ \e -> queueE' cio e   where     cio c = atomically $ modifyTVar' tvar (mappend c) @@ -58,4 +59,4 @@ -- | 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)+  Cont $ \c -> queueC' c $ \e -> atomically $ modifyTVar' tvar (mappend e)

@@ -16,6 +16,7 @@   ) where +import Prelude import Control.Lens hiding ((.>), (:>), (<|), (|>)) import Control.Monad.Conc.Class as C import Data.Functor.Constant

@@ -27,6 +27,7 @@   ) where +import Prelude import Box.Box import Box.Committer import Box.Cont@@ -84,19 +85,19 @@  -- | 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)+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)+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+fuseEmit e = Cont $ \eaction -> queueE' (fuseSTM_ e) eaction  -- | 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+fuseEmitM e = Cont $ \eaction -> queueEM' (fuse_ e) eaction  -- | merge two emitters --@@ -110,8 +111,8 @@     with e $ \e' ->       fst         <$> C.concurrently-          (queueE (fuseSTM_ (fst e')) eaction)-          (queueE (fuseSTM_ (snd e')) eaction)+          (queueE' (fuseSTM_ (fst e')) eaction)+          (queueE' (fuseSTM_ (snd e')) eaction)  -- | monadic version emergeM ::@@ -123,8 +124,8 @@     with e $ \e' ->       fst         <$> C.concurrently-          (queueEM (fuse_ (fst e')) eaction)-          (queueEM (fuse_ (snd e')) eaction)+          (queueEM' (fuse_ (fst e')) eaction)+          (queueEM' (fuse_ (snd e')) eaction)  -- | split a committer (STM m) splitCommitSTM ::@@ -135,8 +136,8 @@   Cont $ \kk ->     with c $ \c' ->       concurrentlyLeft-        (queueC (kk . Left) (`fuseSTM_` c'))-        (queueC (kk . Right) (`fuseSTM_` c'))+        (queueC' (kk . Left) (`fuseSTM_` c'))+        (queueC' (kk . Right) (`fuseSTM_` c'))  -- | split a committer splitCommit ::@@ -147,8 +148,8 @@   Cont $ \kk ->     with c $ \c' ->       concurrentlyLeft-        (queueCM (kk . Left) (`fuse_` c'))-        (queueCM (kk . Right) (`fuse_` c'))+        (queueCM' (kk . Left) (`fuse_` c'))+        (queueCM' (kk . Right) (`fuse_` c'))  -- | use a split committer contCommit :: Either (Committer m a) (Committer m b) -> (Committer m a -> Committer m b) -> Committer m b

@@ -11,10 +11,9 @@   ) where +import Prelude import Control.Applicative import Control.Monad.IO.Class (MonadIO (liftIO))-import Data.Monoid (Monoid (..))-import Data.Semigroup (Semigroup (..))  -- | A continuation similar to ` Control.Monad.ContT` but where the result type is swallowed by an existential newtype Cont m a

@@ -12,24 +12,35 @@ module Box.Control   ( ControlRequest (..),     ControlResponse (..),+    Toggle (..),     ControlBox,+    ControlBox_,     ControlConfig (..),     defaultControlConfig,     consoleControlBox,+    consoleControlBox_,     parseControlRequest,     controlBox,-    -- runControlBox,-    testBox,+    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@@ -41,117 +52,387 @@ import qualified Data.Text.IO as Text import GHC.Generics import qualified Streaming.Prelude as S-import Text.Read (readMaybe)+import System.IO+import System.Process.Typed+import Prelude  -- | request ADT data ControlRequest   = Check -- check for existence-  | Stop -- stop (without shutting down)-  | Quit -- stop, quit & cancel thread-  | Start -- start (if not yet started)+  | Start -- start (if not yet started) idempotent+  | Stop -- cancel (without shutting down) idempotent   | Reset -- stop and start (potentially cancelling a previous instance)-  | Kill -- immediately exit+  | 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-  = ShutDown -- action died-  | On Bool -- are we live?-  | Log Text-  deriving (Show, Read, Eq, Data, Typeable, Generic)+  = ShuttingDown -- shutdown+  | Status (Toggle, Int) -- on/off and number restarts left+  | Info Text+  deriving (Show, Read, Eq, Generic) --- | A 'Box' that communicates via 'ControlRequest' and 'ControlResponse'-type ControlBox m = (MonadConc m) => Cont m (Box (STM m) ControlResponse ControlRequest)+-- | A 'Box' that only communicates via 'ControlRequest' and 'ControlResponse'+type ControlBox_ m = (MonadConc m) => Cont m (Box (STM m) ControlResponse ControlRequest) --- | Should the box be kept alive?+-- | 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-  = KeepAlive Double-  | AllowDeath-  deriving (Show, Eq)+  = 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) --- | Defauilt is to let the box die.+-- | Default is one start, manual start and no autorestart. defaultControlConfig :: ControlConfig-defaultControlConfig = AllowDeath+defaultControlConfig = ControlConfig 1 False Nothing False  -- | a command-line control box.-consoleControlBox :: ControlBox IO+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+            <$> ( eParse (parseControlRequest A.takeText)                     <$> eStdin 1000                 )         ) --- | Parse command line requests-parseControlRequest :: A.Parser ControlRequest-parseControlRequest =-  A.string "q" $> Stop-    <|> A.string "s" $> Start-    <|> A.string "x" $> Quit-    <|> A.string "c" $> Check-    <|> A.string "r" $> Reset-    <|> do-      res <- readMaybe . Text.unpack <$> A.takeText-      case res of-        Nothing -> mzero-        Just a -> return a+-- | a command-line control box.+consoleControlBox_ :: ControlBox_ IO+consoleControlBox_ =+  bmap (pure . Just . Left) (pure . either Just (const Nothing))+    <$> consoleControlBox --- | an effect that can be started and stopped+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 app (Box c e) = do-  ref' <- C.newIORef Nothing-  _ <- C.atomically (commit c (On False))-  _ <- go ref'-  Text.putStrLn ("after go ref'" :: Text)+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-    go ref = do+    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 ref+        Nothing -> go r s         Just msg' ->           case msg' of-            Check -> do-              a <- C.readIORef ref-              _ <--                C.atomically $ commit c $ On (bool True False (isNothing a))-              go ref+            Check ->+              go r s             Start -> do-              a <- C.readIORef ref-              when (isNothing a) (void $ start ref c)-              go ref-            Stop -> cancel' ref c >> go ref-            Quit -> cancel' ref c >> C.atomically (commit c ShutDown)-            Reset -> do-              a <- C.readIORef ref-              unless (isNothing a) (void $ cancel' ref c)-              _ <- start ref c-              go ref-            _ -> go ref-    start ref c' = do-      a' <- async (app >> C.atomically (commit c' (On False)))-      C.writeIORef ref (Just a')-      C.atomically $ commit c' (On True)-    cancel' ref c' = do-      mapM_ cancel =<< C.readIORef ref-      C.writeIORef ref Nothing-      C.atomically $ commit c' (On False)+              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 --- | send Start, wait for a Ready signal, run action, wait x secs, then send Quit-testBox :: IO ()-testBox = cb+-- 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-    action =-      replicateM_ 3 (sleep 1 >> Text.putStrLn ("beep" :: Text))-    cb = with consoleControlBox (controlBox action)+    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+timeOut :: Double -> ControlBox m a b timeOut t =-  Box <$> mempty <*> ((lift (sleep t) >> S.yield Stop) & toEmit)+  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)

@@ -19,6 +19,7 @@   ) where +import Prelude import Control.Applicative import Control.Monad import Control.Monad.Conc.Class as C

@@ -19,7 +19,11 @@     eStdoutM,     eStdout',     cStdout,+    cStdout',+    eStdin',     showStdout,+    getLine,+    putLine,     consolePlug,     emitLines,     commitLines,@@ -51,7 +55,8 @@ import Streaming (Of (..), Stream) import qualified Streaming.Internal as S import qualified Streaming.Prelude as S-import System.IO+import System.IO hiding (getLine)+import Prelude hiding (getLine)  -- * console @@ -109,9 +114,40 @@ 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+ -- | 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+  -- | console box -- > etc () (Trans $ \s -> s & S.takeWhile (/="q") & S.map ("echo: " <>)) (console 5)

@@ -18,6 +18,7 @@   ) where +import Prelude import Box.Box import Box.Committer import Box.Cont@@ -33,11 +34,11 @@  -- | 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+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+emitPlugM cio = Cont $ \eio -> queueEM' cio eio  -- | create a double-queued box plug boxPlug ::

@@ -14,9 +14,13 @@   ( Queue (..),     queue,     queueC,+    queueC',     queueE,+    queueE',     queueCM,+    queueCM',     queueEM,+    queueEM',     waitCancel,     ends,     withQ,@@ -28,6 +32,7 @@   ) where +import Prelude import Box.Box import Box.Committer import Box.Emitter@@ -51,7 +56,7 @@ ends qu =   case qu of     Bounded n -> do-      q <- newTBQueue n+      q <- newTBQueue (fromIntegral n)       return (writeTBQueue q, readTBQueue q)     Unbounded -> do       q <- newTQueue@@ -66,7 +71,7 @@       m <- newEmptyTMVar       return (\x -> tryTakeTMVar m *> putTMVar m x, takeTMVar m)     Newest n -> do-      q <- newTBQueue n+      q <- newTBQueue (fromIntegral n)       let write x = writeTBQueue q x <|> (tryReadTBQueue q *> write x)       return (write, readTBQueue q) @@ -216,6 +221,43 @@           (eio (emitter box) `C.finally` seal)     ) +-- | connect a committer and emitter action via spawning a queue, and wait for both to complete.+withQEM ::+  (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 =+  C.bracket+    (spawner q)+    snd+    ( \(box, seal) ->+        concurrentlyRight+          (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.+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) =>@@ -232,6 +274,13 @@   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) =>@@ -240,6 +289,14 @@   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) =>@@ -249,6 +306,14 @@ 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) ->@@ -256,6 +321,15 @@   m r queueEM cm em = snd <$> withQM 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 = withQEM Unbounded toBoxM cm em++ -- | -- -- The one-in-the-chamber problem@@ -289,9 +363,9 @@ -- 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+-- 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+-- 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_)

@@ -21,6 +21,7 @@   ) where +import Prelude import Box.Committer import Box.Cont import Box.Emitter@@ -36,7 +37,7 @@ -- | 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))+  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)@@ -54,13 +55,13 @@  -- | 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))+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)+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 ()

@@ -17,6 +17,7 @@   ) where +import Prelude import Box.Cont import Box.Emitter import Box.Stream

@@ -11,8 +11,8 @@   ) where +import Prelude import Box-import Control.Applicative (Applicative ((<*>), pure)) import Control.Foldl (Fold (..), FoldM (..)) import qualified Control.Foldl as Foldl import Control.Monad.Conc.Class as C@@ -84,7 +84,7 @@  -- | 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'+updates (Updater (FoldM step begin done) e) = Cont $ \e' -> queueE' cio e'   where     ioref c = do       x <- begin

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

@@ -1,366 +0,0 @@-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE RankNTypes #-}-{-# LANGUAGE ScopedTypeVariables #-}-{-# 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 #-}---- | dejavu testing----module Main where--import Prelude-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 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)---- | fuse-exFuse :: (MonadConc m) => Int -> m [Int]-exFuse n = do-  (b, res) <- testBox n-  fuse (pure . pure) (liftB <$> b)-  res--exFuseSTM :: (MonadConc m) => Int -> m [Int]-exFuseSTM n = do-  (b, res) <- testBox n-  fuseSTM (pure . pure) b-  res--exEtc :: (MonadConc m) => Int -> m [Int]-exEtc n = do-  (b, res) <- testBox n-  etc () (Transducer id) b-  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--exe1c2 :: (MonadConc m) => Int -> m ([Int],[Int])-exe1c2 n = e1c2 (toEmit (S.take n $ S.each [0..]))---- | 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--exe1c2IO :: (MonadConc m) => Int -> m ([Int],[Int])-exe1c2IO n = e1c2IO (toEmit (S.take n $ S.each [0..]))---- | 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--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)--

@@ -2,7 +2,11 @@  module Main where +import Prelude import Test.DocTest  main :: IO ()-main = doctest ["src/Box.hs"]+main = doctest+  [ "src/Box.hs"+    -- "src/Box/Control.hs"+  ]