packages feed

wolf 0.2.3 → 0.2.5

raw patch · 9 files changed

+72/−50 lines, 9 filesdep +async

Dependencies added: async

Files

main/Act.hs view
@@ -1,5 +1,4 @@ {-# LANGUAGE ScopedTypeVariables #-}- module Act   ( main   ) where@@ -36,7 +35,6 @@   containerFile    <*>   containerless    <*>   gzip-  parser :: ParserInfo Args parser =
main/Decide.hs view
@@ -3,6 +3,7 @@   ) where  import BasicPrelude+import Control.Concurrent.Async import Control.Monad.Trans.Resource import Data.Yaml hiding ( Parser ) import Network.AWS.Flow@@ -23,12 +24,19 @@     <> header   "decide: Decide a workflow"     <> progDesc "Decide a workflow" +decodePlans :: FilePath -> IO (Maybe [Plan])+decodePlans file = do+  plan <- decodeFile file+  plans <- decodeFile file+  return $ plans <|> pure <$> plan+ call :: Args -> IO () call Args{..} = do   config <- decodeFile aConfig >>= maybeThrow (userError "Bad Config")-  plan <- decodeFile aPlan >>= maybeThrow (userError "Bad Plan")+  plans <- decodePlans aPlan >>= maybeThrow (userError "Bad Plan")   env <- flowEnv config-  forever $ runResourceT $ runFlowT env $ decide plan+  void $ runConcurrently $ sequenceA $ flip map plans $ \plan ->+    Concurrently $ forever $ runResourceT $ runFlowT env $ decide plan  main :: IO () main = execParser parser >>= call
main/Execute.hs view
@@ -3,6 +3,7 @@   ) where  import BasicPrelude hiding ( readFile )+import Control.Concurrent.Async import Control.Monad.Trans.Resource import Data.Text.IO import Data.Yaml hiding ( Parser )@@ -25,16 +26,23 @@     <> header   "execute: Execute a workflow"     <> progDesc "Execute a workflow" +decodePlans :: FilePath -> IO (Maybe [Plan])+decodePlans file = do+  plan <- decodeFile file+  plans <- decodeFile file+  return $ plans <|> pure <$> plan+ call :: Args -> IO () call Args{..} = do   config <- decodeFile aConfig >>= maybeThrow (userError "Bad Config")-  plan <- decodeFile aPlan >>= maybeThrow (userError "Bad Plan")+  plans <- decodePlans aPlan >>= maybeThrow (userError "Bad Plan")   input <- readFileMaybe aInput   env <- flowEnv config-  runResourceT $ runFlowT env $-    execute (strtTask $ plnStart plan) input where-      readFileMaybe =-        maybe (return Nothing) ((>>= return . Just) . readFile)+  void $ runConcurrently $ sequenceA $ flip map plans $ \plan ->+    Concurrently $ runResourceT $ runFlowT env $+      execute (strtTask $ plnStart plan) input where+        readFileMaybe =+          maybe (return Nothing) ((>>= return . Just) . readFile)  main :: IO () main = execParser parser >>= call
main/Register.hs view
@@ -3,6 +3,7 @@   ) where  import BasicPrelude+import Control.Concurrent.Async import Control.Monad.Trans.Resource hiding ( register ) import Data.Yaml hiding ( Parser ) import Network.AWS.Flow@@ -23,12 +24,19 @@     <> header   "register: Register a workflow"     <> progDesc "Register a workflow" +decodePlans :: FilePath -> IO (Maybe [Plan])+decodePlans file = do+  plan <- decodeFile file+  plans <- decodeFile file+  return $ plans <|> pure <$> plan+ call :: Args -> IO () call Args{..} = do   config' <- decodeFile aConfig >>= maybeThrow (userError "Bad Config")-  plan <- decodeFile aPlan >>= maybeThrow (userError "Bad Plan")+  plans <- decodePlans aPlan >>= maybeThrow (userError "Bad Plan")   env <- flowEnv config'-  runResourceT $ runFlowT env $ register plan+  void $ runConcurrently $ sequenceA $ flip map plans $ \plan ->+    Concurrently $ runResourceT $ runFlowT env $ register plan  main :: IO () main = execParser parser >>= call
src/Network/AWS/Flow.hs view
@@ -74,16 +74,6 @@   logInfo' $ sformat ("event=execute uid=" % stext) uid   startWorkflowExecutionAction uid tskName tskVersion tskQueue input -serializeError :: MonadFlow m => Error -> m ()-serializeError = \case-  e@(SerializeError s) ->-    unless check $ throwM e where-      check =-        s ^. serializeStatus  == ok200 &&-        s ^. serializeAbbrev  == "SWF" &&-        s ^. serializeMessage == "key \"taskToken\" not present"-  e -> throwM e- actException :: MonadFlow m => Token -> SomeException -> m () actException token e = do   logError' "event=act-exception-begin"@@ -92,31 +82,32 @@   respondActivityTaskFailedAction token  act :: MonadFlow m => Queue -> (Uid -> Metadata -> [Blob] -> m (Metadata, [Artifact])) -> m ()-act queue action =-  handle serializeError $ do-    logInfo' "event=act"-    (token, uid, input) <- pollForActivityTaskAction queue-    logInfo' $ sformat ("event=act-begin uid=" % stext) uid-    keys <- listObjectsAction uid-    unless (null keys) $ logInfo' $ sformat ("event=list-blobs uid=" % stext) uid-    blobs <- forM keys $ getObjectAction uid-    unless (null blobs) $ logInfo' $ sformat ("event=blobs uid=" % stext) uid-    handle (actException token) $ do-      (output, artifacts) <- action uid input blobs-      logInfo' $ sformat ("event=act-finish uid=" % stext) uid-      forM_ artifacts $ putObjectAction uid-      unless (null artifacts) $ logInfo' $ sformat ("event=artifacts uid=" % stext) uid-      respondActivityTaskCompletedAction token output+act queue action = do+  logInfo' "event=act"+  (token', uid, input) <- pollForActivityTaskAction queue+  token <- maybeThrow (userError "No Token") token'+  logInfo' $ sformat ("event=act-begin uid=" % stext) uid+  maybe_ input $ logDebug' . sformat ("event=act-input " % stext)+  keys <- listObjectsAction uid+  unless (null keys) $ logInfo' $ sformat ("event=list-blobs uid=" % stext) uid+  blobs <- forM keys $ getObjectAction uid+  unless (null blobs) $ logInfo' $ sformat ("event=blobs uid=" % stext) uid+  handle (actException token) $ do+    (output, artifacts) <- action uid input blobs+    maybe_ output $ logDebug' . sformat ("event=act-output " % stext)+    logInfo' $ sformat ("event=act-finish uid=" % stext) uid+    forM_ artifacts $ putObjectAction uid+    unless (null artifacts) $ logInfo' $ sformat ("event=artifacts uid=" % stext) uid+    respondActivityTaskCompletedAction token output  decide :: MonadFlow m => Plan -> m ()-decide plan@Plan{..} =-  handle serializeError $ do-    logInfo' "event=decide"-    (token', events) <- pollForDecisionTaskAction (tskQueue $ strtTask plnStart)-    token <- maybeThrow (userError "No Token") token'-    logger <- asks feLogger-    decisions <- runDecide logger plan events select-    respondDecisionTaskCompletedAction token decisions+decide plan@Plan{..} = do+  logInfo' "event=decide"+  (token', events) <- pollForDecisionTaskAction (tskQueue $ strtTask plnStart)+  token <- maybeThrow (userError "No Token") token'+  logger <- asks feLogger+  decisions <- runDecide logger plan events select+  respondDecisionTaskCompletedAction token decisions  -- Decisions
src/Network/AWS/Flow/Prelude.hs view
@@ -3,9 +3,14 @@   , module Control.Lens   , module Control.Monad.Reader   , module Control.Monad.Trans.AWS+  , maybe_   ) where  import BasicPrelude hiding ( (<.>), uncons ) import Control.Lens import Control.Monad.Reader import Control.Monad.Trans.AWS++maybe_ :: Monad m => Maybe a -> (a -> m ()) -> m ()+maybe_ m f = maybe (return ()) f m+
src/Network/AWS/Flow/SWF.hs view
@@ -63,7 +63,7 @@       sTaskList .~ Just (taskList queue) &       sInput .~ input -pollForActivityTaskAction :: MonadFlow m => Queue -> m (Token, Uid, Metadata)+pollForActivityTaskAction :: MonadFlow m => Queue -> m (Maybe Token, Uid, Metadata) pollForActivityTaskAction queue = do   timeout' <- asks fePollTimeout   domain <- asks feDomain@@ -97,7 +97,7 @@       pfdtMaximumPageSize .~ Just 100)         $$ consume     return-      ( liftM (^. pfdtrsTaskToken) (headMay rs)+      ( headMay rs >>= (^. pfdtrsTaskToken)       , concatMap (^. pfdtrsEvents) rs)  respondDecisionTaskCompletedAction :: MonadFlow m => Token -> [Decision] -> m ()
src/Network/AWS/Flow/Types.hs view
@@ -1,3 +1,4 @@+{-# OPTIONS  -fno-warn-orphans          #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE ConstraintKinds            #-} {-# LANGUAGE FlexibleInstances          #-}
wolf.cabal view
@@ -1,5 +1,5 @@ name:                wolf-version:             0.2.3+version:             0.2.5 synopsis:            Amazon Simple Workflow Service Wrapper. homepage:            https://github.com/swift-nav/wolf license:             MIT@@ -88,7 +88,8 @@   other-modules:       Options   hs-source-dirs:      main   ghc-options:         -Wall -main-is Register-  build-depends:       base+  build-depends:       async+                     , base                      , basic-prelude                      , optparse-applicative                      , resourcet@@ -105,7 +106,8 @@   other-modules:       Options   hs-source-dirs:      main   ghc-options:         -Wall -main-is Execute-  build-depends:       base+  build-depends:       async+                     , base                      , basic-prelude                      , optparse-applicative                      , resourcet@@ -122,7 +124,8 @@   other-modules:       Options   hs-source-dirs:      main   ghc-options:         -Wall -main-is Decide-  build-depends:       base+  build-depends:       async+                     , base                      , basic-prelude                      , optparse-applicative                      , resourcet