dataflower 0.1.2.0 → 0.1.3.0

raw patch · 3 files changed

+136/−2 lines, 3 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• dataflower.cabal +4/−2
• src/Dataflow/Operators.hs +75/−0
• test/Dataflow/OperatorsSpec.hs +57/−0

@@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: b1bed6fdab70f8f031023bf1ce16822eb629e5e6ad13326cddbf6a46c525563d+-- hash: d5432eeb692ca789729a8cbc13ee3678552e761c8f8fc1bcdb0eb05befc63ab4  name:           dataflower-version:        0.1.2.0+version:        0.1.3.0 synopsis:       A Pure-Haskell Timely Dataflow System description:    See README homepage:       https://github.com/doublecrowngaming/dataflower#readme@@ -24,6 +24,7 @@ library   exposed-modules:       Dataflow+      Dataflow.Operators       Test.Dataflow   other-modules:       Dataflow.Primitives@@ -48,6 +49,7 @@   type: exitcode-stdio-1.0   main-is: Spec.hs   other-modules:+      Dataflow.OperatorsSpec       DataflowSpec       Paths_dataflower   hs-source-dirs:

@@ -0,0 +1,75 @@+{-|+Module      : Dataflow+Description : Timely Dataflow for Haskell+Copyright   : (c) Double Crown Gaming Co. 2020+License     : BSD3+Maintainer  : jesse.kempf@doublecrown.co+Stability   : experimental++Common utility operators for data flows++@since 0.1.3.0+-}++module Dataflow.Operators (+  fanout,+  map,+  join2,+  join3+) where++import           Dataflow.Primitives (Dataflow, Edge, StateRef, Timestamp,+                                      Vertex (StatefulVertex), newState,+                                      registerFinalizer, registerVertex, send)+import           Dataflow.Vertices   (statelessVertex)+import           Prelude             (mapM_, ($), (<$>), (<*>))+++-- | Construct a stateless vertex that sends each input to every 'Edge' in the output list.+--+-- @since 0.1.3.0+fanout :: [Edge a] -> Dataflow (Edge a)+fanout nexts = statelessVertex $ \timestamp x -> mapM_ (\next -> send next timestamp x) nexts++-- | Construct a stateless vertex that applies the provided function to every input+-- and sends the result to the output.+--+-- @since 0.1.3.0+map :: (i -> o) -> Edge o -> Dataflow (Edge i)+map f next = statelessVertex $ \timestamp x -> send next timestamp (f x)++-- | Construct a stateful vertex with two input edges.+--+-- @since 0.1.3.0+join2 ::+  state+  -> (StateRef state -> Timestamp -> i -> Dataflow ())+  -> (StateRef state -> Timestamp -> j -> Dataflow ())+  -> (StateRef state -> Timestamp -> Dataflow ())+  -> Dataflow (Edge i, Edge j)+join2 initState callbackI callbackJ finalizer = do+  stateRef <- newState initState++  registerFinalizer $ finalizer stateRef++  (,) <$> registerVertex (StatefulVertex stateRef callbackI)+      <*> registerVertex (StatefulVertex stateRef callbackJ)++-- | Construct a stateful vertex with three input edges.+--+-- @since 0.1.3.0+join3 ::+  state+  -> (StateRef state -> Timestamp -> i -> Dataflow ())+  -> (StateRef state -> Timestamp -> j -> Dataflow ())+  -> (StateRef state -> Timestamp -> k -> Dataflow ())+  -> (StateRef state -> Timestamp -> Dataflow ())+  -> Dataflow (Edge i, Edge j, Edge k)+join3 initState callbackI callbackJ callbackK finalizer = do+  stateRef <- newState initState++  registerFinalizer $ finalizer stateRef++  (,,) <$> registerVertex (StatefulVertex stateRef callbackI)+       <*> registerVertex (StatefulVertex stateRef callbackJ)+       <*> registerVertex (StatefulVertex stateRef callbackK)

@@ -0,0 +1,57 @@+{-# LANGUAGE ScopedTypeVariables #-}++module Dataflow.OperatorsSpec (spec) where++import           Data.Functor.Identity (Identity (..))+import           Data.Semigroup        (Sum (..))+import           Dataflow+import           Dataflow.Operators+import           Prelude               hiding (map)+import qualified Prelude               (map)++import           Test.Dataflow         (runDataflow)+import           Test.Hspec+import           Test.QuickCheck       hiding (discard)+++spec :: Spec+spec = do+  describe "fanout" $+    it "sends all input to each output" $ property $ do+      let fanout' next = fanout [next, next, next]++      \(numbers :: [Int]) -> do+        actual <- runDataflow fanout' numbers+        actual `shouldMatchList` (numbers <> numbers <> numbers)++  describe "map" $+    it "applies a function to all values passing through" $ property $+      \(numbers :: [Int]) -> runDataflow (map (+ 1)) numbers `shouldReturn` Prelude.map (+ 1) numbers++  describe "join2" $+    it "accepts two different kinds of input" $ property $ \numbers -> do+      let numJoin next = join2 (0 :: Int)+                          (\sref _ i            -> modifyState sref (+ i))+                          (\sref _ (Identity j) -> modifyState sref (+ j))+                          (\sref t -> send next t =<< readState sref)+          numbers'     = Prelude.map Identity numbers++      runDataflow (\o -> fst <$> numJoin o) numbers  `shouldReturn` [sum numbers]+      runDataflow (\o -> snd <$> numJoin o) numbers' `shouldReturn` [sum numbers]++  describe "join3" $+    it "accepts three different kinds of input" $ property $ \numbers -> do+      let numJoin next   = join3 (0 :: Int)+                            (\sref _ i            -> modifyState sref (+ i))+                            (\sref _ (Identity j) -> modifyState sref (+ j))+                            (\sref _ (Sum k)      -> modifyState sref (+ k))+                            (\sref t -> send next t =<< readState sref)+          numbers'       = Prelude.map Identity numbers+          numbers''      = Prelude.map Sum numbers+          fst3 (a, _, _) = a+          snd3 (_, b, _) = b+          trd3 (_, _, c) = c++      runDataflow (\o -> fst3 <$> numJoin o) numbers   `shouldReturn` [sum numbers]+      runDataflow (\o -> snd3 <$> numJoin o) numbers'  `shouldReturn` [sum numbers]+      runDataflow (\o -> trd3 <$> numJoin o) numbers'' `shouldReturn` [sum numbers]