box (empty) → 0.0.1.1

raw patch · 22 files changed

+2286/−0 lines, 22 filesdep +asyncdep +attoparsecdep +basesetup-changed

Dependencies added: async, attoparsec, base, box, concurrency, contravariant, data-default, dejafu, doctest, exceptions, flow, foldl, generic-lens, lens, mmorph, pipes, profunctors, protolude, random, stm, streaming, text, time, transformers, transformers-base

Files

LICENSE +30/−0
Setup.hs +2/−0
box.cabal +146/−0
readme.md +13/−0
src/Box.hs +149/−0
src/Box/Box.hs +52/−0
src/Box/Broadcast.hs +59/−0
src/Box/Committer.hs +90/−0
src/Box/Connectors.hs +172/−0
src/Box/Cont.hs +75/−0
src/Box/Control.hs +154/−0
src/Box/Emitter.hs +126/−0
src/Box/IO.hs +177/−0
src/Box/Plugs.hs +53/−0
src/Box/Queue.hs +262/−0
src/Box/Stream.hs +94/−0
src/Box/Time.hs +80/−0
src/Box/Transducer.hs +73/−0
src/Box/Updater.hs +101/−0
stack.yaml +6/−0
test/ctest.hs +362/−0
test/test.hs +10/−0

+ LICENSE view

@@ -0,0 +1,30 @@+Copyright Tony Day (c) 2018++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * Redistributions in binary form must reproduce the above+      copyright notice, this list of conditions and the following+      disclaimer in the documentation and/or other materials provided+      with the distribution.++    * Neither the name of Tony Day nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

+ Setup.hs view

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

+ box.cabal view

@@ -0,0 +1,146 @@+cabal-version: 1.12++-- This file has been generated from package.yaml by hpack version 0.31.1.+--+-- see: https://github.com/sol/hpack+--+-- hash: 1e95b5568e907743dae547d71a55b1b3cd4d5845c032f31a2e81464d8a14b5b0++name:           box+version:        0.0.1.1+synopsis:       boxes+description:    concurrent, effectful boxes+category:       project+homepage:       https://github.com/tonyday567/box#readme+bug-reports:    https://github.com/tonyday567/box/issues+author:         Tony Day+maintainer:     tonyday567@gmail.com+copyright:      Tony Day (c) 2017+license:        BSD3+license-file:   LICENSE+build-type:     Simple+extra-source-files:+    stack.yaml+    readme.md++source-repository head+  type: git+  location: https://github.com/tonyday567/box++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+  other-modules:+      Paths_box+  hs-source-dirs:+      src+  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+    , data-default+    , dejafu+    , exceptions+    , flow+    , foldl+    , generic-lens+    , lens+    , mmorph+    , pipes+    , profunctors+    , protolude+    , random+    , stm+    , streaming+    , text+    , time+    , transformers+    , transformers-base+  default-language: Haskell2010++executable box-test+  main-is: ctest.hs+  other-modules:+      Paths_box+  hs-source-dirs:+      test+  ghc-options: -funbox-strict-fields -fforce-recomp -threaded -rtsopts -with-rtsopts=-N+  build-depends:+      async+    , attoparsec+    , base >=4.7 && <5+    , box+    , concurrency+    , contravariant+    , data-default+    , dejafu+    , exceptions+    , flow+    , foldl+    , generic-lens+    , lens+    , mmorph+    , pipes+    , profunctors+    , protolude+    , random+    , stm+    , streaming+    , text+    , time+    , transformers+    , transformers-base+  default-language: Haskell2010++test-suite test+  type: exitcode-stdio-1.0+  main-is: test.hs+  other-modules:+      Paths_box+  hs-source-dirs:+      test+  ghc-options: -Wall -Wcompat -Wincomplete-record-updates -Wincomplete-uni-patterns -Wredundant-constraints+  build-depends:+      async+    , attoparsec+    , base >=4.7 && <5+    , box+    , concurrency+    , contravariant+    , data-default+    , dejafu+    , doctest+    , exceptions+    , flow+    , foldl+    , generic-lens+    , lens+    , mmorph+    , pipes+    , profunctors+    , protolude+    , random+    , stm+    , streaming+    , text+    , time+    , transformers+    , transformers-base+  default-language: Haskell2010

+ readme.md view

@@ -0,0 +1,13 @@+[box](https://tonyday567.github.io/box/index.html) [![Build Status](https://travis-ci.org/tonyday567/box.svg)](https://travis-ci.org/tonyday567/box)+===++A concurrent, effectful, composable box to put stuff in and take stuff out.++compilation recipe+---++`box-test` takes about 5 minutes to run.++```+stack build --test --exec "$(stack path --local-install-root)/bin/box-test" --file-watch+```

+ src/Box.hs view

@@ -0,0 +1,149 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NoImplicitPrelude #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}+{-# OPTIONS_GHC -Wall #-}++-- | 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.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.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++-- $setup+-- >>> :set -XOverloadedStrings+-- >>> :set -XGADTs+-- >>> import Protolude+-- >>> import Box+-- >>> import qualified Streaming.Prelude as S+-- >>> import Control.Monad.Conc.Class as C+-- >>> 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: " <>)++-- $commit+-- committing+--+-- >>> with (cStdout 100) $ \c -> C.atomically (commit c "something")+-- something+-- True+--+-- The monoid instance sends each commit to both mappended committers. Delaying effects are introduced to these examples to keep stdout clear of race effects.+--+-- turned off in doctest until the muddled upedness is sorted...+-- > let cDelay = cmap (\b -> sleep 0.1 >> pure (Just b)) <$> (liftC <$> cStdout 100)+-- > let cImmediate = liftC <$> cStdout 100+-- > (etcM () transducer' $ (Box <$> (cImmediate <> cDelay) <*> (liftE <$> emitter'))) >> sleep 1+-- echo: hi+-- echo: hi+-- echo: bye+-- echo: bye+--+-- >>> let c = fmap liftC $ cStdout 10+-- >>> let e = fmap liftE $ toEmit (S.each ["hi","bye","q","x"])+-- >>> let c' = cmap (\a -> if a=="q" then (sleep 1 >> putStrLn "stolen!" >> sleep 1 >> pure (Nothing)) else (pure (Just a))) <$> c :: Cont IO (Committer IO Text)+-- >>> 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+-- FIXME:+-- > cs <- splitCommit (cStdout 100)+-- > let c2 = contCommit <$> cs+-- > (etcM () transducer' $ (Box <$> c2 <*> (liftE <$> emitter'))) >> sleep 1+--++-- $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 show <$> (toEmit $ delayTimed (S.each (zip (fromIntegral <$> [1..10]) ['a'..]))) :: Cont IO (Emitter (C.STM IO) Text)+-- >>> let e2 = fmap show <$> (toEmit $ delayTimed (S.each (zip ((\x -> fromIntegral x + 0.1) <$> [1..10]) (reverse ['a'..'z'])))) :: Cont IO (Emitter (C.STM IO) Text)+-- >>> let e12 = e1 <> e2+-- >>> etc () (Transducer identity) $ Box <$> cStdout 6 <*> emerge ((,) <$> e1 <*> e2)+-- 'a'+-- 'z'+-- 'b'+-- 'y'+-- 'c'+-- 'x'+--+-- >>> etc () (Transducer identity) $ Box <$> cStdout 6 <*> (liftA2 (<>) e1 e2)+-- 'a'+-- 'z'+-- 'b'+-- 'y'+-- 'c'+-- 'x'+--++-- $transduce+--+-- >>> etc () transducer' box'+-- echo: hi+-- echo: bye+--++-- | broadcasting+--+-- > (bcast, bcom) <- C.atomically broadcast+-- > (funn, fem) <- C.atomically funnel+-- >+

+ src/Box/Box.hs view

@@ -0,0 +1,52 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NoImplicitPrelude #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}+{-# OPTIONS_GHC -Wall #-}+{-# OPTIONS_GHC -fno-warn-type-defaults #-}++-- | A box is something that commits and emits+--+module Box.Box+  ( Box(..)+  , liftB+  , bmap+  ) where++import Control.Lens hiding ((:>), (.>), (<|), (|>))+import Data.Semigroup hiding (First, getFirst)+import Box.Committer+import Box.Emitter+import Control.Monad.Conc.Class+import Protolude hiding ((.), (<>), STM)++-- | 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+  }++instance (Functor m) => Profunctor (Box m) where+  dimap f g (Box c e) = Box (contramap f c) (fmap g e)++instance (Alternative m, Monad m) => Semigroup (Box m c e) where+  (<>) (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)

+ src/Box/Broadcast.hs view

@@ -0,0 +1,59 @@+{-# LANGUAGE NoImplicitPrelude #-}+{-# OPTIONS_GHC -Wall #-}++module Box.Broadcast+  ( Broadcaster(..)+  , broadcast+  , subscribe+  , Funneler(..)+  , funnel+  , widen+  ) where++import Box.Committer+import Box.Cont+import Box.Emitter+import Box.Queue+import Protolude hiding (STM, atomically, (.), (<>))+import Control.Concurrent.Classy.STM as C+import Control.Monad.Conc.Class as C++-- | 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)

+ src/Box/Committer.hs view

@@ -0,0 +1,90 @@+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NoImplicitPrelude #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}+{-# OPTIONS_GHC -Wall #-}++-- | `commit`+module Box.Committer+  ( Committer(..)+  , liftC+  , cmap+  , handles+  ) where++import Control.Category+import Control.Lens hiding ((:>), (.>), (<|), (|>))+import Data.Functor.Constant+import Data.Functor.Contravariant.Divisible+import Data.Semigroup hiding (First, getFirst)+import Protolude hiding ((.), (<>), STM, atomically)+import Control.Monad.Conc.Class as C++-- | 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+  }++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+        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++-- | 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+    go b = do+      fb <- f b+      case fb of+        Nothing -> pure True+        Just fb' -> commit c fb'++-- | prism handler+handles ::+     (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)+  where+    match = getFirst . getConstant . k (Constant . First . Just)

+ src/Box/Connectors.hs view

@@ -0,0 +1,172 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NoImplicitPrelude #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}+{-# OPTIONS_GHC -Wall #-}+{-# OPTIONS_GHC -fno-warn-type-defaults #-}++-- | various ways to connect things up+module Box.Connectors+  ( fuse_+  , fuseSTM_+  , fuse+  , fuseSTM+  , forkEmit+  , feedback+  , feedbackE+  , fuseEmit+  , fuseEmitM+  , fuseCommit+  , emerge+  , emergeM+  , splitCommit+  , contCommit+  ) where++import Control.Category+import Control.Lens hiding ((:>), (.>), (<|), (|>))+import Data.Semigroup hiding (First, getFirst)+import Box.Box+import Box.Queue+import Box.Committer+import Box.Cont+import Box.Emitter+import Protolude hiding (STM, (.), (<>))+import Control.Monad.Conc.Class as C+import Control.Concurrent.Classy.Async as C++-- * primitives+-- | 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++-- | 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++-- | fuse a box+--+-- > (fuse (pure . Just) $ liftB <$> (Box <$> cStdout 2 <*> emitter')) >> sleep 1+-- hi+-- 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++-- | 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++-- | fuse-branch an emitter+forkEmit :: (Monad m) => Emitter m a -> Committer m a -> Emitter m a+forkEmit e c =+  Emitter $ do+    a <- emit e+    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 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++-- | 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+++-- | 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) ->+  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 ::+  (MonadConc m) =>+  Cont m (Emitter m a, Emitter m a) ->+  Cont m (Emitter m a)+emergeM e =+  Cont $ \eaction ->+    with e $ \e' ->+      fst <$>+      C.concurrently+        (queueEM (fuse_ (fst e')) eaction)+        (queueEM (fuse_ (snd e')) eaction)++-- | merge two committers+--+-- not working+--+splitCommit :: (MonadConc m) =>+     Cont m (Committer (STM m) a)+  -> Cont m (Either (Committer (STM m) a) (Committer (STM m) a))+splitCommit c =+  Cont $ \kk ->+    with c $ \c' ->+      fst <$>+      C.concurrently+        (queueC (kk . Left) (`fuseSTM_` c'))+        (queueC (kk . Right) (`fuseSTM_` c'))++-- | a failed attempt to understand the either continuation style+contCommit :: Either (Committer m Text) (Committer m Text) -> Committer m Text+contCommit ec =+  Committer $ \a ->+    case ec of+      Left lc -> commit (contramap ("left " <>) lc) a+      Right rc -> commit rc a++-- | a box modifier that feeds commits back to the emitter+feedback ::+  (MonadConc m) =>+  (a -> m (Maybe b)) ->+  Cont m (Box m b a) ->+  Cont m (Box m b a)+feedback f box =+  Cont $ \bio ->+    with box $ \(Box c e) -> do+      fuse_ (emap f e) c+      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))

+ src/Box/Cont.hs view

@@ -0,0 +1,75 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RankNTypes #-}+{-# OPTIONS_GHC -Wall #-}++module Box.Cont+  ( Cont(..)+  , Cont_(..)+  ) where++import Control.Applicative+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+  }++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)++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 ()+  }++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)++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

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

+ src/Box/Emitter.hs view

@@ -0,0 +1,126 @@+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NoImplicitPrelude #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}+{-# OPTIONS_GHC -Wall #-}+{-# OPTIONS_GHC -fno-warn-type-defaults #-}++-- | `emit`+module Box.Emitter+  ( Emitter(..)+  , liftE+  , emap+  , keeps+  , eRead+  , eParse+  ) where++import Control.Category ((.))+import Data.Functor.Constant+import Data.Semigroup hiding (First, getFirst)+import Protolude hiding ((.), (<>), STM, atomically)+import qualified Data.Attoparsec.Text as A+import qualified Data.Text as Text+import Control.Monad.Conc.Class as C++-- | 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.+--+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+      case ma of+        Nothing -> return Nothing+        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)+      case ma of+        Nothing -> emit i+        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 = (<>)++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+    go = do+      a <- emit e+      case a of+        Nothing -> pure Nothing+        Just a' -> do+          fa <- f a'+          case fa of+            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)++-- | 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++-- | read parse emitter+eRead ::+     (Functor m, Read a)+  => Emitter m Text+  -> Emitter m (Either Text a)+eRead = fmap $ parsed . Text.unpack+  where+    parsed str =+      case reads str of+        [(a, "")] -> Right a+        _ -> Left (Text.pack str)

+ src/Box/IO.hs view

@@ -0,0 +1,177 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NoImplicitPrelude #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}+{-# OPTIONS_GHC -Wall #-}+{-# OPTIONS_GHC -fno-warn-type-defaults #-}++-- | IO actions+module Box.IO+  ( cStdin_+  , cStdin+  , cStdin'+  , eStdin+  , readStdin+  , eStdout_+  , eStdout+  , eStdoutM+  , eStdout'+  , cStdout+  , showStdout+  , consolePlug+  , emitLines+  , commitLines+  , cCRef+  , cCRefM+  , toListM+  , getCommissions+  , getEmissions+  ) where++import Control.Category+import qualified Control.Foldl as L+import Control.Lens hiding ((:>), (.>), (<|), (|>))+import Data.Semigroup hiding (First, getFirst)+import qualified Data.Text.IO as Text+import Box.Box+import Box.Committer+import Box.Cont+import Box.Emitter+import Box.Plugs+import Box.Stream+import Box.Transducer+import Flow+import Protolude hiding ((.), (<>), STM)+import Streaming (Of(..), Stream)+import qualified Streaming.Internal as S+import qualified Streaming.Prelude as S+import qualified Control.Monad.Conc.Class as C+import qualified Control.Concurrent.Classy.IORef as C+import Control.Monad.Conc.Class (STM)++-- * console+-- | a single stdin committer action+cStdin_ :: Committer (STM IO) Text -> IO ()+cStdin_ c = do+  a <- getLine+  void $ atomically $ commit c a++-- | a finite stdin committer action+cStdin :: Int -> Committer (STM IO) Text -> IO ()+cStdin n c = replicateM_ n (cStdin_ c)++-- | a forever stdin committer action+cStdin' :: Committer (STM IO) Text -> IO ()+cStdin' = forever . cStdin_++-- | a Cont stdin emitter+eStdin :: Int -> Cont IO (Emitter (STM IO) Text)+eStdin n = cStdin n |> emitPlug++-- | 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_ :: (Print a) => Emitter (STM IO) a -> IO ()+eStdout_ e = do+  a <- atomically $ emit e+  case a of+    Nothing -> pure ()+    Just a' -> putStrLn a'++-- | a single stdout emitter action+eStdoutM_ :: (Print a) => Emitter IO a -> IO ()+eStdoutM_ e = do+  a <- emit e+  case a of+    Nothing -> pure ()+    Just a' -> putStrLn a'++-- | a finite stdout emitter action+eStdout :: (Print a) => Int -> Emitter (STM IO) a -> IO ()+eStdout n = replicateM_ n . eStdout_++-- | a finite stdout emitter action+eStdoutM :: (Print a) => Int -> Emitter IO a -> IO ()+eStdoutM n = replicateM_ n . eStdoutM_++-- | a forever stdout emitter action+eStdout' :: (Print a) => Emitter (STM IO) a -> IO ()+eStdout' = forever . eStdout_++-- | a Cont stdout committer+cStdout :: Print a => Int -> Cont IO (Committer (STM IO) a)+cStdout n = eStdout n |> commitPlug++-- | show to stdout+showStdout :: Show a => Cont IO (Committer (STM IO) a)+showStdout = contramap show <$> (cStdout 1000 :: Cont IO (Committer (STM IO) Text))++-- | 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)++-- * file operations+-- | emit lines from a file+emitLines :: FilePath -> Cont IO (Emitter (STM IO) Text)+emitLines filePath = Cont (withFile filePath ReadMode) >>= fromHandle .> toEmit+  where+    fromHandle :: Handle -> Stream (Of Text) IO ()+    fromHandle h =+      forever $ do+        a <- liftIO $ Text.hGetLine h+        S.yield a++-- | commit lines to a file+commitLines :: FilePath -> Cont IO (Committer (STM IO) Text)+commitLines filePath =+  Cont (withFile filePath WriteMode) >>= toHandle .> toCommit+  where+    toHandle h = loop+      where+        loop str =+          case str of+            S.Return r -> return r+            S.Effect m -> m >>= loop+            S.Step (s :> rest) -> do+              liftIO (Text.hPutStrLn h s)+              loop rest++-- * concurrent refs+-- | commit to a list CRef+cCRef :: (C.MonadConc m) => m (C.IORef m [b], Cont m (Committer (C.STM m) b), m [b])+cCRef = do+  ref <- C.newIORef []+  let c = toCommitFold $+        L.FoldM (\x a -> C.modifyIORef x (a :) >> pure x) (pure ref) (const $ pure ())+  let 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++-- | 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)

+ src/Box/Plugs.hs view

@@ -0,0 +1,53 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NoImplicitPrelude #-}+{-# 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 Control.Category+import Box.Committer+import Box.Cont+import Box.Box+import Box.Queue+import Box.Emitter+import GHC.Conc+import Protolude hiding ((.), (<>))++-- * plugs+-- | hook an emitter action to a queue, creating a committer continuation+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))

+ src/Box/Queue.hs view

@@ -0,0 +1,262 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NoImplicitPrelude #-}+{-# 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+  , queueCLog+  , queueELog+  , waitCancel+  , toBoxLog+  , ends+  ) where++import Box.Box+import Box.Committer+import Box.Emitter+-- import GHC.Conc+import Protolude hiding (STM, check, wait, cancel, atomically, withAsync, concurrently)+import Control.Concurrent.Classy.STM as C+import Control.Monad.Conc.Class as C+import Control.Concurrent.Classy.Async as C+import Control.Monad.Catch as C++-- | 'Queue' specifies how messages are queued+data Queue a+  = Unbounded+  | Bounded Int+  | Single+  | Latest a+  | Newest Int+  | New++-- | create a queue, returning the ends+ends :: MonadSTM stm => Queue a -> stm (a -> stm (), stm a)+ends qu =+  case qu of+    Bounded n -> do+      q <- newTBQueue n+      return (writeTBQueue q, readTBQueue q)+    Unbounded -> do+      q <- newTQueue+      return (writeTQueue q, readTQueue q)+    Single -> do+      m <- newEmptyTMVar+      return (putTMVar m, takeTMVar m)+    Latest a -> do+      t <- newTVar a+      return (writeTVar t, readTVar t)+    New -> do+      m <- newEmptyTMVar+      return (\x -> tryTakeTMVar m *> putTMVar m x, takeTMVar m)+    Newest n -> do+      q <- newTBQueue n+      let write x = writeTBQueue q x <|> (tryReadTBQueue q *> write x)+      return (write, readTBQueue q)++-- | write to a queue, checking the seal+writeCheck :: (MonadSTM stm) => TVar stm Bool -> (a -> stm ()) -> a -> stm Bool+writeCheck sealed i a = do+  b <- readTVar sealed+  if b+    then pure False+    else do+      i a+      pure True++-- | 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)++-- | turn a queue into a box (and a seal)+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+        (Committer (writeCheck sealed i))+        (Emitter (readCheck sealed o)),+        seal)++-- | write to a queue, checking the seal+writeCheckLog :: (Show a, MonadIO m, MonadConc m) => TVar (STM m) Bool -> (a -> (STM m) ()) -> a -> m Bool+writeCheckLog sealed i a = do+  b <- atomically $ readTVar sealed+  if b+    then do+      putStrLn ("writeCheckLog sealed" :: Text)+      pure False+    else do+      putStrLn $ "writeCheckLog writing: " <> (show a:: Text)+      atomically $ i a+      pure True++-- | read from a queue, and retry if not sealed+readCheckLog :: (Show a, MonadIO m, MonadConc m) => TVar (STM m) Bool -> (STM m) a -> m (Maybe a)+readCheckLog sealed o = do+  o' <- atomically o+  putStrLn $ "readCheckLog reading" <> (show o' :: Text)+  atomically (Just <$> o <|> do+                            b <- readTVar sealed+                            C.check b+                            pure Nothing)++toBoxM :: (MonadConc m) =>+  Queue a -> m (Box m a a, m ())+toBoxM q = do+  (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)++toBoxLog :: (Show a, MonadIO m, MonadConc m) =>+  Queue a -> m (Box m a a, m ())+toBoxLog q = do+  (i, o) <- atomically $ ends q+  sealed <- atomically $ newTVarN "sealed" False+  initSeal <- atomically $ readTVar sealed+  putStrLn $ "toBoxLog: seal: " <> (show initSeal :: Text)+  let seal = atomically $ writeTVar sealed True+  pure (Box+        (Committer (writeCheckLog sealed i))+        (Emitter (readCheckLog sealed o)),+        seal)++-- | wait for the first action, and then cancel the second+waitCancel :: (MonadConc m) => m b -> m a -> m b+waitCancel a b =+  withAsync a $ \a' ->+    withAsync b $ \b' -> do+      a'' <- wait a'+      cancel b'+      pure a''++-- | 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 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 =+  C.bracket+    (spawner q)+    (\(_, seal) -> seal)+    (\(box, seal) ->+       concurrently+         (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.+withQMLog :: (MonadConc m, MonadIO m) =>+     Queue a+  -> (Queue a -> m (Box m a a, m ()))+  -> (Committer m a -> m l)+  -> (Emitter m a -> m r)+  -> m (l, r)+withQMLog q spawner cio eio =+  C.bracket+    (spawner q)+    (\(_, seal) -> seal)+    (\(box, seal) ->+       concurrently+         (cio (committer box) `C.finally` (putStrLn ("committer sealed" :: Text) >> seal))+         (eio (emitter box) `C.finally` (putStrLn ("emitter sealed" :: Text) >> 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++-- | 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 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+queueEM ::+  (MonadConc m) =>+  (Committer m a -> m l) ->+  (Emitter m a -> m r) ->+  m r+queueEM cm em = snd <$> withQM Unbounded toBoxM cm em++-- | create an unbounded queue, returning the emitter result+queueELog ::+  (Show a, MonadConc m, MonadIO m) =>+  (Committer m a -> m l) ->+  (Emitter m a -> m r) ->+  m r+queueELog cm em = snd <$> withQMLog Unbounded toBoxLog+  (\c -> putStrLn ("cm acted" :: Text) >> cm c)+  (\c -> putStrLn ("em acted" :: Text) >> em c)++-- | create an unbounded queue, returning the committer result+queueCLog ::+  (Show a, MonadConc m, MonadIO m) =>+  (Committer m a -> m l) ->+  (Emitter m a -> m r) ->+  m l+queueCLog cm em = fst <$> withQM Unbounded toBoxLog cm em

+ src/Box/Stream.hs view

@@ -0,0 +1,94 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NoImplicitPrelude #-}+{-# 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 Control.Category++import Box.Committer+import Box.Cont+import Box.Emitter+import Box.Queue+-- import Box.Transducer+import Flow+import Streaming (Of(..), Stream)+import qualified Control.Foldl as L+import qualified Streaming.Prelude as S+import Control.Monad.Conc.Class as C+import Protolude hiding ((<>), (.), STM, check, wait, cancel, atomically, withAsync, concurrently)++-- * 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) (toStream .> o)++-- | 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')++

+ src/Box/Time.hs view

@@ -0,0 +1,80 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE NoImplicitPrelude #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# OPTIONS_GHC -fno-warn-name-shadowing #-}+{-# OPTIONS_GHC -fno-warn-orphans #-}+{-# OPTIONS_GHC -fno-warn-type-defaults #-}++-- | timing effects+module Box.Time+  ( sleep+  , keepOpen+  , delayTimed+  , Stamped(..)+  , stampNow+  , emitStamp+  ) where++import Control.Applicative+import Control.Monad+import Data.Maybe+import Data.Time+import Box.Cont+import Box.Emitter+import Box.Stream+import Protolude hiding (threadDelay, STM)+import qualified Streaming.Prelude as S+import qualified Streaming as S+import Control.Monad.Conc.Class as C++-- | sleep for x seconds+sleep :: (MonadConc m) => Double -> m ()+sleep x = threadDelay (floor $ x * 1e6)++-- | 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)++-- | 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))++data Stamped a = Stamped+  { timestamp :: UTCTime+  , value :: a+  } deriving (Eq, Show, Read)++stampNow :: (MonadConc m, MonadIO m) => a -> m (Stamped a)+stampNow a = do+  t <- liftIO getCurrentTime+  pure $ Stamped t a++-- | adding a time stamp+-- todo: how to do this properly?+emitStamp ::+  (MonadConc m, MonadIO m) =>+  Cont m (Emitter m a) ->+  Cont m (Emitter m (Stamped a))+emitStamp e = emap (fmap Just . stampNow) <$> e++

+ src/Box/Transducer.hs view

@@ -0,0 +1,73 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NoImplicitPrelude #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}+{-# OPTIONS_GHC -Wall #-}+{-# OPTIONS_GHC -fno-warn-type-defaults #-}++-- | `transduce`+--+module Box.Transducer+  ( Transducer(..)+  , etc+  , etcM+  , asPipe+  ) where++import Control.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 Flow+import qualified Pipes+import qualified Pipes.Prelude as Pipes+import Protolude hiding ((.), (<>))+import Streaming (Of(..), Stream)+import qualified Streaming.Prelude as S+import Control.Monad.Conc.Class as C++-- | 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) ()+  }++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++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+  where+    liftC' c = Committer $ liftBase . commit c+    liftE' = Emitter . liftBase . emit+

+ src/Box/Updater.hs view

@@ -0,0 +1,101 @@+{-# LANGUAGE NoImplicitPrelude #-}+{-# 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 Control.Applicative (Applicative((<*>), pure))+import Control.Foldl (Fold(..), FoldM(..))+import qualified Control.Foldl as Foldl+import Data.IORef (newIORef, readIORef, writeIORef)+import Box+import Protolude++-- | An updater of a value a, where the updating process consists of an IO fold over an emitter+data Updater a =+  forall e. Updater (FoldM IO e a)+                    (Cont IO (Emitter STM e))++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 `Updatable` value using a pure `Fold`+updater :: Fold e a -> Cont IO (Emitter 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'++updates :: Updater a -> Cont IO (Emitter STM a)+updates (Updater (FoldM step begin done) e) = Cont $ \e' -> queueE cio e'+  where+    ioref c = do+      x <- begin+      a <- done x+      _ <- atomically $ commit c a+      newIORef x+    cio c =+      with e $ \e' -> do+        ioref' <- ioref c+        x <- 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+            writeIORef ioref' x'

+ stack.yaml view

@@ -0,0 +1,6 @@+resolver: lts-13.21++packages:+  - .++extra-deps: []

+ test/ctest.hs view

@@ -0,0 +1,362 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NoImplicitPrelude #-}+{-# 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 Control.Category (id)+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 Data.String+import Protolude hiding (STM)+import Test.DejaFu+import Test.DejaFu.Conc+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++-- | 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 :: (MonadConc n, MonadIO n, Eq b, Show b) =>+     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+  putStrLn $ "eref: " <> (show eres :: Text)+  (,) <$> 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 $ newTVar mempty+  let com = Committer $ \a -> do+        putStrLn $ "broadcaster': " <> (show a :: Text)+        c <- C.atomically $ readTVar ref+        commit c a+  pure (Broadcaster' ref, com)++-- | subscribe to a broadcaster+subscribe' :: (Show a, MonadIO m, MonadConc m) => Broadcaster' m a -> Cont m (Emitter m a)+subscribe' (Broadcaster' tvar) = Cont $ \e -> queueELog cio e+  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+      putStrLn $ "fuse_' emit: " <> (show a :: Text)+      c' <- maybe (pure False) (commit c) a+      putStrLn $ "fuse_' commit: " <> (show c' :: Text)+      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 <- Protolude.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)++

+ test/test.hs view

@@ -0,0 +1,10 @@+{-# LANGUAGE NoImplicitPrelude #-}+{-# OPTIONS_GHC -Wall #-}++module Main where++import Protolude+import Test.DocTest++main :: IO ()+main = doctest ["src/Box.hs"]