box 0.1.0 → 0.2.0
raw patch · 15 files changed
+412/−339 lines, 15 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
Files
- box.cabal +1/−1
- src/Box.hs +15/−19
- src/Box/Box.hs +14/−12
- src/Box/Broadcast.hs +18/−14
- src/Box/Committer.hs +33/−26
- src/Box/Connectors.hs +38/−39
- src/Box/Cont.hs +37/−19
- src/Box/Control.hs +14/−5
- src/Box/Emitter.hs +40/−29
- src/Box/Plugs.hs +13/−12
- src/Box/Queue.hs +104/−90
- src/Box/Stream.hs +16/−18
- src/Box/Time.hs +23/−18
- src/Box/Transducer.hs +28/−24
- src/Box/Updater.hs +18/−13
box.cabal view
@@ -1,5 +1,5 @@ name: box-version: 0.1.0+version: 0.2.0 synopsis: boxes description: concurrent, effectful boxes category: project
src/Box.hs view
@@ -9,27 +9,26 @@ -- | Boxes that `emit`, `transduce` & `commit` -- -- 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 -- $commit -- $emit -- $transduce- 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+ 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@@ -139,14 +138,12 @@ -- 'y' -- 'c' -- 'x'--- -- $transduce -- -- >>> etc () transducer' box' -- echo: hi -- echo: bye--- -- | broadcasting --@@ -154,4 +151,3 @@ -- -- > (funn, fem) <- C.atomically funnel -- >-
src/Box/Box.hs view
@@ -8,28 +8,28 @@ {-# OPTIONS_GHC -fno-warn-type-defaults #-} -- | A box is something that commits and emits--- module Box.Box- ( Box(..)- , liftB- , bmap- ) where+ ( Box (..),+ liftB,+ bmap,+ )+where -import Control.Lens hiding ((:>), (.>), (<|), (|>)) import Box.Committer import Box.Emitter-import Control.Monad.Conc.Class import Control.Applicative+import Control.Lens hiding ((.>), (:>), (<|), (|>))+import Control.Monad.Conc.Class -- | 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. -- -- a Box can also be seen as having an input tape and output tape, thus available for turing and finite-state machine metaphorics.----data Box m c e = Box- { committer :: Committer m c- , emitter :: Emitter m e- }+data Box m c e+ = Box+ { committer :: Committer m c,+ emitter :: Emitter m e+ } instance (Functor m) => Profunctor (Box m) where dimap f g (Box c e) = Box (contramap f c) (fmap g e)@@ -38,7 +38,9 @@ (<>) (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
src/Box/Broadcast.hs view
@@ -1,13 +1,15 @@ {-# OPTIONS_GHC -Wall #-} +-- | This module is experimental and may not work. module Box.Broadcast- ( Broadcaster(..)- , broadcast- , subscribe- , Funneler(..)- , funnel- , widen- ) where+ ( Broadcaster (..),+ broadcast,+ subscribe,+ Funneler (..),+ funnel,+ widen,+ )+where import Box.Committer import Box.Cont@@ -16,10 +18,11 @@ 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)- }+-- | 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)@@ -37,9 +40,10 @@ cio c = atomically $ modifyTVar' tvar (mappend c) -- | a funneler-newtype Funneler m a = Funneler- { unFunnel :: TVar m (Emitter m a)- }+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)
src/Box/Committer.hs view
@@ -1,54 +1,61 @@-{-# LANGUAGE UndecidableInstances #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-} {-# OPTIONS_GHC -Wall #-} -- | `commit` module Box.Committer- ( Committer(..)- , liftC- , cmap- , handles- ) where+ ( Committer (..),+ liftC,+ cmap,+ handles,+ )+where -import Control.Lens hiding ((:>), (.>), (<|), (|>))+import Control.Lens hiding ((.>), (:>), (<|), (|>))+import Control.Monad.Conc.Class as C import Data.Functor.Constant import Data.Functor.Contravariant.Divisible-import Control.Monad.Conc.Class as C+import Data.Monoid (First (..)) import Data.Void (absurd)-import Data.Monoid (First(..)) -- | a Committer a "commits" values of type a. A Sink and a Consumer are some other metaphors for this. ----- A Committer 'absorbs' the value being committed; the value disappears into the opaque thing that is a Committer from the pov of usage.----newtype Committer m a = Committer- { commit :: a -> m Bool- }+-- A Committer absorbs the value being committed; the value disappears into the opaque thing that is a Committer from the pov of usage.+newtype Committer m a+ = Committer+ { commit :: a -> m Bool+ } 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 = (<>) instance Contravariant (Committer m) where contramap f (Committer a) = Committer (a . f) instance (Applicative m) => Divisible (Committer m) where+ conquer = Committer (\_ -> pure False)+ divide f i1 i2 = Committer $ \a -> case f a of (b, c) -> (||) <$> commit i1 b <*> commit i2 c instance (Applicative m) => Decidable (Committer m) where+ lose f = Committer (absurd . f)+ choose f i1 i2 = Committer $ \a -> case f a of@@ -59,8 +66,7 @@ liftC :: (MonadConc m) => Committer (STM m) a -> Committer m a liftC c = Committer $ atomically . commit c --- | This is a contramapMaybe, if such a thing existed, as the contravariant version of a `mapMaybe`. See [witherable](https://hackage.haskell.org/package/witherable)---+-- | 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 where@@ -72,17 +78,18 @@ -- | prism handler handles ::- (Monad m)- => ((b -> Constant (First b) b) -> (a -> Constant (First b) a))- -- ^- -> Committer m b- -- ^- -> Committer m a+ (Monad m) =>+ -- |+ ((b -> Constant (First b) b) -> (a -> Constant (First b) a)) ->+ -- |+ Committer m b ->+ Committer m a handles k (Committer commit_) = Committer- (\a ->- case match a of- Nothing -> return False- Just b -> commit_ b)+ ( \a ->+ case match a of+ Nothing -> return False+ Just b -> commit_ b+ ) where match = getFirst . getConstant . k (Constant . First . Just)
src/Box/Connectors.hs view
@@ -8,34 +8,36 @@ -- | various ways to connect things up module Box.Connectors- ( fuse_- , fuseSTM_- , fuse- , fuseSTM- , forkEmit- , feedback- , feedbackE- , fuseEmit- , fuseEmitM- , fuseCommit- , fuseCommitM- , emerge- , emergeM- , splitCommit- , splitCommitSTM- , contCommit- ) where+ ( fuse_,+ fuseSTM_,+ fuse,+ fuseSTM,+ forkEmit,+ feedback,+ feedbackE,+ fuseEmit,+ fuseEmitM,+ fuseCommit,+ fuseCommitM,+ emerge,+ emergeM,+ splitCommit,+ splitCommitSTM,+ contCommit,+ )+where import Box.Box-import Box.Queue import Box.Committer import Box.Cont import Box.Emitter-import Control.Monad.Conc.Class as C+import Box.Queue import Control.Concurrent.Classy.Async as C import Control.Monad+import Control.Monad.Conc.Class as C -- * primitives+ -- | fuse an emitter directly to a committer fuse_ :: (Monad m) => Emitter m a -> Committer m a -> m () fuse_ e c = go@@ -63,7 +65,6 @@ -- bye -- -- > 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 @@ -80,6 +81,7 @@ 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)@@ -96,11 +98,9 @@ fuseEmitM :: (MonadConc m) => Emitter m a -> Cont m (Emitter m a) fuseEmitM e = Cont $ \eaction -> queueEM (fuse_ e) eaction - -- | merge 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) ->@@ -108,13 +108,12 @@ emerge e = Cont $ \eaction -> with e $ \e' ->- fst <$>- C.concurrently- (queueE (fuseSTM_ (fst e')) eaction)- (queueE (fuseSTM_ (snd e')) eaction)+ fst+ <$> C.concurrently+ (queueE (fuseSTM_ (fst e')) eaction)+ (queueE (fuseSTM_ (snd e')) eaction) -- | monadic version--- emergeM :: (MonadConc m) => Cont m (Emitter m a, Emitter m a) ->@@ -122,16 +121,16 @@ emergeM e = Cont $ \eaction -> with e $ \e' ->- fst <$>- C.concurrently- (queueEM (fuse_ (fst e')) eaction)- (queueEM (fuse_ (snd e')) eaction)+ fst+ <$> C.concurrently+ (queueEM (fuse_ (fst e')) eaction)+ (queueEM (fuse_ (snd 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 ::+ (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' ->@@ -140,10 +139,10 @@ (queueC (kk . Right) (`fuseSTM_` c')) -- | split a committer----splitCommit :: (MonadConc m) =>- Cont m (Committer m a)- -> Cont m (Either (Committer m a) (Committer m a))+splitCommit ::+ (MonadConc m) =>+ Cont m (Committer m a) ->+ Cont m (Either (Committer m a) (Committer m a)) splitCommit c = Cont $ \kk -> with c $ \c' ->
src/Box/Cont.hs view
@@ -4,72 +4,90 @@ {-# LANGUAGE RankNTypes #-} {-# OPTIONS_GHC -Wall #-} +-- | A continuation type. module Box.Cont- ( Cont(..)- , Cont_(..)- ) where+ ( Cont (..),+ Cont_ (..),+ )+where import Control.Applicative-import Control.Monad.IO.Class (MonadIO(liftIO))-import Data.Monoid (Monoid(..))-import Data.Semigroup (Semigroup(..))+import Control.Monad.IO.Class (MonadIO (liftIO))+import Data.Monoid (Monoid (..))+import Data.Semigroup (Semigroup (..)) --- | A continuation similar to `ContT` but where the result type is swallowed by an existential-newtype Cont m a = Cont- { with :: forall r. (a -> m r) -> m r- }+-- | A continuation similar to ` Control.Monad.ContT` but where the result type is swallowed by an existential+newtype Cont m a+ = Cont+ { with :: forall r. (a -> m r) -> m r+ } instance Functor (Cont m) where 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) instance (MonadIO m) => MonadIO (Cont m) where liftIO m = Cont- (\return_ -> do- a <- liftIO m- return_ a)+ ( \return_ -> do+ a <- liftIO m+ return_ a+ ) instance (Semigroup a) => Semigroup (Cont m a) where (<>) = liftA2 (<>) instance (Functor m, Semigroup a, Monoid a) => Monoid (Cont m a) where+ mempty = pure mempty+ mappend = (<>) -- | sometimes you have no choice but to void it up-newtype Cont_ m a = Cont_- { with_ :: (a -> m ()) -> m ()- }+newtype Cont_ m a+ = Cont_+ { with_ :: (a -> m ()) -> m ()+ } instance Functor (Cont_ m) where 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) instance (MonadIO m) => MonadIO (Cont_ m) where liftIO m = Cont_- (\return_ -> do- a <- liftIO m- return_ a)+ ( \return_ -> do+ a <- liftIO m+ return_ a+ ) instance (Semigroup a) => Semigroup (Cont_ m a) where (<>) = liftA2 (<>) instance (Functor m, Semigroup a, Monoid a) => Monoid (Cont_ m a) where+ mempty = pure mempty+ mappend = (<>)
src/Box/Control.hs view
@@ -8,6 +8,7 @@ {-# OPTIONS_GHC -Wall #-} {-# OPTIONS_GHC -Wno-redundant-constraints #-} +-- | An example of a Box for the command line. module Box.Control ( ControlRequest (..), ControlResponse (..),@@ -29,19 +30,20 @@ import Control.Lens hiding ((|>)) import Control.Monad import Control.Monad.Conc.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 qualified Data.Text.IO as Text import Data.Text (Text)+import qualified Data.Text.IO as Text import GHC.Generics import qualified Streaming.Prelude as S import Text.Read (readMaybe)-import Data.Functor-import Control.Monad.Trans.Class-import Data.Bool-import Data.Maybe +-- | request ADT data ControlRequest = Check -- check for existence | Stop -- stop (without shutting down)@@ -51,22 +53,27 @@ | Kill -- immediately exit deriving (Show, Read, Eq, Data, Typeable, Generic) +-- | response ADT data ControlResponse = ShutDown -- action died | On Bool -- are we live? | Log Text deriving (Show, Read, Eq, Data, Typeable, Generic) +-- | A 'Box' that communicates via 'ControlRequest' and 'ControlResponse' type ControlBox m = (MonadConc m) => Cont m (Box (STM m) ControlResponse ControlRequest) +-- | Should the box be kept alive? data ControlConfig = KeepAlive Double | AllowDeath deriving (Show, Eq) +-- | Defauilt is to let the box die. defaultControlConfig :: ControlConfig defaultControlConfig = AllowDeath +-- | a command-line control box. consoleControlBox :: ControlBox IO consoleControlBox = Box@@ -79,6 +86,7 @@ ) ) +-- | Parse command line requests parseControlRequest :: A.Parser ControlRequest parseControlRequest = A.string "q" $> Stop@@ -143,6 +151,7 @@ replicateM_ 3 (sleep 1 >> Text.putStrLn ("beep" :: Text)) cb = with consoleControlBox (controlBox action) +-- | A box with a self-destruct timer. timeOut :: Double -> ControlBox m timeOut t = Box <$> mempty <*> ((lift (sleep t) >> S.yield Stop) & toEmit)
src/Box/Emitter.hs view
@@ -1,49 +1,54 @@-{-# LANGUAGE UndecidableInstances #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-} {-# OPTIONS_GHC -Wall #-} {-# OPTIONS_GHC -fno-warn-type-defaults #-} -- | `emit` module Box.Emitter- ( Emitter(..)- , liftE- , emap- , keeps- , eRead- , eParse- ) where+ ( Emitter (..),+ liftE,+ emap,+ keeps,+ eRead,+ eParse,+ )+where -import Data.Functor.Constant-import qualified Data.Attoparsec.Text as A-import qualified Data.Text as Text-import Data.Text (Text)-import Control.Monad.Conc.Class as C import Control.Applicative import Control.Monad+import 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) --- | an `Emitter` "emits" values of type a. A Source & a Producer (of 'a's) are the two other alternative but overloaded metaphors out there.------ An Emitter 'reaches into itself' for the value to emit, where itself is an opaque thing from the pov of usage. An Emitter is named for its main action: it emits.+-- | an `Emitter` "emits" values of type a. A Source & a Producer (of a's) are the two other alternative but overloaded metaphors out there. ---newtype Emitter m a = Emitter- { emit :: m (Maybe a)- }+-- An Emitter "reaches into itself" for the value to emit, where itself is an opaque thing from the pov of usage. An Emitter is named for its main action: it emits.+newtype Emitter m a+ = Emitter+ { emit :: m (Maybe a)+ } 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 = Emitter $ do ma <- emit m@@ -52,7 +57,9 @@ Just a -> emit (f a) instance (Monad m, Alternative m) => Alternative (Emitter m) where+ empty = Emitter (pure Nothing)+ x <|> y = Emitter $ do (i, ma) <- fmap ((,) y) (emit x) <|> fmap ((,) x) (emit y)@@ -61,21 +68,25 @@ Just a -> pure (Just a) instance (Alternative m, Monad m) => MonadPlus (Emitter m) where+ mzero = empty+ mplus = (<|>) instance (Alternative m, Monad m) => Semigroup (Emitter m a) where (<>) = (<|>) 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 where@@ -91,12 +102,12 @@ -- | prism handler keeps ::- (Monad m)- => ((b -> Constant (First b) b) -> (a -> Constant (First b) a))- -- ^- -> Emitter m a- -- ^- -> Emitter m b+ (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@@ -115,9 +126,9 @@ -- | read parse emitter eRead ::- (Functor m, Read a)- => Emitter m Text- -> Emitter m (Either Text a)+ (Functor m, Read a) =>+ Emitter m Text ->+ Emitter m (Either Text a) eRead = fmap $ parsed . Text.unpack where parsed str =
src/Box/Plugs.hs view
@@ -9,23 +9,24 @@ -- | plugs -- box continuations--- module Box.Plugs- ( commitPlug- , emitPlug- , emitPlugM- , boxPlug- , boxForgetPlug- ) where+ ( commitPlug,+ emitPlug,+ emitPlugM,+ boxPlug,+ boxForgetPlug,+ )+where +import Box.Box import Box.Committer import Box.Cont-import Box.Box-import Box.Queue 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@@ -40,9 +41,9 @@ -- | create a double-queued box plug boxPlug ::- (Emitter STM a -> IO ())- -> (Committer STM b -> IO ())- -> Cont IO (Box STM a b)+ (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
src/Box/Queue.hs view
@@ -1,41 +1,41 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeFamilies #-} {-# OPTIONS_GHC -Wall #-} {-# OPTIONS_GHC -fno-warn-type-defaults #-}-{-# LANGUAGE OverloadedStrings #-} -- | queues -- Follows [pipes-concurrency](https://hackage.haskell.org/package/pipes-concurrency)--- module Box.Queue- ( Queue(..)- , queue- , queueC- , queueE- , queueCM- , queueEM- , waitCancel- , ends- , withQ- , withQE- , withQC- , toBox- , concurrentlyLeft- , concurrentlyRight- ) where+ ( Queue (..),+ queue,+ queueC,+ queueE,+ queueCM,+ queueEM,+ waitCancel,+ ends,+ withQ,+ withQE,+ withQC,+ toBox,+ concurrentlyLeft,+ concurrentlyRight,+ )+where import Box.Box import Box.Committer import Box.Emitter-import Control.Concurrent.Classy.STM as C-import Control.Monad.Conc.Class as C+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.Applicative+import Control.Monad.Conc.Class as C -- | 'Queue' specifies how messages are queued data Queue a@@ -82,33 +82,44 @@ -- | read from a queue, and retry if not sealed readCheck :: MonadSTM stm => TVar stm Bool -> stm a -> stm (Maybe a)-readCheck sealed o = (Just <$> o) <|> (do- b <- readTVar sealed- C.check b- pure Nothing)+readCheck sealed o =+ (Just <$> o)+ <|> ( do+ b <- readTVar sealed+ C.check b+ pure Nothing+ ) -- | turn a queue into a box (and a seal)-toBox :: (MonadSTM stm) =>- Queue a -> stm (Box stm a a, stm ())+toBox ::+ (MonadSTM stm) =>+ Queue a ->+ stm (Box stm a a, stm ()) toBox q = do (i, o) <- ends q sealed <- newTVarN "sealed" False let seal = writeTVar sealed True- pure (Box+ pure+ ( Box (Committer (writeCheck sealed i)) (Emitter (readCheck sealed o)),- seal)+ seal+ ) -toBoxM :: (MonadConc m) =>- Queue a -> m (Box m a a, m ())+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+ pure+ ( Box (Committer (atomically . writeCheck sealed i)) (Emitter (atomically $ readCheck sealed o)),- seal)+ seal+ ) -- | wait for the first action, and then cancel the second waitCancel :: (MonadConc m) => m b -> m a -> m b@@ -123,79 +134,87 @@ concurrentlyLeft :: MonadConc m => m a -> m b -> m a concurrentlyLeft left right = withAsync left $ \a ->- withAsync right $ \_ ->- wait a+ withAsync right $ \_ ->+ wait a -- | run two actions concurrently, but wait and return on the right result. concurrentlyRight :: MonadConc m => m a -> m b -> m b concurrentlyRight left right = withAsync left $ \_ ->- withAsync right $ \b ->- wait b+ withAsync right $ \b ->+ wait b -- | connect a committer and emitter action via spawning a queue, and wait for both to complete.-withQ :: (MonadConc m) =>- Queue a- -> (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 ::+ (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))+ ( \(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 ::+ (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))+ ( \(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 ::+ (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))+ ( \(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 ::+ (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 = C.bracket (spawner q) snd- (\(box, seal) ->- concurrently- (cio (committer box) `C.finally` seal)- (eio (emitter box) `C.finally` seal))+ ( \(box, seal) ->+ concurrently+ (cio (committer box) `C.finally` seal)+ (eio (emitter box) `C.finally` seal)+ ) -- | create an unbounded queue queue ::@@ -237,18 +256,15 @@ m r queueEM cm em = snd <$> withQM 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.---}+-- |+--+-- 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@@ -295,5 +311,3 @@ -- subbing withQE fixes! -- withQ Unbounded toBox (\c -> (withQE Unbounded toBox (\c' -> cStdin_ c' *> cStdin_ c') ((\e -> (fromStream . S.takeWhile (/="q") . toStream $ e) c)))) eStdout_--
src/Box/Stream.hs view
@@ -8,31 +8,31 @@ {-# OPTIONS_GHC -fno-warn-type-defaults #-} -- | Streaming functionality---- module Box.Stream- ( toStream- , fromStream- , toCommit- , toCommitFold- , toCommitSink- , toEmit- , queueStream- , toStreamM- , fromStreamM- ) where+ ( toStream,+ fromStream,+ toCommit,+ toCommitFold,+ toCommitSink,+ toEmit,+ queueStream,+ toStreamM,+ fromStreamM,+ )+where import Box.Committer import Box.Cont import Box.Emitter import Box.Queue-import Streaming (Of(..), Stream) import qualified Control.Foldl as L-import qualified Streaming.Prelude as S-import Control.Monad.Conc.Class as C 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 =@@ -59,7 +59,7 @@ -- | 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 ())+ (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@@ -85,5 +85,3 @@ forM_ eNxt $ \(a, str') -> do continue <- commit c a when continue (go str')--
src/Box/Time.hs view
@@ -8,23 +8,24 @@ -- | timing effects module Box.Time- ( sleep- , keepOpen- , delayTimed- , Stamped(..)- , stampNow- , emitStamp- ) where+ ( sleep,+ keepOpen,+ delayTimed,+ Stamped (..),+ stampNow,+ emitStamp,+ )+where -import Data.Time import Box.Cont import Box.Emitter import Box.Stream-import qualified Streaming.Prelude as S-import qualified Streaming as S import Control.Monad.Conc.Class as C import Control.Monad.IO.Class import Control.Monad.Trans.Class+import Data.Time+import qualified Streaming as S+import qualified Streaming.Prelude as S -- | sleep for x seconds sleep :: (MonadConc m) => Double -> m ()@@ -36,8 +37,10 @@ -- | 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 ::+ (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@@ -56,11 +59,15 @@ -- sleep gap threadDelay (truncate (gap * 1000000)) -data Stamped a = Stamped- { timestamp :: UTCTime- , value :: a- } deriving (Eq, Show, Read)+-- | A value with a timestamp annotation.+data Stamped a+ = Stamped+ { timestamp :: UTCTime,+ value :: a+ }+ deriving (Eq, Show, Read) +-- | Add the current time stampNow :: (MonadConc m, MonadIO m) => a -> m (Stamped a) stampNow a = do t <- liftIO getCurrentTime@@ -73,5 +80,3 @@ Cont m (Emitter m a) -> Cont m (Emitter m (Stamped a)) emitStamp e = emap (fmap Just . stampNow) <$> e--
src/Box/Transducer.hs view
@@ -8,60 +8,65 @@ {-# OPTIONS_GHC -fno-warn-type-defaults #-} -- | `transduce`--- module Box.Transducer- ( Transducer(..)- , etc- , etcM- , asPipe- ) where+ ( Transducer (..),+ etc,+ etcM,+ asPipe,+ )+where -import Prelude hiding ((.), id)-import Control.Category (Category(..))-import Control.Lens hiding ((:>), (.>), (<|), (|>))-import Control.Monad.Base (MonadBase, liftBase) import Box.Box import Box.Committer import Box.Cont import Box.Emitter import Box.Stream+import Control.Category (Category (..))+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 Streaming (Of(..), Stream)+import Streaming (Of (..), Stream) import qualified Streaming.Prelude as S-import Control.Monad.Conc.Class as C-import Control.Monad.Trans.State.Lazy+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.----newtype Transducer s a b = Transducer- { transduce :: forall m. Monad m =>- Stream (Of a) (StateT s m) () -> Stream (Of b) (StateT s m) ()- }+newtype Transducer s a b+ = Transducer+ { transduce ::+ forall m.+ Monad m =>+ Stream (Of a) (StateT s m) () ->+ Stream (Of b) (StateT s m) ()+ } 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) ())+ (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.+-- 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 +-- | 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) ->@@ -69,4 +74,3 @@ where liftC' c = Committer $ liftBase . commit c liftE' = Emitter . liftBase . emit-
src/Box/Updater.hs view
@@ -4,23 +4,26 @@ -- | based on https://github.com/Gabriel439/Haskell-MVC-Updates-Library module Box.Updater- ( Updater(..)- , updater- , listen- , updates- ) where+ ( Updater (..),+ updater,+ listen,+ updates,+ )+where -import Control.Applicative (Applicative((<*>), pure))-import Control.Foldl (Fold(..), FoldM(..))-import qualified Control.Foldl as Foldl 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 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))+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@@ -48,13 +51,15 @@ 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 `Updatable` value using a pure `Fold`+-- | 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') @@ -62,7 +67,6 @@ -- > 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@@ -78,6 +82,7 @@ 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