atelier-core 0.2.0.0 → 0.3.0.0

raw patch · 15 files changed

+419/−195 lines, 15 filesdep ~atelier-corenew-uploaderPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: atelier-core

API changes (from Hackage documentation)

Files

• CHANGELOG.md +18/−0
• README.md +4/−4
• atelier-core.cabal +8/−5
• src/Atelier/Effects/Conc.hs +24/−1
• src/Atelier/Effects/Input.hs +17/−0
• src/Atelier/Effects/Internal/Coroutine.hs +2/−2
• src/Atelier/Effects/Iterator.hs +2/−2
• src/Atelier/Effects/Process.hs +16/−0
• src/Atelier/Effects/Publishing.hs +30/−134
• src/Atelier/Effects/Publishing/Pub.hs +57/−0
• src/Atelier/Effects/Publishing/Sub.hs +152/−0
• test/Unit/Atelier/Effects/Conc/TeardownStressSpec.hs +5/−2
• test/Unit/Atelier/Effects/IteratorSpec.hs +12/−9
• test/Unit/Atelier/Effects/Publishing/PubSpec.hs +53/−0
• test/Unit/Atelier/Effects/PublishingSpec.hs +19/−36

@@ -5,6 +5,24 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to the [PVP](https://pvp.haskell.org/). +## [Unreleased]++## [0.3.0.0] - 2026-08-06++### Added++- `Atelier.Effects.Process`: `getExecutablePath` returns the absolute path of+  the currently running executable, for re-invoking the program as a+  subprocess.+- `Atelier.Effects.Publishing.Pub:` `map` and `mapM` allows mapping over+  published effects, producing a new `Pub` effect with the transformed values.++### Changed++- Moved `Pub` and `Sub` out of `Atelier.Effects.Publishing` to separate modules,+  `Atelier.Effects.Publishing.Pub` and `Atelier.Effects.Publishing.Sub`+  respectively.+ ## [0.2.0.0] - 2026-06-26  ### Added

@@ -23,10 +23,10 @@  ## Part of atelier -- [`atelier-prelude`](https://github.com/atelier-hub/tricorder/tree/main/atelier-prelude) — relude-based prelude with Effectful conventions-- [`atelier-core`](https://github.com/atelier-hub/tricorder/tree/main/atelier-core) — this package-- [`atelier-db`](https://github.com/atelier-hub/tricorder/tree/main/atelier-db) — relational database effect (Hasql/Rel8)-- [`atelier-testing`](https://github.com/atelier-hub/tricorder/tree/main/atelier-testing) — database-backed test utilities+- [`atelier-prelude`](https://github.com/tweag/tricorder/tree/main/atelier-prelude) — relude-based prelude with Effectful conventions+- [`atelier-core`](https://github.com/tweag/tricorder/tree/main/atelier-core) — this package+- [`atelier-db`](https://github.com/tweag/tricorder/tree/main/atelier-db) — relational database effect (Hasql/Rel8)+- [`atelier-testing`](https://github.com/tweag/tricorder/tree/main/atelier-testing) — database-backed test utilities  ## License 

@@ -5,12 +5,12 @@ -- see: https://github.com/sol/hpack  name:           atelier-core-version:        0.2.0.0+version:        0.3.0.0 synopsis:       Foundational Effectful-based effects and utilities description:    Core effects and utilities for effect-based applications, built on Effectful — part of the atelier toolkit. category:       Control-homepage:       https://github.com/atelier-hub/tricorder#readme-bug-reports:    https://github.com/atelier-hub/tricorder/issues+homepage:       https://github.com/tweag/tricorder#readme+bug-reports:    https://github.com/tweag/tricorder/issues author:         Christian Georgii maintainer:     christian.georgii@tweag.io license:        MIT@@ -27,7 +27,7 @@  source-repository head   type: git-  location: https://github.com/atelier-hub/tricorder+  location: https://github.com/tweag/tricorder  library   exposed-modules:@@ -64,6 +64,8 @@       Atelier.Effects.Process       Atelier.Effects.Process.Internal       Atelier.Effects.Publishing+      Atelier.Effects.Publishing.Pub+      Atelier.Effects.Publishing.Sub       Atelier.Effects.Tally       Atelier.Effects.Timeout       Atelier.Effects.UUID@@ -165,6 +167,7 @@       Unit.Atelier.Effects.FileWatcherSpec       Unit.Atelier.Effects.IteratorSpec       Unit.Atelier.Effects.LogSpec+      Unit.Atelier.Effects.Publishing.PubSpec       Unit.Atelier.Effects.PublishingSpec       Unit.Atelier.Effects.TallySpec       Unit.Atelier.Effects.YieldSpec@@ -199,7 +202,7 @@   build-depends:       aeson ==2.2.*     , async ==2.2.*-    , atelier-core ==0.2.*+    , atelier-core ==0.3.*     , atelier-prelude ==0.1.*     , base >=4.18 && <4.23     , bytestring >=0.11 && <0.13

@@ -19,6 +19,7 @@       -- * Scope     , Scope (..)     , scoped+    , restartableFork     , restartableForkWith     , restartableForkLoop @@ -87,11 +88,33 @@ makeEffect ''Conc  +-- | Forks an action in a loop. Each time @signal@ returns, the current fork is+-- cancelled, and forked again.+restartableFork+    :: (Conc :> es)+    => Eff es x+    -- ^ Signal action+    -> Eff es a+    -- ^ Cancellable action+    -> Eff es Void+restartableFork signal action = forever $ scoped do+    _ <- fork action+    signal++ -- | Forks an action in a loop, with a setup step that runs in the scope before -- each fork. Each time @signal@ returns, the current fork is cancelled, and -- setup and fork are run again. The setup result is passed to the forked -- action, structurally guaranteeing it completes before the fork starts.-restartableForkWith :: (Conc :> es) => Eff es () -> Eff es r -> (r -> Eff es a) -> Eff es Void+restartableForkWith+    :: (Conc :> es)+    => Eff es x+    -- ^ Signalling action to block on.+    -> Eff es r+    -- ^ Setup action that is run for each iteration+    -> (r -> Eff es a)+    -- ^ Action to run with the result of the setup+    -> Eff es Void restartableForkWith signal setup action = forever $ scoped do     r <- setup     _ <- fork (action r)

@@ -8,13 +8,19 @@     , input     , runInputEff     , runInputConst+    , toReader+    , fromState     ) where  import Effectful (Effect) import Effectful.Dispatch.Dynamic (interpret_)+import Effectful.Reader.Static (Reader, runReader)+import Effectful.State.Static.Shared (State) import Effectful.TH (makeEffect) +import Effectful.State.Static.Shared qualified as State + -- | Request a value from the effect system. -- -- This effect does not carry any@@ -45,3 +51,14 @@ -- 'Effectful.Reader.Local', but is available here for testing. runInputConst :: i -> Eff (Input i : es) a -> Eff es a runInputConst = runInputEff . pure+++-- | Retrieves the value once, and exposes it as a 'Reader' effect.+toReader :: (Input i :> es) => Eff (Reader i : es) a -> Eff es a+toReader act = do+    x <- input+    runReader x act+++fromState :: (State i :> es) => Eff (Input i : es) a -> Eff es a+fromState = interpret_ \Input -> State.get

@@ -52,11 +52,11 @@  import Atelier.Effects.Chan (Chan) import Atelier.Effects.Conc (Conc)-import Atelier.Effects.Publishing (Sub)+import Atelier.Effects.Publishing.Sub (Sub)  import Atelier.Effects.Chan qualified as Chan import Atelier.Effects.Conc qualified as Conc-import Atelier.Effects.Publishing qualified as Sub+import Atelier.Effects.Publishing.Sub qualified as Sub   -- * Yield

@@ -17,11 +17,11 @@  import Atelier.Effects.Chan (Chan) import Atelier.Effects.Conc (Conc)-import Atelier.Effects.Publishing (Sub)+import Atelier.Effects.Publishing.Sub (Sub)  import Atelier.Effects.Chan qualified as Chan import Atelier.Effects.Conc qualified as Conc-import Atelier.Effects.Publishing qualified as Sub+import Atelier.Effects.Publishing.Sub qualified as Sub   -- | A pull-based iterator of (potentially infinite) values.

@@ -28,11 +28,14 @@        -- * Operations     , readProcessStdout+    , readProcess+    , runProcess     , readProcessSafe     , withProcessGroup     , terminateProcessGroup     , interruptProcessGroup     , waitExitCode+    , getExecutablePath        -- * Interpreters     , runProcessIO@@ -58,6 +61,7 @@     , shell     ) +import System.Environment qualified as Env import System.Process.Typed qualified as TP  import Atelier.Effects.Process.Internal (RunningProcess (..))@@ -81,6 +85,15 @@ data Process :: Effect where     -- | Run a process to completion, returning its exit code and captured stdout.     ReadProcessStdout :: ProcessConfig i o e -> Process m (ExitCode, LByteString)+    -- | Run a process to completion, capturing both stdout and stderr along with+    -- its exit code.+    ReadProcess :: ProcessConfig i o e -> Process m (ExitCode, LByteString, LByteString)+    -- | Run a process to completion, returning only its exit code. Unlike+    -- 'ReadProcessStdout' this does not capture stdout.+    RunProcess :: ProcessConfig i o e -> Process m ExitCode+    -- | The absolute path of the currently running executable, for re-invoking+    -- this program as a subprocess.+    GetExecutablePath :: Process m FilePath     -- | Spawn a process and return its handle. Internal; callers use 'withProcessGroup'.     StartProcess :: ProcessConfig i o e -> Process m (RunningProcess i o e)     -- | Terminate the leader and close its streams. Internal; does not reach the@@ -148,6 +161,9 @@ runProcessIO :: (IOE :> es) => Eff (Process : es) a -> Eff es a runProcessIO = interpret_ \case     ReadProcessStdout cfg -> liftIO $ TP.readProcessStdout cfg+    ReadProcess cfg -> liftIO $ TP.readProcess cfg+    RunProcess cfg -> liftIO $ TP.runProcess cfg+    GetExecutablePath -> liftIO Env.getExecutablePath     StartProcess cfg -> liftIO $ RunningProcess <$> TP.startProcess cfg     StopProcess (RunningProcess p) -> liftIO $ TP.stopProcess p     WaitExitCode (RunningProcess p) -> liftIO $ TP.waitExitCode p

@@ -1,144 +1,23 @@--- | A typed publish/subscribe effect pair.------ 'Pub' publishes events of a given type; 'Sub' subscribes and delivers each--- published event to a listener. 'runPubSub' wires the two together over an--- internal broadcast channel, propagating tracing context from publisher to--- listener; 'runPubWriter' instead records published events to a 'Writer', for--- tests.------ Subscriptions are established asynchronously, so a listener you intend to--- publish to should be started with 'forkListener' or 'forkListener_', which--- block until the subscription is live and therefore cannot miss early events. module Atelier.Effects.Publishing-    ( Pub-    , Sub-    , listen-    , listen_-    , listenWith-    , listenWith_-    , listenOnce-    , listenOnce_-    , forkListener-    , forkListener_-    , publish-    , runPubSub-    , runPubWriter+    ( runPubSub+    , runPubSub_     ) where  import Data.Time (UTCTime)-import Effectful (Effect)-import Effectful.Concurrent.STM-    ( Concurrent-    , atomically-    , newEmptyTMVar-    , putTMVar-    , takeTMVar-    )-import Effectful.Dispatch.Dynamic (interpretWith, interpretWith_, interpret_, localSeqUnlift)-import Effectful.Error.Static (runErrorNoCallStack, throwError)-import Effectful.TH (makeEffect)-import Effectful.Writer.Static.Shared (Writer, tell)--import Text.Show qualified as S+import Effectful.Dispatch.Dynamic (interpret, interpretWith, interpretWith_, interpret_, localSeqUnlift)  import Atelier.Effects.Chan (Chan) import Atelier.Effects.Clock (Clock)-import Atelier.Effects.Conc (Conc, fork_) import Atelier.Effects.Monitoring.Tracing (SpanContext, Tracing)+import Atelier.Effects.Publishing.Pub (Pub (..))+import Atelier.Effects.Publishing.Sub (Sub (..))  import Atelier.Effects.Chan qualified as Chan import Atelier.Effects.Clock qualified as Clock import Atelier.Effects.Monitoring.Tracing qualified as Tracing  --- | Effect for publishing events of type @event@.-data Pub (event :: Type) :: Effect where-    -- | Publish an event to all current subscribers.-    Publish :: event -> Pub event m ()----- | Effect for subscribing to events of type @event@.-data Sub (event :: Type) :: Effect where-    -- | Subscribe, then run @onSubscribed@ once the subscription is established-    -- — after the internal channel has been duplicated and before any event is-    -- delivered — and thereafter deliver every published event to the listener,-    -- forever. The @onSubscribed@ hook lets a caller synchronize on "subscribed"-    -- so a concurrently-started publisher cannot race ahead of the subscription-    -- and have its events missed. Most callers want 'listen' (no hook); a caller-    -- that forks the listener and then publishes must wait on this hook first.-    ListenWith :: m () -> (UTCTime -> event -> m ()) -> Sub event m Void---makeEffect ''Pub-makeEffect ''Sub----- | Subscribe and deliver every published event to the listener, forever.--- Defined in terms of 'listenWith' with a no-op subscribed hook.-listen :: (Sub event :> es) => (UTCTime -> event -> Eff es ()) -> Eff es Void-listen = listenWith (pure ())----- | Like 'listen', but the listener ignores the event timestamp.-listen_ :: (Sub event :> es) => (event -> Eff es ()) -> Eff es Void-listen_ listener = listen $ \_timestamp event -> listener event----- | Like 'listen_', but runs @onSubscribed@ once the subscription is--- established and before any event is delivered. See 'ListenWith'.-listenWith_ :: (Sub event :> es) => Eff es () -> (event -> Eff es ()) -> Eff es Void-listenWith_ onSubscribed listener = listenWith onSubscribed $ \_timestamp event -> listener event----- | Fork a background listener and block until it has actually subscribed,--- then return. The listener runs until the enclosing 'Conc' scope closes.------ This is the safe way to start a listener you intend to publish to: a plain--- @'fork_' . 'listen'@ followed by a 'publish' races the subscription (which--- happens asynchronously in the forked thread) and, under scheduler pressure,--- can drop early events and wedge the listener forever. 'forkListener' closes--- that window by waiting on the subscribed hook before returning.-forkListener-    :: forall event es-     . (Conc :> es, Concurrent :> es, Sub event :> es)-    => (UTCTime -> event -> Eff es ())-    -> Eff es ()-forkListener listener = do-    subscribed <- atomically newEmptyTMVar-    fork_ $ listenWith (atomically (putTMVar subscribed ())) listener-    atomically (takeTMVar subscribed)----- | Like 'forkListener', but the listener ignores the timestamp.-forkListener_-    :: forall event es-     . (Conc :> es, Concurrent :> es, Sub event :> es)-    => (event -> Eff es ())-    -> Eff es ()-forkListener_ listener = forkListener @event (\_timestamp event -> listener event)----- | Wait for a single event and then return said event.-listenOnce :: forall event es. (Sub event :> es) => Eff es (UTCTime, event)-listenOnce = do-    res <- runErrorNoCallStack-        $ listen-        $ \timestamp event -> throwError $ OnceEx (timestamp, event)-    case res of-        Left (OnceEx x) -> pure x-        Right v -> absurd v----- | Same as 'listenOnce', but discards the timestamp.-listenOnce_ :: (Sub event :> es) => Eff es event-listenOnce_ = snd <$> listenOnce---data OnceEx ev = OnceEx ev-instance Show (OnceEx ev) where show _ = "OnceEx"-- -- | Internal wrapper for events with trace context data TracedEvent event = TracedEvent     { event :: event@@ -147,8 +26,9 @@     }  --- | Runs Pub and Sub effects with an internal channel for a specific event type.--- Automatically captures span context from the publisher and creates linked spans in listeners.+-- | Runs 'Pub' and 'Sub' effects with an internal channel for a specific event+-- type. Automatically captures span context from the publisher and creates+-- linked spans in listeners. runPubSub     :: forall event es a      . ( Chan :> es@@ -177,9 +57,25 @@     handleSub . handlePub $ action  --- | Handler that uses a provided Writer effect instead of actually publishing.--- Useful for testing and inspecting what events were published.-runPubWriter :: forall event es a. (Writer [event] :> es) => Eff (Pub event : es) a -> Eff es a-runPubWriter =-    interpret_ \case-        Publish event -> tell [event]+-- | Runs 'Pub' and 'Sub' effects with an internal channel for a specific event+-- type.+runPubSub_+    :: forall event es a+     . (Chan :> es, Clock :> es)+    => Eff (Pub event : Sub event : es) a -> Eff es a+runPubSub_ action = do+    (inChan, _) <- Chan.newChan @(UTCTime, event)+    let handlePub = interpret_ \case+            Publish event -> do+                timestamp <- Clock.currentTime+                Chan.writeChan inChan (timestamp, event)++        handleSub = interpret \env -> \case+            ListenWith onSubscribed listener -> localSeqUnlift env \unlift -> do+                chan <- Chan.dupChan inChan+                unlift onSubscribed+                forever do+                    (timestamp, event) <- Chan.readChan chan+                    unlift $ listener timestamp event++    handleSub . handlePub $ action

@@ -0,0 +1,57 @@+module Atelier.Effects.Publishing.Pub+    ( -- * Effect+      Pub (..)+    , publish++      -- * Interpeters+    , runNoOp+    , toWriter+    , map+    , mapM+    , consume+    ) where++import Effectful (Effect)+import Effectful.Dispatch.Dynamic (interpret_)+import Effectful.TH (makeEffect)+import Effectful.Writer.Static.Shared (Writer, tell)+import Prelude hiding (map, mapM)+++-- | Effect for publishing events of type @event@.+data Pub (event :: Type) :: Effect where+    -- | Publish an event to all current subscribers.+    Publish :: event -> Pub event m ()+++makeEffect ''Pub+++runNoOp :: forall event es a. Eff (Pub event : es) a -> Eff es a+runNoOp = interpret_ \(Publish _) -> pure ()+++-- | Handler that uses a provided Writer effect instead of actually publishing.+-- Useful for testing and inspecting what events were published.+toWriter :: forall event es a. (Writer [event] :> es) => Eff (Pub event : es) a -> Eff es a+toWriter =+    interpret_ \case+        Publish event -> tell [event]+++-- | Convert published events of one type into another, utilizing an existing+-- effect in the effect stack.+map :: forall e1 e2 es a. (Pub e2 :> es) => (e1 -> e2) -> Eff (Pub e1 : es) a -> Eff es a+map f = interpret_ \(Publish event) -> publish $ f event+++-- | Convert published events of one type into another with an effectful+-- transformation, utilizing an existing effect int he effect stack.+mapM :: forall e1 e2 es a. (Pub e2 :> es) => (e1 -> Eff es e2) -> Eff (Pub e1 : es) a -> Eff es a+mapM f = interpret_ \(Publish event) -> f event >>= publish+++-- | Perform an effectful action for each published event.+consume :: (event -> Eff es ()) -> Eff (Pub event : es) a -> Eff es a+consume handler = interpret_ \case+    Publish event -> handler event

@@ -0,0 +1,152 @@+module Atelier.Effects.Publishing.Sub+    ( Sub (..)+    , listen+    , listen_+    , listenWith+    , listenWith_+    , listenOnce+    , listenOnce_+    , listenUntil+    , listenUntil_+    , listenUntilM+    , listenUntilM_+    , forkListener+    , forkListener_+    ) where++import Data.Time (UTCTime)+import Effectful (Effect, inject)+import Effectful.Concurrent.STM+    ( Concurrent+    , atomically+    , newEmptyTMVar+    , putTMVar+    , takeTMVar+    )+import Effectful.Error.Static (runErrorNoCallStack, throwError)+import Effectful.TH (makeEffect)++import Text.Show qualified as S++import Atelier.Effects.Conc (Conc, fork_)+++-- | Effect for subscribing to events of type @event@.+data Sub (event :: Type) :: Effect where+    -- | Subscribe, then run @onSubscribed@ once the subscription is established+    -- — after the internal channel has been duplicated and before any event is+    -- delivered — and thereafter deliver every published event to the listener,+    -- forever. The @onSubscribed@ hook lets a caller synchronize on "subscribed"+    -- so a concurrently-started publisher cannot race ahead of the subscription+    -- and have its events missed. Most callers want 'listen' (no hook); a caller+    -- that forks the listener and then publishes must wait on this hook first.+    ListenWith+        :: m ()+        -- ^ @onSubscribed@ hook to better synchronize on "subscribed".+        -> (UTCTime -> event -> m ())+        -- ^ Listener function to react to events.+        -> Sub event m Void+++makeEffect ''Sub+++-- | Subscribe and deliver every published event to the listener, forever.+-- Defined in terms of 'listenWith' with a no-op subscribed hook.+listen :: (Sub event :> es) => (UTCTime -> event -> Eff es ()) -> Eff es Void+listen = listenWith (pure ())+++-- | Like 'listen', but the listener ignores the event timestamp.+listen_ :: (Sub event :> es) => (event -> Eff es ()) -> Eff es Void+listen_ listener = listen $ \_timestamp event -> listener event+++-- | Like 'listen_', but runs @onSubscribed@ once the subscription is+-- established and before any event is delivered. See 'ListenWith'.+listenWith_ :: (Sub event :> es) => Eff es () -> (event -> Eff es ()) -> Eff es Void+listenWith_ onSubscribed listener = listenWith onSubscribed $ \_timestamp event -> listener event+++-- | Listens until the passed function returns @Just a@, returning said @a@+-- with a timestamp.+listenUntil :: (Sub event :> es) => (UTCTime -> event -> Maybe a) -> Eff es (UTCTime, a)+listenUntil f = do+    res <- runErrorNoCallStack+        $ listen+        $ \timestamp event -> whenJust (f timestamp event) \a -> do+            throwError $ OnceEx (timestamp, a)+    case res of+        Left (OnceEx x) -> pure x+        Right v -> absurd v+++listenUntil_ :: (Sub event :> es) => (event -> Maybe a) -> Eff es a+listenUntil_ f = snd <$> listenUntil (\_ -> f)+++-- | Listens until the passed event handler returns @Just a@, returning said+-- @a@ with a timestamp.+listenUntilM :: (Sub event :> es) => (UTCTime -> event -> Eff es (Maybe a)) -> Eff es (UTCTime, a)+listenUntilM f = do+    res <- runErrorNoCallStack+        $ listen+        $ \timestamp event -> whenJustM (inject $ f timestamp event) \a -> do+            throwError $ OnceEx (timestamp, a)+    case res of+        Left (OnceEx x) -> pure x+        Right v -> absurd v+++-- | Listens until the passed event handler returns @Just a@, returning said+-- @a@.+listenUntilM_ :: (Sub event :> es) => (event -> Eff es (Maybe a)) -> Eff es a+listenUntilM_ f = snd <$> listenUntilM (\_ -> f)+++-- | Fork a background listener and block until it has actually subscribed,+-- then return. The listener runs until the enclosing 'Conc' scope closes.+--+-- This is the safe way to start a listener you intend to publish to: a plain+-- @'fork_' . 'listen'@ followed by a 'publish' races the subscription (which+-- happens asynchronously in the forked thread) and, under scheduler pressure,+-- can drop early events and wedge the listener forever. 'forkListener' closes+-- that window by waiting on the subscribed hook before returning.+forkListener+    :: forall event es+     . (Conc :> es, Concurrent :> es, Sub event :> es)+    => (UTCTime -> event -> Eff es ())+    -> Eff es ()+forkListener listener = do+    subscribed <- atomically newEmptyTMVar+    fork_ $ listenWith (atomically (putTMVar subscribed ())) listener+    atomically (takeTMVar subscribed)+++-- | Like 'forkListener', but the listener ignores the timestamp.+forkListener_+    :: forall event es+     . (Conc :> es, Concurrent :> es, Sub event :> es)+    => (event -> Eff es ())+    -> Eff es ()+forkListener_ listener = forkListener @event (\_timestamp event -> listener event)+++-- | Wait for a single event and then return said event.+listenOnce :: forall event es. (Sub event :> es) => Eff es (UTCTime, event)+listenOnce = do+    res <- runErrorNoCallStack+        $ listen+        $ \timestamp event -> throwError $ OnceEx (timestamp, event)+    case res of+        Left (OnceEx x) -> pure x+        Right v -> absurd v+++-- | Same as 'listenOnce', but discards the timestamp.+listenOnce_ :: (Sub event :> es) => Eff es event+listenOnce_ = snd <$> listenOnce+++data OnceEx ev = OnceEx ev+instance Show (OnceEx ev) where show _ = "OnceEx"

@@ -28,9 +28,12 @@ import Atelier.Effects.Clock (Clock, runClock) import Atelier.Effects.Conc (Conc, fork, fork_, runConc, scoped) import Atelier.Effects.Monitoring.Tracing (Tracing, runTracingNoOp)-import Atelier.Effects.Publishing (Pub, Sub, publish, runPubSub)+import Atelier.Effects.Publishing (runPubSub)+import Atelier.Effects.Publishing.Pub (Pub)+import Atelier.Effects.Publishing.Sub (Sub)  import Atelier.Effects.Iterator qualified as Iter+import Atelier.Effects.Publishing.Pub qualified as Pub   spec_ConcTeardownStress :: Spec@@ -78,7 +81,7 @@                     $ Iter.fromEvents @Int \iter -> do                         _ <- fork do                             liftIO (threadDelay publishDelayUs)-                            traverse_ publish [1, 2, 3 :: Int]+                            traverse_ Pub.publish [1, 2, 3 :: Int]                         _ <- replicateM 3 (Iter.next iter)                         pure ()             completed `shouldBe` Just ()

@@ -9,9 +9,12 @@ import Atelier.Effects.Clock (Clock, runClock) import Atelier.Effects.Conc (Conc, fork, runConc) import Atelier.Effects.Monitoring.Tracing (Tracing, runTracingNoOp)-import Atelier.Effects.Publishing (Pub, Sub, publish, runPubSub)+import Atelier.Effects.Publishing (runPubSub)+import Atelier.Effects.Publishing.Pub (Pub)+import Atelier.Effects.Publishing.Sub (Sub)  import Atelier.Effects.Iterator qualified as Iter+import Atelier.Effects.Publishing.Pub qualified as Pub   spec_Iterator :: Spec@@ -28,7 +31,7 @@             Iter.fromEvents @Int \iter -> do                 _ <- fork do                     liftIO $ threadDelay 10-                    publish (42 :: Int)+                    Pub.publish (42 :: Int)                 Iter.next iter         result `shouldBe` 42 @@ -37,7 +40,7 @@             Iter.fromEvents @Int \iter -> do                 _ <- fork do                     liftIO $ threadDelay 10-                    traverse_ publish [1, 2, 3]+                    traverse_ Pub.publish [1, 2, 3]                 replicateM 3 (Iter.next iter)         result `shouldBe` [1, 2, 3] @@ -46,7 +49,7 @@             Iter.fromEvents @Int \iter -> do                 _ <- fork do                     liftIO $ threadDelay 10-                    traverse_ publish [1, 2, 3]+                    traverse_ Pub.publish [1, 2, 3]                 liftIO $ threadDelay 5_000                 replicateM 3 (Iter.next iter)         result `shouldBe` [1, 2, 3]@@ -59,7 +62,7 @@             Iter.fromEvents @Int \iter -> do                 _ <- fork do                     liftIO $ threadDelay 10-                    traverse_ publish [1 .. 4]+                    traverse_ Pub.publish [1 .. 4]                 Iter.next (Iter.filter even iter)         result `shouldBe` 2 @@ -68,7 +71,7 @@             Iter.fromEvents @Int \iter -> do                 _ <- fork do                     liftIO $ threadDelay 10-                    traverse_ publish [1 .. 6]+                    traverse_ Pub.publish [1 .. 6]                 replicateM 3 (Iter.next (Iter.filter even iter))         result `shouldBe` [2, 4, 6] @@ -80,7 +83,7 @@             Iter.fromEvents @Int \iter -> do                 _ <- fork do                     liftIO $ threadDelay 10-                    traverse_ publish [0, 0, 1]+                    traverse_ Pub.publish [0, 0, 1]                 Iter.next (Iter.changes 0 iter)         result `shouldBe` 1 @@ -89,7 +92,7 @@             Iter.fromEvents @Int \iter -> do                 _ <- fork do                     liftIO $ threadDelay 10-                    traverse_ publish [1, 2, 3]+                    traverse_ Pub.publish [1, 2, 3]                 replicateM 3 (Iter.next (Iter.changes 0 iter))         result `shouldBe` [1, 2, 3] @@ -98,7 +101,7 @@             Iter.fromEvents @Int \iter -> do                 _ <- fork do                     liftIO $ threadDelay 10-                    traverse_ publish [0, 1, 0, 2, 0, 3]+                    traverse_ Pub.publish [0, 1, 0, 2, 0, 3]                 replicateM 3 (Iter.next (Iter.changes 0 iter))         result `shouldBe` [1, 2, 3] 

@@ -0,0 +1,53 @@+module Unit.Atelier.Effects.Publishing.PubSpec (spec_Pub) where++import Effectful (runPureEff)+import Effectful.Writer.Static.Shared (execWriter)+import Test.Hspec (Spec, context, describe, it, shouldMatchList)++import Atelier.Effects.Publishing.Pub qualified as Pub+++data TestEvent = TestEvent Text+    deriving stock (Eq, Show)+++spec_Pub :: Spec+spec_Pub = do+    describe "toWriter" do+        context "no events published" do+            it "doesn't record events" do+                let events =+                        runPureEff . execWriter . Pub.toWriter @TestEvent+                            $ pure ()++                events `shouldMatchList` []++        context "events published" do+            it "records events" do+                let events =+                        runPureEff . execWriter . Pub.toWriter @TestEvent $ do+                            Pub.publish $ TestEvent "payload"+                            pure ()++                events `shouldMatchList` [TestEvent "payload"]++    describe "map" do+        it "maps over one event" do+            let events =+                    runPureEff+                        . execWriter+                        . Pub.toWriter @Text+                        . Pub.map show+                        $ Pub.publish @Int 1++            events `shouldMatchList` ["1"]++        it "maps over many events" do+            let events =+                    runPureEff+                        . execWriter+                        . Pub.toWriter @Text+                        . Pub.map show+                        $ traverse Pub.publish [1 .. 10 :: Int]++            events `shouldMatchList` ["1", "2", "3", "4", "5", "6", "7", "8", "9", "10"]

@@ -1,53 +1,36 @@-module Unit.Atelier.Effects.PublishingSpec (spec_Pub) where+module Unit.Atelier.Effects.PublishingSpec (spec_Publishing) where  import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar) import Data.Time (UTCTime, getCurrentTime)-import Effectful (IOE, runEff, runPureEff)+import Effectful (IOE, runEff) import Effectful.Concurrent (Concurrent, runConcurrent)-import Effectful.Writer.Static.Shared (runWriter)-import Test.Hspec (Spec, context, describe, expectationFailure, it, shouldBe)+import Test.Hspec (Spec, describe, it, shouldBe)  import Atelier.Effects.Chan (Chan, runChan) import Atelier.Effects.Clock (Clock, runClock, runClockConst) import Atelier.Effects.Conc (Conc, runConc) import Atelier.Effects.Monitoring.Tracing (Tracing, runTracingNoOp)-import Atelier.Effects.Publishing (Pub, Sub, forkListener, forkListener_, publish, runPubSub, runPubWriter)+import Atelier.Effects.Publishing (runPubSub)+import Atelier.Effects.Publishing.Pub (Pub)+import Atelier.Effects.Publishing.Sub (Sub) +import Atelier.Effects.Publishing.Pub qualified as Pub+import Atelier.Effects.Publishing.Sub qualified as Sub + data TestEvent = TestEvent Text     deriving stock (Eq, Show)  -spec_Pub :: Spec-spec_Pub = do-    describe "Writer implementation" $ do-        context "no events published" do-            it "doesn't record events" $ do-                let ((), events) = runPureEff . runWriter . runPubWriter @TestEvent $ do-                        pure ()--                length events `shouldBe` 0--        context "events published" do-            it "records events" $ do-                let ((), events) = runPureEff . runWriter . runPubWriter @TestEvent $ do-                        publish $ TestEvent "payload"-                        pure ()--                length events `shouldBe` 1-                case events of-                    [TestEvent payload] ->-                        payload `shouldBe` "payload"-                    xs ->-                        expectationFailure $ "Expected 1 TestEvent event, got: " <> show (length xs)--    describe "PubSub implementation" do+spec_Publishing :: Spec+spec_Publishing = do+    describe "runPubSub" do         it "listener receives a published event" do             result <- runPubSubTest $ do                 received <- liftIO newEmptyMVar-                forkListener_ @TestEvent \event ->+                Sub.forkListener_ @TestEvent \event ->                     liftIO $ putMVar received event-                publish (TestEvent "hello")+                Pub.publish (TestEvent "hello")                 liftIO $ takeMVar received             result `shouldBe` TestEvent "hello" @@ -55,9 +38,9 @@             t0 <- getCurrentTime             result <- runPubSubTestWithClock t0 $ do                 received <- liftIO newEmptyMVar-                forkListener @TestEvent \ts _event ->+                Sub.forkListener @TestEvent \ts _event ->                     liftIO $ putMVar received ts-                publish (TestEvent "hello")+                Pub.publish (TestEvent "hello")                 liftIO $ takeMVar received             result `shouldBe` t0 @@ -65,9 +48,9 @@             result <- runPubSubTest $ do                 recv1 <- liftIO newEmptyMVar                 recv2 <- liftIO newEmptyMVar-                forkListener_ @TestEvent \event -> liftIO $ putMVar recv1 event-                forkListener_ @TestEvent \event -> liftIO $ putMVar recv2 event-                publish (TestEvent "hello")+                Sub.forkListener_ @TestEvent \event -> liftIO $ putMVar recv1 event+                Sub.forkListener_ @TestEvent \event -> liftIO $ putMVar recv2 event+                Pub.publish (TestEvent "hello")                 e1 <- liftIO $ takeMVar recv1                 e2 <- liftIO $ takeMVar recv2                 pure (e1, e2)