wolf 0.3.1 → 0.3.2
raw patch · 27 files changed
+222/−1752 lines, 27 filesdep −asyncdep −fast-loggerdep −formattingdep ~base
Dependencies removed: async, fast-logger, formatting, http-conduit, http-types, lens, lifted-base, monad-logger, mtl, mtl-compat, regex-applicative, regex-compat, safe, shelly, system-filepath, tasty, tasty-hunit, template-haskell, text-manipulate, transformers, transformers-base, unordered-containers, zlib
Dependency ranges changed: base
Files
- LICENSE +1/−1
- README.md +0/−20
- Shakefile.hs +3/−69
- main/Act.hs +0/−169
- main/Act2.hs +0/−131
- main/Decide.hs +0/−44
- main/Execute.hs +0/−50
- main/Options.hs +0/−82
- main/Register.hs +0/−44
- src/Network/AWS/Flow.hs +0/−299
- src/Network/AWS/Flow/Env.hs +0/−17
- src/Network/AWS/Flow/Logger.hs +0/−43
- src/Network/AWS/Flow/Prelude.hs +0/−21
- src/Network/AWS/Flow/S3.hs +0/−51
- src/Network/AWS/Flow/SWF.hs +0/−170
- src/Network/AWS/Flow/Types.hs +0/−288
- src/Network/AWS/Flow/Uid.hs +0/−12
- src/Network/AWS/Wolf/Act.hs +2/−0
- src/Network/AWS/Wolf/Ctx.hs +19/−0
- src/Network/AWS/Wolf/Decide.hs +78/−6
- src/Network/AWS/Wolf/S3.hs +1/−0
- src/Network/AWS/Wolf/SWF.hs +48/−2
- src/Network/AWS/Wolf/Types/Ctx.hs +36/−1
- src/Network/AWS/Wolf/Types/Product.hs +11/−17
- test/Test.hs +0/−12
- test/Test/Network/AWS/Flow.hs +0/−29
- wolf.cabal +23/−174
LICENSE view
@@ -1,4 +1,4 @@-Copyright (C) 2015 Swift Navigation Inc.+Copyright (C) 2015-2016 Swift Navigation Inc. Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the
− README.md
@@ -1,20 +0,0 @@-[![Wolf][8]][1]--# [Wolf][1]--[![Package version][2]][3]-[![Build status][4]][5]-[![Dependency status][6]][7]--Wolf is a wrapper around Amazon Simple Workflow Service. More-documentation is coming, but in the meantime see the examples-directory.--[1]: https://github.com/swift-nav/wolf-[2]: https://img.shields.io/hackage/v/wolf.svg?style=flat-[3]: https://hackage.haskell.org/package/wolf-[4]: https://img.shields.io/travis/swift-nav/wolf/master.svg?style=flat-[5]: https://travis-ci.org/swift-nav/wolf-[6]: https://img.shields.io/hackage-deps/v/wolf.svg?style=flat-[7]: http://packdeps.haskellers.com/feed?needle=wolf-[8]: https://cloud.githubusercontent.com/assets/60851/8178609/a077a326-13c4-11e5-9d54-3e417fc6dd6c.jpg
Shakefile.hs view
@@ -2,7 +2,6 @@ {- stack runghc --package basic-prelude- --package directory --package shake -} @@ -14,8 +13,6 @@ import BasicPrelude import Data.Char import Development.Shake-import Development.Shake.FilePath-import System.Directory -- | This file used for version change detection. --@@ -37,13 +34,6 @@ fakeDir :: FilePath fakeDir = buildDir </> "fake" --- | Build directory where docker files are kept.----dockerDir :: Action FilePath-dockerDir = do- dir <- liftIO getCurrentDirectory- return $ buildDir </> takeFileName dir- -- | Fake directory path builder. -- fd :: FilePath -> FilePath@@ -64,33 +54,11 @@ cmdArgs_ :: String -> [String] -> Action () cmdArgs_ c as = unit $ cmd c as --- | Run commands in a dir with return string.----_cmdArgsDir :: FilePath -> String -> [String] -> Action String-_cmdArgsDir d c as = rstrip . fromStdout <$> cmd (Cwd d) c as---- | Run commands in a dir with no return.----cmdArgsDir_ :: FilePath -> String -> [String] -> Action ()-cmdArgsDir_ d c as = unit $ cmd (Cwd d) c as---- | Run docker command in docker dir.----docker :: [String] -> Action ()-docker args = do- dir <- dockerDir- cmdArgsDir_ dir "docker" args- -- | Stack command. -- stack :: [String] -> Action () stack = cmdArgs_ "stack" --- | Stack exec command.----_stackExec :: String -> [String] -> Action ()-_stackExec cmd' args = stack $ "exec" : cmd' : "--" : args- -- | Sylish command. -- stylish :: [String] -> Action ()@@ -121,13 +89,6 @@ touchFile :: FilePath -> Action () touchFile = flip writeFile' mempty --- | Copy a file if changed, creating parent directories.----copyFileChanged' :: FilePath -> FilePath -> Action ()-copyFileChanged' a b = do- liftIO $ createDirectoryIfMissing True $ dropFileName b- copyFileChanged a b- -- | Preprocess a file with m4 -- preprocess :: FilePattern -> FilePath -> Action [(String, String)] -> Rules ()@@ -252,8 +213,9 @@ -- | sanity --- phony "sanity" $- need [ "tests-error", "lint" ]+ fake pats "sanity" $ \_files -> do+ need [ "lint" ]+ stack [ "build", "--fast", "--test", "--no-run-tests", "--ghc-options=-Werror" ] -- | Haskell source rules --@@ -277,33 +239,6 @@ fake pats "lint" $ \files -> lint files --- | Docker rules----dockerRules :: Rules ()-dockerRules = do- let pats =- [ "Dockerfile"- , "Shakefile.hs"- , ".dockerignore"- , "stack.yaml"- , "registrar.cabal"- , "main//*.hs"- , "src//*.hs"- ]-- -- | docker:setup- --- fake pats "docker:setup" $ \files ->- forM_ files $ \file -> do- dir <- dockerDir- copyFileChanged' file $ dir </> file-- -- | docker:build- --- phony "docker:build" $ do- need [ "docker:setup" ]- docker [ "build", "." ]- -- | Main entry point -- main :: IO ()@@ -313,4 +248,3 @@ want [ "tests-error", "lint", "format" ] globalRules hsRules- dockerRules
− main/Act.hs
@@ -1,169 +0,0 @@-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE RecordWildCards #-}-{-# LANGUAGE ScopedTypeVariables #-}-module Act- ( main- ) where--import BasicPrelude hiding (ByteString, find, hash, length,- readFile, (<.>), (</>))-import Codec.Compression.GZip-import Control.Monad.Trans.Resource-import Data.Aeson.Encode-import Data.ByteString (length)-import qualified Data.ByteString.Lazy as BL-import Data.Text (pack, strip)-import Data.Text.Lazy (toStrict)-import Data.Text.Lazy.Builder hiding (fromText)-import Data.Yaml hiding (Parser)-import Filesystem.Path (dropExtension, (<.>))-import Network.AWS.Data.Crypto-import Network.AWS.Flow-import Options-import Options.Applicative hiding (action)-import Shelly hiding (FilePath, bash, (<.>))--data Args = Args- { aConfig :: FilePath- , aQueue :: Queue- , aContainer :: FilePath- , aContainerless :: Maybe String- , aGzip :: Bool- } deriving ( Eq, Read, Show )--args :: Parser Args-args = Args <$>- configFile <*>- (pack <$> queue) <*>- containerFile <*>- containerless <*>- gzip--parser :: ParserInfo Args-parser =- info ( helper <*> args ) $ fullDesc- <> header "act: Workflow activity"- <> progDesc "Workflow activity"--data Container = Container- { cImage :: Text- , cCommand :: Text- , cVolumes :: [Text]- , cDevices :: [Text]- , cEnvironment :: [Text]- , cLink :: [Text]- } deriving ( Eq, Read, Show )--instance FromJSON Container where- parseJSON (Object v) =- Container <$>- v .: "image" <*>- v .: "command" <*>- v .:? "volumes" .!= [] <*>- v .:? "devices" .!= [] <*>- v .:? "environment" .!= [] <*>- v .:? "links" .!= []- parseJSON _ = mzero--data Control = Control- { cUid :: Uid- } deriving ( Eq, Read, Show )--instance ToJSON Control where- toJSON Control{..} = object- [ "run_uid" .= cUid- ]--encodeText :: ToJSON a => a -> Text-encodeText = toStrict . toLazyText . encodeToTextBuilder . toJSON--handler :: MonadBaseControl IO m => m () -> m (Maybe SomeException)-handler a = handle (return . Just) $ a >> return Nothing--exec :: MonadIO m => Args -> Container -> Uid -> Metadata -> [Blob] -> m (Metadata, [Artifact], Maybe SomeException)-exec Args{..} container uid metadata blobs =- shelly $ withDir $ \dir dataDir storeDir -> do- control $ dataDir </> pack "control.json"- storeInput $ storeDir </> pack "input"- dataInput $ dataDir </> pack "input.json"- e <- maybe (docker dataDir storeDir container) (bash dir container) aContainerless- result <- dataOutput $ dataDir </> pack "output.json"- artifacts <- storeOutput $ storeDir </> pack "output"- return (result, artifacts, e) where- withDir action =- withTmpDir $ \dir -> do- mkdir $ dir </> pack "data"- mkdir $ dir </> pack "store"- mkdir $ dir </> pack "store/input"- mkdir $ dir </> pack "store/output"- action dir (dir </> pack "data") (dir </> pack "store")- control file =- writefile file $ encodeText $ Control uid- writeArtifact file blob =- if aGzip then- writeBinary (dropExtension file) $ BL.toStrict $ decompress blob- else- writeBinary file $ BL.toStrict blob- readArtifact dir file =- if aGzip then do- key <- relativeTo dir file- blob <- BL.toStrict . compress . BL.fromStrict <$> readBinary file- return ( toTextIgnore (key <.> "gz")- , hash blob- , fromIntegral $ length blob- , BL.fromStrict blob- )- else do- key <- relativeTo dir file- blob <- readBinary file- return ( toTextIgnore key- , hash blob- , fromIntegral $ length blob- , BL.fromStrict blob- )- dataInput file =- maybe (return ()) (writefile file) metadata- dataOutput file =- catch_sh_maybe (readfile file) where- catch_sh_maybe action =- catch_sh (Just <$> action) $ \(_ :: SomeException) -> return Nothing- storeInput dir =- forM_ blobs $ \(key, blob) -> do- paths <- strip <$> run "dirname" [key]- mkdir_p $ dir </> paths- writeArtifact (dir </> key) blob- storeOutput dir = do- artifacts <- findWhen test_f dir- forM artifacts $ readArtifact dir- docker dataDir storeDir Container{..} =- handler $ do- devices <- forM cDevices $ \device ->- strip <$> run "readlink" ["-f", device]- run_ "docker" $ concat- [["run"]- , concatMap (("--device" :) . return) devices- , concatMap (("--env" :) . return) cEnvironment- , concatMap (("--link" :) . return) cLink- , concatMap (("--volume" :) . return) $- toTextIgnore dataDir <> ":/app/data" :- toTextIgnore storeDir <> ":/app/store" : cVolumes- , [cImage]- , words cCommand- ]- bash dir Container{..} bashDir =- handler $ do- files <- ls $ fromText $ pack bashDir- forM_ files $ flip cp_r dir- cd dir- maybe (return ()) (uncurry $ run_ . fromText) $ uncons $ words cCommand--call :: Args -> IO ()-call Args{..} = do- config <- decodeFile aConfig >>= maybeThrow (userError "Bad Config")- container <- decodeFile aContainer >>= maybeThrow (userError "Bad Container")- env <- flowEnv config- forever $ runResourceT $ runFlowT env $- act aQueue $ exec Args{..} container--main :: IO ()-main = execParser parser >>= call
− main/Act2.hs
@@ -1,131 +0,0 @@-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE RecordWildCards #-}-{-# LANGUAGE ScopedTypeVariables #-}-module Act2- ( main- ) where--import BasicPrelude hiding (ByteString, find, hash, length,- readFile, (<.>), (</>))-import Codec.Compression.GZip-import Control.Monad.Trans.Resource-import Data.Aeson.Encode-import Data.ByteString (length)-import qualified Data.ByteString.Lazy as BL-import Data.Text (pack, strip)-import Data.Text.Lazy (toStrict)-import Data.Text.Lazy.Builder hiding (fromText)-import Data.Yaml hiding (Parser)-import Filesystem.Path (dropExtension, (<.>))-import Network.AWS.Data.Crypto-import Network.AWS.Flow-import Options-import Options.Applicative hiding (action)-import Shelly hiding (FilePath, bash, (<.>))--data Args = Args- { aConfig :: FilePath- , aQueue :: Queue- , aCommandLine :: Text- , aGzipless :: Bool- } deriving ( Eq, Read, Show )--args :: Parser Args-args = Args <$>- configFile <*>- (pack <$> queue) <*>- (pack <$> commandLine) <*>- gzipless--parser :: ParserInfo Args-parser =- info ( helper <*> args ) $ fullDesc- <> header "act: Workflow activity"- <> progDesc "Workflow activity"--data Control = Control- { cUid :: Uid- } deriving ( Eq, Read, Show )--instance ToJSON Control where- toJSON Control{..} = object- [ "run_uid" .= cUid- ]--encodeText :: ToJSON a => a -> Text-encodeText = toStrict . toLazyText . encodeToTextBuilder . toJSON--handler :: MonadBaseControl IO m => m () -> m (Maybe SomeException)-handler a = handle (return . Just) $ a >> return Nothing--exec :: MonadIO m => Args -> Text -> Uid -> Metadata -> [Blob] -> m (Metadata, [Artifact], Maybe SomeException)-exec Args{..} cmdline uid metadata blobs =- shelly $ withDir $ \dir dataDir storeDir -> do- control $ dataDir </> pack "control.json"- storeInput $ storeDir </> pack "input"- dataInput $ dataDir </> pack "input.json"- e <- bash dir- result <- dataOutput $ dataDir </> pack "output.json"- artifacts <- storeOutput $ storeDir </> pack "output"- return (result, artifacts, e) where- withDir action =- withTmpDir $ \dir -> do- mkdir $ dir </> pack "data"- mkdir $ dir </> pack "store"- mkdir $ dir </> pack "store/input"- mkdir $ dir </> pack "store/output"- action dir (dir </> pack "data") (dir </> pack "store")- control file =- writefile file $ encodeText $ Control uid- writeArtifact file blob =- if aGzipless then- writeBinary file $ BL.toStrict blob- else- writeBinary (dropExtension file) $ BL.toStrict $ decompress blob- readArtifact dir file = do- key <- relativeTo dir file- if aGzipless then do- blob <- readBinary file- return ( toTextIgnore key- , hash blob- , fromIntegral $ length blob- , BL.fromStrict blob- )- else do- blob <- BL.toStrict . compress . BL.fromStrict <$> readBinary file- return ( toTextIgnore (key <.> "gz")- , hash blob- , fromIntegral $ length blob- , BL.fromStrict blob- )- dataInput file =- maybe (return ()) (writefile file) metadata- dataOutput file =- catch_sh_maybe (readfile file) where- catch_sh_maybe action =- catch_sh (Just <$> action) $ \(_ :: SomeException) -> return Nothing- storeInput dir =- forM_ blobs $ \(key, blob) -> do- paths <- strip <$> run "dirname" [key]- mkdir_p $ dir </> paths- writeArtifact (dir </> key) blob- storeOutput dir = do- artifacts <- findWhen test_f dir- forM artifacts $ readArtifact dir- bash dir =- handler $ do- bashDir <- pwd- files <- ls bashDir- forM_ files $ flip cp_r dir- cd dir- maybe (return ()) (uncurry $ run_ . fromText) $ uncons $ words cmdline--call :: Args -> IO ()-call Args{..} = do- config <- decodeFile aConfig >>= maybeThrow (userError "Bad Config")- env <- flowEnv config- forever $ runResourceT $ runFlowT env $- act aQueue $ exec Args{..} aCommandLine--main :: IO ()-main = execParser parser >>= call
− main/Decide.hs
@@ -1,44 +0,0 @@-{-# LANGUAGE RecordWildCards #-}--module Decide- ( main- ) where--import BasicPrelude-import Control.Concurrent.Async-import Control.Monad.Trans.Resource-import Data.Yaml hiding (Parser)-import Network.AWS.Flow-import Options-import Options.Applicative--data Args = Args- { aConfig :: FilePath- , aPlan :: FilePath- } deriving ( Eq, Read, Show )--args :: Parser Args-args = Args <$> configFile <*> planFile--parser :: ParserInfo Args-parser =- info ( helper <*> args ) $ fullDesc- <> 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")- plans <- decodePlans aPlan >>= maybeThrow (userError "Bad Plan")- env <- flowEnv config- void $ runConcurrently $ sequenceA $ flip map plans $ \plan ->- Concurrently $ forever $ runResourceT $ runFlowT env $ decide plan--main :: IO ()-main = execParser parser >>= call
− main/Execute.hs
@@ -1,50 +0,0 @@-{-# LANGUAGE RecordWildCards #-}--module Execute- ( main- ) where--import BasicPrelude hiding (readFile)-import Control.Concurrent.Async-import Control.Monad.Trans.Resource-import Data.Text.IO-import Data.Yaml hiding (Parser)-import Network.AWS.Flow-import Options-import Options.Applicative--data Args = Args- { aConfig :: FilePath- , aPlan :: FilePath- , aInput :: Maybe FilePath- } deriving ( Eq, Read, Show )--args :: Parser Args-args = Args <$> configFile <*> planFile <*> inputFile--parser :: ParserInfo Args-parser =- info ( helper <*> args ) $ fullDesc- <> 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")- plans <- decodePlans aPlan >>= maybeThrow (userError "Bad Plan")- input <- readFileMaybe aInput- env <- flowEnv config- 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/Options.hs
@@ -1,82 +0,0 @@-module Options- ( configFile- , planFile- , inputFile- , containerFile- , queue- , containerless- , gzip- , gzipless- , commandLine- ) where--import BasicPrelude-import Options.Applicative--configFile :: Parser String-configFile =- strOption- $ long "config"- <> short 'c'- <> metavar "FILE"- <> help "AWS SWF Service Flow config"--planFile :: Parser String-planFile =- strOption- $ long "plan"- <> short 'p'- <> metavar "FILE"- <> help "AWS SWF Service Flow plan"--inputFile :: Parser (Maybe String)-inputFile =- optional $ strOption- $ long "input"- <> short 'i'- <> metavar "FILE"- <> help "AWS SWF Service Flow input"--containerFile :: Parser String-containerFile =- strOption- $ long "container"- <> short 'x'- <> metavar "FILE"- <> help "AWS SWF Service Flow worker container"--queue :: Parser String-queue =- strOption- $ long "queue"- <> short 'q'- <> metavar "NAME"- <> help "AWS SWF Service Flow queue"--containerless :: Parser (Maybe String)-containerless =- optional $ strOption- $ long "containerless"- <> metavar "DIR"- <> help "Run outside of container in directory"--gzip :: Parser Bool-gzip =- switch- $ long "gzip"- <> help "GZIP contents of artifacts"--gzipless :: Parser Bool-gzipless =- switch- $ long "gzipless"- <> help "disable GZIP contents of artifacts"--commandLine :: Parser String-commandLine =- strOption- $ long "command-line"- <> short 'l'- <> metavar "COMMAND"- <> help "Command to run"-
− main/Register.hs
@@ -1,44 +0,0 @@-{-# LANGUAGE RecordWildCards #-}--module Register- ( main- ) where--import BasicPrelude-import Control.Concurrent.Async-import Control.Monad.Trans.Resource hiding (register)-import Data.Yaml hiding (Parser)-import Network.AWS.Flow-import Options-import Options.Applicative--data Args = Args- { aConfig :: FilePath- , aPlan :: FilePath- } deriving ( Eq, Read, Show )--args :: Parser Args-args = Args <$> configFile <*> planFile--parser :: ParserInfo Args-parser =- info ( helper <*> args ) $ fullDesc- <> 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")- plans <- decodePlans aPlan >>= maybeThrow (userError "Bad Plan")- env <- flowEnv config'- void $ runConcurrently $ sequenceA $ flip map plans $ \plan ->- Concurrently $ runResourceT $ runFlowT env $ register plan--main :: IO ()-main = execParser parser >>= call
− src/Network/AWS/Flow.hs
@@ -1,299 +0,0 @@-{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE RecordWildCards #-}--module Network.AWS.Flow- ( register- , execute- , act- , decide- , flowEnv- , runFlowT- , runDecide- , nextEvent- , select- , maybeThrow- , Uid- , Queue- , Metadata- , Artifact- , Blob- , Task (..)- , Timer (..)- , Start (..)- , Spec (..)- , End (..)- , Plan (..)- ) where--import Network.AWS.Flow.Env-import Network.AWS.Flow.Logger-import Network.AWS.Flow.Prelude hiding (ByteString, Metadata, handle)-import Network.AWS.Flow.S3-import Network.AWS.Flow.SWF-import Network.AWS.Flow.Types-import Network.AWS.Flow.Uid--import Control.Monad.Catch-import Data.Char-import qualified Data.HashMap.Strict as Map-import Data.Text (pack)-import Formatting hiding (string)-import Network.AWS.SWF-import Network.HTTP.Types-import Safe-import Text.Regex.Applicative---- Interface--serviceError :: MonadFlow m => ErrorCode -> Error -> m ()-serviceError code = \case- e@(ServiceError s) ->- unless check $ throwM e where- check =- s ^. serviceStatus == badRequest400 &&- s ^. serviceAbbrev == "SWF" &&- s ^. serviceCode == code- e -> throwM e--register :: MonadFlow m => Plan -> m ()-register Plan{..} = do- logInfo' "event=register"- handle (serviceError "DomainAlreadyExists") registerDomainAction- handle (serviceError "TypeAlreadyExists") $ registerWorkflowTypeAction- (tskName $ strtTask plnStart)- (tskVersion $ strtTask plnStart)- (tskTimeout $ strtTask plnStart)- mapM_ go plnSpecs where- go Work{..} =- handle (serviceError "TypeAlreadyExists") $ registerActivityTypeAction- (tskName wrkTask)- (tskVersion wrkTask)- (tskTimeout wrkTask)- go Sleep{..} =- return ()--execute :: MonadFlow m => Task -> Metadata -> m ()-execute Task{..} input = do- uid <- newUid- logInfo' $ sformat ("event=execute uid=" % stext) uid- startWorkflowExecutionAction uid tskName tskVersion tskQueue input--exitCode :: RE Char Int-exitCode =- many anySym *> string "exit status: " *> num <* many anySym where- num = read . pack <$> many (psym isDigit)--actException :: MonadFlow m => Token -> SomeException -> m ()-actException token e = do- logError' $ sformat ("event=act-exception " % stext) $ show e- maybe' (textToString (show e) =~ exitCode) (respondActivityTaskFailedAction token) $ \code ->- if code == 255 then respondActivityTaskCanceledAction token else- respondActivityTaskFailedAction token--act :: MonadFlow m => Queue -> (Uid -> Metadata -> [Blob] -> m (Metadata, [Artifact], Maybe SomeException)) -> m ()-act queue action = do- logInfo' "event=act"- (token', uid', input) <- pollForActivityTaskAction queue- maybe_ token' $ \token ->- maybe_ uid' $ \uid -> do- 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- (output, artifacts, e) <- 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- maybe (respondActivityTaskCompletedAction token output) (actException token) e--decide :: MonadFlow m => Plan -> m ()-decide plan@Plan{..} = do- logInfo' "event=decide"- (token', events) <- pollForDecisionTaskAction (tskQueue $ strtTask plnStart)- maybe_ token' $ \token -> do- logger <- asks feLogger- decisions <- runDecide logger plan events select- respondDecisionTaskCompletedAction token decisions---- Decisions--runDecide :: Log -> Plan -> [HistoryEvent] -> DecideT m a -> m a-runDecide logger plan events =- runDecideT env where- env = DecideEnv logger plan events findEvent where- findEvent =- flip Map.lookup $ Map.fromList $ flip map events $ \e ->- (e ^. heEventId, e)--nextEvent :: MonadDecide m => [EventType] -> m HistoryEvent-nextEvent ets = do- events <- asks deEvents- maybeThrow (userError "No Next Event") $ flip find events $ \e ->- e ^. heEventType `elem` ets--workNext :: MonadDecide m => Name -> m (Maybe Spec)-workNext name = do- specs <- asks (plnSpecs . dePlan)- return $ tailMay (dropWhile p specs) >>= headMay where- p Work{..} = tskName wrkTask /= name- p _ = True--sleepNext :: MonadDecide m => Name -> m (Maybe Spec)-sleepNext name = do- specs <- asks (plnSpecs . dePlan)- return $ tailMay (dropWhile p specs) >>= headMay where- p Sleep{..} = tmrName slpTimer /= name- p _ = True--select :: MonadDecide m => m [Decision]-select = do- event <- nextEvent [ WorkflowExecutionStarted- , ActivityTaskCompleted- , ActivityTaskFailed- , ActivityTaskCanceled- , TimerFired- , StartChildWorkflowExecutionInitiated ]- case event ^. heEventType of- WorkflowExecutionStarted -> start event- ActivityTaskCompleted -> completed event- ActivityTaskFailed -> failed event- ActivityTaskCanceled -> canceled event- TimerFired -> timer event- StartChildWorkflowExecutionInitiated -> child- _ -> throwM (userError "Unknown Select Event")--start :: MonadDecide m => HistoryEvent -> m [Decision]-start event = do- logInfo' "event=start"- input <- maybeThrow (userError "No Start Information") $ do- attrs <- event ^. heWorkflowExecutionStartedEventAttributes- return $ attrs ^. weseaInput- specs <- asks (plnSpecs . dePlan)- schedule input $ headMay specs--completed :: MonadDecide m => HistoryEvent -> m [Decision]-completed event = do- logInfo' "event=completed"- findEvent <- asks deFindEvent- (input, name) <- maybeThrow (userError "No Completed Information") $ do- attrs <- event ^. heActivityTaskCompletedEventAttributes- event' <- findEvent $ attrs ^. atceaScheduledEventId- attrs' <- event' ^. heActivityTaskScheduledEventAttributes- return (attrs ^. atceaResult, attrs' ^. atseaActivityType ^. atName)- next <- workNext name- schedule input next--failed :: MonadDecide m => HistoryEvent -> m [Decision]-failed _event = do- logInfo' "event=failed"- return [failWorkflowExecutionDecision]--canceled :: MonadDecide m => HistoryEvent -> m [Decision]-canceled _event = do- logInfo' "event=canceled"- return [cancelWorkflowExecutionDecision]--timer :: MonadDecide m => HistoryEvent -> m [Decision]-timer event = do- logInfo' "event=timer"- findEvent <- asks deFindEvent- name <- maybeThrow (userError "No Timer Information") $ do- attrs <- event ^. heTimerFiredEventAttributes- event' <- findEvent $ attrs ^. tfeaStartedEventId- attrs' <- event' ^. heTimerStartedEventAttributes- attrs' ^. tseaControl- event' <- nextEvent [WorkflowExecutionStarted, ActivityTaskCompleted]- case event' ^. heEventType of- WorkflowExecutionStarted -> timerStart event' name- ActivityTaskCompleted -> timerCompleted event' name- _ -> throwM (userError "Unknown Timer Event")--timerStart :: MonadDecide m => HistoryEvent -> Name -> m [Decision]-timerStart event name = do- logInfo' $ sformat ("event=timer-start name=" % stext) name- input <- maybeThrow (userError "No Timer Start Information") $ do- attrs <- event ^. heWorkflowExecutionStartedEventAttributes- return $ attrs ^. weseaInput- next <- sleepNext name- schedule input next--timerCompleted :: MonadDecide m => HistoryEvent -> Name -> m [Decision]-timerCompleted event name = do- logInfo' $ sformat ("event=timer-completed name=" % stext) name- input <- maybeThrow (userError "No Timer Completed Information") $ do- attrs <- event ^. heActivityTaskCompletedEventAttributes- return $ attrs ^. atceaResult- next <- sleepNext name- schedule input next--schedule :: MonadDecide m => Metadata -> Maybe Spec -> m [Decision]-schedule input = maybe (scheduleEnd input) (scheduleSpec input)--scheduleSpec :: MonadDecide m => Metadata -> Spec -> m [Decision]-scheduleSpec input spec = do- uid <- newUid- logInfo' $ sformat ("event=schedule-spec uid=" % stext) uid- case spec of- Work{..} ->- return [scheduleActivityTaskDecision uid- (tskName wrkTask)- (tskVersion wrkTask)- (tskQueue wrkTask)- input]- Sleep{..} ->- return [startTimerDecision uid- (tmrName slpTimer)- (tmrTimeout slpTimer)]--scheduleEnd :: MonadDecide m => Metadata -> m [Decision]-scheduleEnd input = do- logInfo' "event=schedule-end"- end <- asks (plnEnd . dePlan)- case end of- Stop -> return [completeWorkflowExecutionDecision input]- Continue -> scheduleContinue--scheduleContinue :: MonadDecide m => m [Decision]-scheduleContinue = do- logInfo' "event=schedule-continue"- event <- nextEvent [WorkflowExecutionStarted]- input <- maybeThrow (userError "No Continue Start Information") $ do- attrs <- event ^. heWorkflowExecutionStartedEventAttributes- return $ attrs ^. weseaInput- uid <- newUid- task <- asks (strtTask . plnStart . dePlan)- return [startChildWorkflowExecutionDecision uid- (tskName task)- (tskVersion task)- (tskQueue task)- input]--child :: MonadDecide m => m [Decision]-child = do- event <- nextEvent [WorkflowExecutionStarted, ActivityTaskCompleted]- case event ^. heEventType of- WorkflowExecutionStarted -> childStart event- ActivityTaskCompleted -> childCompleted event- _ -> throwM (userError "Unknown Child Event")--childStart :: MonadDecide m => HistoryEvent -> m [Decision]-childStart event = do- logInfo' "event=child-start"- input <- maybeThrow (userError "No Child Start Information") $ do- attrs <- event ^. heWorkflowExecutionStartedEventAttributes- return $ attrs ^. weseaInput- return [completeWorkflowExecutionDecision input]--childCompleted :: MonadDecide m => HistoryEvent -> m [Decision]-childCompleted event = do- logInfo' "event=child-completed"- input <- maybeThrow (userError "No Child Completed Information") $ do- attrs <- event ^. heActivityTaskCompletedEventAttributes- return $ attrs ^. atceaResult- return [completeWorkflowExecutionDecision input]---- Helpers--maybeThrow :: (MonadThrow m, Exception e) => e -> Maybe a -> m a-maybeThrow e = maybe (throwM e) return
− src/Network/AWS/Flow/Env.hs
@@ -1,17 +0,0 @@-{-# LANGUAGE RecordWildCards #-}--module Network.AWS.Flow.Env- ( flowEnv- ) where--import Network.AWS.Flow.Logger-import Network.AWS.Flow.Prelude-import Network.AWS.Flow.Types--import System.IO--flowEnv :: FlowConfig -> IO FlowEnv-flowEnv FlowConfig{..} = do- logger <- newLogger Info stderr- env <- newEnv fcRegion fcCredentials <&> envLogger .~ logger- return $ FlowEnv logStrLn env (fromIntegral fcTimeout) (fromIntegral fcPollTimeout) fcDomain fcBucket fcPrefix
− src/Network/AWS/Flow/Logger.hs
@@ -1,43 +0,0 @@-module Network.AWS.Flow.Logger- ( logStrLn- , logDebug'- , logInfo'- , logWarn'- , logError'- ) where--import Network.AWS.Flow.Prelude--import Control.Monad.Logger-import Data.Time.Clock-import Data.Time.Format-import Data.Version-import Formatting hiding (now)-import Paths_wolf-import System.Log.FastLogger--prefix :: IO LogStr-prefix = do- now <- getCurrentTime- return $ toLogStr $ sformat (string % " name=wolf v=" % string % " ")- (formatTime defaultTimeLocale "%FT%T%z" now)- (showVersion version)--logStrLn :: LogStr -> IO ()-logStrLn s = do- loggerSet <- newStderrLoggerSet defaultBufSize- p <- prefix- mapM_ (pushLogStr loggerSet) [p, s, "\n"]- flushLogStr loggerSet--logDebug' :: MonadLogger m => Text -> m ()-logDebug' = logDebugN--logInfo' :: MonadLogger m => Text -> m ()-logInfo' = logInfoN--logWarn' :: MonadLogger m => Text -> m ()-logWarn' = logWarnN--logError' :: MonadLogger m => Text -> m ()-logError' = logErrorN
− src/Network/AWS/Flow/Prelude.hs
@@ -1,21 +0,0 @@-module Network.AWS.Flow.Prelude- ( module BasicPrelude- , module Control.Lens- , module Control.Monad.Reader- , module Control.Monad.Trans.AWS- , maybe_- , maybe'- ) where--import BasicPrelude hiding (uncons, (<.>))-import Control.Lens-import Control.Monad.Reader-import Control.Monad.Trans.AWS--{-# ANN module ("HLint: ignore Use import/export shortcut"::String) #-}--maybe_ :: Monad m => Maybe a -> (a -> m ()) -> m ()-maybe_ m f = maybe (return ()) f m--maybe' :: Maybe a -> b -> (a -> b) -> b-maybe' m b a = maybe b a m
− src/Network/AWS/Flow/S3.hs
@@ -1,51 +0,0 @@-module Network.AWS.Flow.S3- ( listObjectsAction- , getObjectAction- , putObjectAction- ) where--import Network.AWS.Flow.Prelude hiding (ByteString, hash, stripPrefix)-import Network.AWS.Flow.Types--import Data.Conduit-import Data.Conduit.Binary-import Data.Conduit.List hiding (concatMap, map, mapMaybe)-import Data.Text hiding (concatMap, map)-import Network.AWS.Data.Body-import Network.AWS.Data.Text-import Network.AWS.S3--{-# ANN module ("HLint: ignore Reduce duplication"::String) #-}---- Actions--listObjectsAction :: MonadFlow m => Uid -> m [Text]-listObjectsAction uid = do- timeout' <- asks feTimeout- bucket' <- asks feBucket- prefix <- asks fePrefix- timeout timeout' $ do- rs <- paginate (listObjects (BucketName bucket') &- loPrefix .~ Just (prefix <> "/" <> uid))- $$ consume- return $- mapMaybe (stripPrefix (prefix <> "/" <> uid <> "/") . toText . (^. oKey)) $- concatMap (^. lorsContents) rs--getObjectAction :: MonadFlow m => Uid -> Text -> m Blob-getObjectAction uid key = do- timeout' <- asks feTimeout- bucket' <- asks feBucket- prefix <- asks fePrefix- timeout timeout' $ do- r <- send $ getObject (BucketName bucket') (ObjectKey $ prefix <> "/" <> uid <> "/" <> key)- blob <- sinkBody (r ^. gorsBody) sinkLbs- return (key, blob)--putObjectAction :: MonadFlow m => Uid -> Artifact -> m ()-putObjectAction uid (key, hash, size, blob) = do- timeout' <- asks feTimeout- bucket' <- asks feBucket- prefix <- asks fePrefix- void $ timeout timeout' $- send $ putObject (BucketName bucket') (ObjectKey $ prefix <> "/" <> uid <> "/" <> key) (Hashed $ hashedBody hash size $ sourceLbs blob)
− src/Network/AWS/Flow/SWF.hs
@@ -1,170 +0,0 @@-module Network.AWS.Flow.SWF- ( registerDomainAction- , registerActivityTypeAction- , registerWorkflowTypeAction- , startWorkflowExecutionAction- , pollForActivityTaskAction- , respondActivityTaskCompletedAction- , respondActivityTaskFailedAction- , respondActivityTaskCanceledAction- , pollForDecisionTaskAction- , respondDecisionTaskCompletedAction- , scheduleActivityTaskDecision- , completeWorkflowExecutionDecision- , failWorkflowExecutionDecision- , cancelWorkflowExecutionDecision- , startTimerDecision- , continueAsNewWorkflowExecutionDecision- , startChildWorkflowExecutionDecision- ) where--import Network.AWS.Flow.Prelude hiding (Metadata)-import Network.AWS.Flow.Types--import Data.Conduit-import Data.Conduit.List hiding (concatMap)-import Network.AWS.SWF-import Safe---- Actions--registerDomainAction :: MonadFlow m => m ()-registerDomainAction = do- timeout' <- asks feTimeout- domain <- asks feDomain- void $ timeout timeout' $- send $ registerDomain domain "30"--registerActivityTypeAction :: MonadFlow m => Name -> Version -> Timeout -> m ()-registerActivityTypeAction name version t = do- timeout' <- asks feTimeout- domain <- asks feDomain- void $ timeout timeout' $- send $ registerActivityType domain name version &- ratDefaultTaskHeartbeatTimeout .~ Just "NONE" &- ratDefaultTaskScheduleToCloseTimeout .~ Just "NONE" &- ratDefaultTaskScheduleToStartTimeout .~ Just "60" &- ratDefaultTaskStartToCloseTimeout .~ Just t--registerWorkflowTypeAction :: MonadFlow m => Name -> Version -> Timeout -> m ()-registerWorkflowTypeAction name version t = do- timeout' <- asks feTimeout- domain <- asks feDomain- void $ timeout timeout' $- send $ registerWorkflowType domain name version &- rwtDefaultChildPolicy .~ Just Abandon &- rwtDefaultExecutionStartToCloseTimeout .~ Just t &- rwtDefaultTaskStartToCloseTimeout .~ Just "60"--startWorkflowExecutionAction :: MonadFlow m- => Uid -> Name -> Version -> Queue -> Metadata -> m ()-startWorkflowExecutionAction uid name version queue input = do- timeout' <- asks feTimeout- domain <- asks feDomain- void $ timeout timeout' $- send $ startWorkflowExecution domain uid (workflowType name version) &- sTaskList .~ Just (taskList queue) &- sInput .~ input--pollForActivityTaskAction :: MonadFlow m => Queue -> m (Maybe Token, Maybe Uid, Metadata)-pollForActivityTaskAction queue = do- timeout' <- asks fePollTimeout- domain <- asks feDomain- timeout timeout' $ do- r <- send $ pollForActivityTask domain (taskList queue)- return- ( r ^. pfatrsTaskToken- , (^. weWorkflowId) <$> r ^. pfatrsWorkflowExecution- , r ^. pfatrsInput )--respondActivityTaskCompletedAction :: MonadFlow m => Token -> Metadata -> m ()-respondActivityTaskCompletedAction token result = do- timeout' <- asks feTimeout- void $ timeout timeout' $- send $ respondActivityTaskCompleted token &- ratcResult .~ result--respondActivityTaskFailedAction :: MonadFlow m => Token -> m ()-respondActivityTaskFailedAction token = do- timeout' <- asks feTimeout- void $ timeout timeout' $- send $ respondActivityTaskFailed token--respondActivityTaskCanceledAction :: MonadFlow m => Token -> m ()-respondActivityTaskCanceledAction token = do- timeout' <- asks feTimeout- void $ timeout timeout' $- send $ respondActivityTaskCanceled token--pollForDecisionTaskAction :: MonadFlow m => Queue -> m (Maybe Token, [HistoryEvent])-pollForDecisionTaskAction queue = do- timeout' <- asks fePollTimeout- domain <- asks feDomain- timeout timeout' $ do- rs <- paginate (pollForDecisionTask domain (taskList queue) &- pfdtReverseOrder .~ Just True &- pfdtMaximumPageSize .~ Just 100)- $$ consume- return- ( headMay rs >>= (^. pfdtrsTaskToken)- , concatMap (^. pfdtrsEvents) rs)--respondDecisionTaskCompletedAction :: MonadFlow m => Token -> [Decision] -> m ()-respondDecisionTaskCompletedAction token decisions = do- timeout' <- asks fePollTimeout- void $ timeout timeout' $- send $ respondDecisionTaskCompleted token &- rdtcDecisions .~ decisions---- Decisions--scheduleActivityTaskDecision :: Uid -> Name -> Version -> Queue -> Metadata -> Decision-scheduleActivityTaskDecision uid name version list input =- decision ScheduleActivityTask &- dScheduleActivityTaskDecisionAttributes .~ Just attrs where- attrs = scheduleActivityTaskDecisionAttributes (activityType name version) uid &- satdaTaskList .~ Just (taskList list) &- satdaInput .~ input--completeWorkflowExecutionDecision :: Metadata -> Decision-completeWorkflowExecutionDecision result =- decision CompleteWorkflowExecution &- dCompleteWorkflowExecutionDecisionAttributes .~ Just attrs where- attrs = completeWorkflowExecutionDecisionAttributes &- cwedaResult .~ result--failWorkflowExecutionDecision :: Decision-failWorkflowExecutionDecision =- decision FailWorkflowExecution &- dFailWorkflowExecutionDecisionAttributes .~ Just attrs where- attrs = failWorkflowExecutionDecisionAttributes--cancelWorkflowExecutionDecision :: Decision-cancelWorkflowExecutionDecision =- decision CancelWorkflowExecution &- dCancelWorkflowExecutionDecisionAttributes .~ Just attrs where- attrs = cancelWorkflowExecutionDecisionAttributes--startTimerDecision :: Uid -> Name -> Timeout -> Decision-startTimerDecision uid name t =- decision StartTimer &- dStartTimerDecisionAttributes .~ Just attrs where- attrs = startTimerDecisionAttributes uid t &- stdaControl .~ Just name--continueAsNewWorkflowExecutionDecision :: Version -> Queue -> Metadata -> Decision-continueAsNewWorkflowExecutionDecision version queue input =- decision ContinueAsNewWorkflowExecution &- dContinueAsNewWorkflowExecutionDecisionAttributes .~ Just attrs where- attrs = continueAsNewWorkflowExecutionDecisionAttributes &- canwedaWorkflowTypeVersion .~ Just version &- canwedaTaskList .~ Just (taskList queue) &- canwedaInput .~ input--startChildWorkflowExecutionDecision :: Uid -> Name -> Version -> Queue -> Metadata -> Decision-startChildWorkflowExecutionDecision uid name version queue input =- decision StartChildWorkflowExecution &- dStartChildWorkflowExecutionDecisionAttributes .~ Just attrs where- attrs = startChildWorkflowExecutionDecisionAttributes (workflowType name version) uid &- scwedaTaskList .~ Just (taskList queue) &- scwedaInput .~ input
− src/Network/AWS/Flow/Types.hs
@@ -1,288 +0,0 @@-{-# OPTIONS -fno-warn-orphans #-}-{-# LANGUAGE ConstraintKinds #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE TypeFamilies #-}-{-# LANGUAGE UndecidableInstances #-}--module Network.AWS.Flow.Types where--import Network.AWS.Flow.Prelude hiding (ByteString, catch)--import Control.Monad.Base-import Control.Monad.Catch-import Control.Monad.Logger-import Control.Monad.Trans.Control-import Control.Monad.Trans.Resource-import Data.Aeson-import Data.ByteString.Lazy-import Network.AWS.Data.Crypto-import Network.AWS.SWF.Types--type Uid = Text-type Name = Text-type Version = Text-type Queue = Text-type Token = Text-type Timeout = Text-type Metadata = Maybe Text-type Artifact = (Text, Digest SHA256, Integer, ByteString)-type Blob = (Text, ByteString)-type Log = LogStr -> IO ()--data FlowConfig = FlowConfig- { fcRegion :: Region- , fcCredentials :: Credentials- , fcTimeout :: Int- , fcPollTimeout :: Int- , fcDomain :: Text- , fcBucket :: Text- , fcPrefix :: Text- }--instance FromJSON Region where- parseJSON (String v)- | v == "eu-west-1" = return Ireland- | v == "eu-central-1" = return Frankfurt- | v == "ap-northeast-1" = return Tokyo- | v == "ap-southeast-1" = return Singapore- | v == "ap-southeast-2" = return Sydney- | v == "cn-north-1" = return Beijing- | v == "us-east-1" = return NorthVirginia- | v == "us-west-1" = return NorthCalifornia- | v == "us-west-2" = return Oregon- | v == "us-gov-west-1" = return GovCloud- | v == "fips-us-gov-west-1" = return GovCloudFIPS- | v == "sa-east-1" = return SaoPaulo- | otherwise = mzero- parseJSON _ = mzero--instance FromJSON Credentials where- parseJSON (Object v) =- FromEnv <$>- v .: "access-key-env-var" <*>- v .: "secret-key-env-var" <*>- pure Nothing- parseJSON _ = mzero--instance FromJSON FlowConfig where- parseJSON (Object v) =- FlowConfig <$>- v .: "region" <*>- v .: "credentials" <*>- v .: "timeout" <*>- v .: "poll-timeout" <*>- v .: "domain" <*>- v .: "bucket" <*>- v .: "prefix"- parseJSON _ = mzero--data FlowEnv = FlowEnv- { feLogger :: Log- , feEnv :: Env- , feTimeout :: Seconds- , fePollTimeout :: Seconds- , feDomain :: Text- , feBucket :: Text- , fePrefix :: Text- }--newtype FlowT m a = FlowT- { unFlowT :: LoggingT (AWST' FlowEnv m) a- } deriving ( Functor- , Applicative- , Monad- , MonadIO- , MonadLogger- )--type MonadFlow m =- ( MonadCatch m- , MonadThrow m- , MonadResource m- , MonadLogger m- , MonadReader FlowEnv m- )--instance MonadThrow m => MonadThrow (FlowT m) where- throwM = lift . throwM--instance MonadCatch m => MonadCatch (FlowT m) where- catch (FlowT m) f = FlowT (catch m (unFlowT . f))--instance MonadBase b m => MonadBase b (FlowT m) where- liftBase = liftBaseDefault--instance MonadTrans FlowT where- lift = FlowT . lift . lift--instance MonadTransControl FlowT where- type StT FlowT a = StT (ReaderT FlowEnv) a-- liftWith f = FlowT $- liftWith $ \g ->- liftWith $ \h ->- f (h . g . unFlowT)-- restoreT = FlowT . restoreT . restoreT--instance MonadBaseControl b m => MonadBaseControl b (FlowT m) where- type StM (FlowT m) a = ComposeSt FlowT m a-- liftBaseWith = defaultLiftBaseWith- restoreM = defaultRestoreM--instance MonadResource m => MonadResource (FlowT m) where- liftResourceT = lift . liftResourceT--instance Monad m => MonadReader FlowEnv (FlowT m) where- ask = FlowT ask- local f = FlowT . local f . unFlowT- reader = FlowT . reader--instance HasEnv FlowEnv where- environment = lens feEnv (\s a -> s { feEnv = a })--runFlowT :: FlowEnv -> FlowT m a -> m a-runFlowT e (FlowT m) = runAWST e (runLoggingT m (const . const . const $ feLogger e))--data DecideEnv = DecideEnv- { deLogger :: Log- , dePlan :: Plan- , deEvents :: [HistoryEvent]- , deFindEvent :: Integer -> Maybe HistoryEvent- }--newtype DecideT m a = DecideT- { unDecideT :: LoggingT (ReaderT DecideEnv m) a- } deriving ( Functor- , Applicative- , Monad- , MonadIO- , MonadLogger- )--type MonadDecide m =- ( MonadThrow m- , MonadLogger m- , MonadIO m- , MonadReader DecideEnv m- )--instance MonadThrow m => MonadThrow (DecideT m) where- throwM = lift . throwM--instance MonadBase b m => MonadBase b (DecideT m) where- liftBase = liftBaseDefault--instance MonadTrans DecideT where- lift = DecideT . lift . lift--instance MonadTransControl DecideT where- type StT DecideT a = StT (ReaderT DecideEnv) a-- liftWith f = DecideT $- liftWith $ \g ->- liftWith $ \h ->- f (h . g . unDecideT)-- restoreT = DecideT . restoreT . restoreT--instance MonadBaseControl b m => MonadBaseControl b (DecideT m) where- type StM (DecideT m) a = ComposeSt DecideT m a-- liftBaseWith = defaultLiftBaseWith- restoreM = defaultRestoreM--instance Monad m => MonadReader DecideEnv (DecideT m) where- ask = DecideT ask- local f = DecideT . local f . unDecideT- reader = DecideT . reader--runDecideT :: DecideEnv -> DecideT m a -> m a-runDecideT e (DecideT m) = runReaderT (runLoggingT m (const . const . const $ deLogger e)) e--data Task = Task- { tskName :: Name- , tskVersion :: Version- , tskQueue :: Queue- , tskTimeout :: Timeout- } deriving ( Eq, Read, Show )--instance FromJSON Task where- parseJSON (Object v) =- Task <$>- v .: "name" <*>- v .: "version" <*>- v .: "queue" <*>- v .: "timeout"- parseJSON _ = mzero--data Timer = Timer- { tmrName :: Name- , tmrTimeout :: Timeout- } deriving ( Eq, Read, Show )--instance FromJSON Timer where- parseJSON (Object v) =- Timer <$>- v .: "name" <*>- v .: "timeout"- parseJSON _ = mzero--data Start = Start- { strtTask :: Task- } deriving ( Eq, Read, Show )--instance FromJSON Start where- parseJSON (Object v) =- Start <$>- v .: "flow"- parseJSON _ = mzero--data Spec- = Work- { wrkTask :: Task- }- | Sleep- { slpTimer :: Timer- } deriving ( Eq, Read, Show )--instance FromJSON Spec where- parseJSON (Object v) =- msum- [ Work <$>- v .: "work"- , Sleep <$>- v .: "sleep"- ]- parseJSON _ =- mzero--data End- = Stop- | Continue- deriving ( Eq, Read, Show )--instance FromJSON End where- parseJSON (String v)- | v == "stop" = return Stop- | v == "continue" = return Continue- | otherwise = mzero- parseJSON _ = mzero--data Plan = Plan- { plnStart :: Start- , plnSpecs :: [Spec]- , plnEnd :: End- } deriving ( Eq, Read, Show )--instance FromJSON Plan where- parseJSON (Object v) =- Plan <$>- v .: "start" <*>- v .: "specs" <*>- v .: "end"- parseJSON _ = mzero-
− src/Network/AWS/Flow/Uid.hs
@@ -1,12 +0,0 @@-module Network.AWS.Flow.Uid- ( newUid- ) where--import Network.AWS.Flow.Prelude-import Network.AWS.Flow.Types--import Data.UUID-import Data.UUID.V4--newUid :: MonadIO m => m Uid-newUid = toText <$> liftIO nextRandom
src/Network/AWS/Wolf/Act.hs view
@@ -1,4 +1,6 @@+{-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE NoImplicitPrelude #-}+{-# LANGUAGE OverloadedStrings #-} -- | SWF Actor logic. --
src/Network/AWS/Wolf/Ctx.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE NoImplicitPrelude #-} {-# LANGUAGE OverloadedStrings #-} @@ -12,10 +13,13 @@ , preAmazonStoreCtx , runAmazonWorkCtx , preAmazonWorkCtx+ , runAmazonDecisionCtx+ , preAmazonDecisionCtx ) where import Control.Monad.Trans.AWS import Data.Aeson+import Network.AWS.SWF import Network.AWS.Wolf.Prelude import Network.AWS.Wolf.Types @@ -82,4 +86,19 @@ preAmazonWorkCtx :: MonadAmazonWork c m => Pairs -> TransT AmazonWorkCtx m a -> m a preAmazonWorkCtx preamble action = do c <- view amazonWorkCtx <&> cPreamble <>~ preamble+ runTransT c action++-- | Run amazon decision context.+--+runAmazonDecisionCtx :: MonadAmazon c m => Plan -> [HistoryEvent] -> TransT AmazonDecisionCtx m a -> m a+runAmazonDecisionCtx p hes action = do+ let preamble = [ "name" .= (p ^. pStart ^. tName) ]+ c <- view amazonCtx <&> cPreamble <>~ preamble+ runTransT (AmazonDecisionCtx c p hes) action++-- | Update amazon context's preamble.+--+preAmazonDecisionCtx :: MonadAmazonDecision c m => Pairs -> TransT AmazonDecisionCtx m a -> m a+preAmazonDecisionCtx preamble action = do+ c <- view amazonDecisionCtx <&> cPreamble <>~ preamble runTransT c action
src/Network/AWS/Wolf/Decide.hs view
@@ -1,4 +1,6 @@+{-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE NoImplicitPrelude #-}+{-# LANGUAGE OverloadedStrings #-} -- | SWF Decider logic. --@@ -8,24 +10,94 @@ ) where import Data.Aeson+import Data.UUID+import Data.UUID.V4+import Network.AWS.SWF import Network.AWS.Wolf.Ctx import Network.AWS.Wolf.File import Network.AWS.Wolf.Prelude import Network.AWS.Wolf.SWF import Network.AWS.Wolf.Types +-- | Successful end of workflow.+--+end :: MonadAmazonDecision c m => Maybe Text -> m Decision+end input = do+ traceInfo "end" mempty+ return $ completeWork input++-- | Next activity in workflow to run.+--+next :: MonadAmazonDecision c m => Maybe Text -> Task -> m Decision+next input t = do+ uid <- liftIO $ toText <$> nextRandom+ traceInfo "next" [ "uid" .= uid, "task" .= t ]+ return $ scheduleWork uid (t ^. tName) (t ^. tVersion) (t ^. tQueue) input++-- | Failed activity, stop the workflow.+--+failed :: MonadAmazonDecision c m => m Decision+failed = do+ traceInfo "failed" mempty+ return failWork++-- | Completed activity, start the next activity.+--+completed :: MonadAmazonDecision c m => HistoryEvent -> m Decision+completed he = do+ traceInfo "completed" mempty+ hes <- view adcEvents+ (input, name) <- maybeThrowIO "No Completed Information" $ do+ atcea <- he ^. heActivityTaskCompletedEventAttributes+ he' <- flip find hes $ (== atcea ^. atceaScheduledEventId) . view heEventId+ name <- view atName . view atseaActivityType <$> he' ^. heActivityTaskScheduledEventAttributes+ return (atcea ^. atceaResult, name)+ p <- view adcPlan+ maybe (end input) (next input) $+ join $ fmap headMay $ tailMay $ flip dropWhile (p ^. pTasks) $ (/= name) . view tName++-- | Beginning of workflow, start the first activity.+--+begin :: MonadAmazonDecision c m => HistoryEvent -> m Decision+begin he = do+ traceInfo "begin" mempty+ input <- maybeThrowIO "No Start Information" $+ view weseaInput <$> he ^. heWorkflowExecutionStartedEventAttributes+ p <- view adcPlan+ maybe (end input) (next input) $ headMay (p ^. pTasks)++-- | Schedule workflow based on historical events.+--+schedule :: MonadAmazonDecision c m => m Decision+schedule = do+ traceInfo "schedule" mempty+ hes <- view adcEvents+ f hes >>=+ maybeThrowIO "No Select Information"+ where+ f [] = return Nothing+ f (he:hes) =+ case he ^. heEventType of+ WorkflowExecutionStarted -> Just <$> begin he+ ActivityTaskCompleted -> Just <$> completed he+ ActivityTaskFailed -> Just <$> failed+ _et -> f hes+ -- | Decider logic - poll for decisions, make decisions. -- decide :: MonadConf c m => Plan -> m ()-decide plan =+decide p = preConfCtx [ "label" .= LabelDecide ] $ runAmazonCtx $- runAmazonWorkCtx (plan ^. pStart ^. ptQueue) $ do+ runAmazonWorkCtx (p ^. pStart ^. tQueue) $ do traceInfo "poll" mempty- (token, _events) <- pollDecision- maybe_ token $ \_token' -> do- traceInfo "start" mempty- traceInfo "finish" mempty+ (token, hes) <- pollDecision+ maybe_ token $ \token' ->+ runAmazonDecisionCtx p hes $ do+ traceInfo "start" mempty+ schedule >>=+ completeDecision token'+ traceInfo "finish" mempty -- | Run decider from main with config file. --
src/Network/AWS/Wolf/S3.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE NoImplicitPrelude #-} -- | S3 Calls.
src/Network/AWS/Wolf/SWF.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE NoImplicitPrelude #-} -- | SWF Calls.@@ -7,6 +8,10 @@ , pollDecision , completeActivity , failActivity+ , completeDecision+ , scheduleWork+ , completeWork+ , failWork ) where import Control.Monad.Trans.AWS@@ -37,11 +42,11 @@ tl <- taskList <$> view awcQueue pfdtrs <- paginate (pollForDecisionTask d tl) $$ consume return- ( join $ listToMaybe $ map (view pfdtrsTaskToken) pfdtrs+ ( join $ headMay $ map (view pfdtrsTaskToken) pfdtrs , reverse $ concatMap (view pfdtrsEvents) pfdtrs ) --- | Successfull job completion.+-- | Successful job completion. -- completeActivity :: MonadAmazon c m => Text -> Maybe Text -> m () completeActivity token output =@@ -52,3 +57,44 @@ failActivity :: MonadAmazon c m => Text -> m () failActivity token = void $ send $ respondActivityTaskFailed token++-- | Successful decision completion.+--+completeDecision :: MonadAmazon c m => Text -> Decision -> m ()+completeDecision token d =+ void $ send $ set rdtcDecisions (return d) $ respondDecisionTaskCompleted token++-- | Schedule decision.+--+scheduleWork :: Text -> Text -> Text -> Text -> Maybe Text -> Decision+scheduleWork uid name version queue input =+ decision ScheduleActivityTask &+ dScheduleActivityTaskDecisionAttributes .~ return satda+ where+ satda =+ scheduleActivityTaskDecisionAttributes (activityType name version) uid &+ satdaTaskList .~ return (taskList queue) &+ satdaInput .~ input++-- | Complete decision.+--+completeWork :: Maybe Text -> Decision+completeWork input =+ decision CompleteWorkflowExecution &+ dCompleteWorkflowExecutionDecisionAttributes .~ return cweda+ where+ cweda =+ completeWorkflowExecutionDecisionAttributes &+ cwedaResult .~ input++-- | Failed decision.+--+failWork :: Decision+failWork =+ decision FailWorkflowExecution &+ dFailWorkflowExecutionDecisionAttributes .~ return fweda+ where+ fweda =+ failWorkflowExecutionDecisionAttributes++
src/Network/AWS/Wolf/Types/Ctx.hs view
@@ -1,4 +1,5 @@ {-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE NoImplicitPrelude #-} {-# LANGUAGE TemplateHaskell #-} @@ -7,6 +8,7 @@ module Network.AWS.Wolf.Types.Ctx where import Control.Monad.Trans.AWS+import Network.AWS.SWF import Network.AWS.Wolf.Prelude import Network.AWS.Wolf.Types.Product @@ -91,7 +93,7 @@ , HasAmazonStoreCtx c ) --- | AmazonStoreCtx+-- | AmazonWorkCtx -- -- Amazon work context. --@@ -120,3 +122,36 @@ ( MonadAmazon c m , HasAmazonWorkCtx c )++-- | AmazonDecisionCtx+--+-- Amazon decision context.+--+data AmazonDecisionCtx = AmazonDecisionCtx+ { _adcAmazonCtx :: AmazonCtx+ -- ^ Parent context.+ , _adcPlan :: Plan+ -- ^ Decision plan.+ , _adcEvents :: [HistoryEvent]+ -- ^ History events.+ }++$(makeClassyConstraints ''AmazonDecisionCtx [''HasAmazonCtx])++instance HasAmazonCtx AmazonDecisionCtx where+ amazonCtx = adcAmazonCtx++instance HasConfCtx AmazonDecisionCtx where+ confCtx = amazonCtx . acConfCtx++instance HasCtx AmazonDecisionCtx where+ ctx = confCtx . ccCtx++instance HasEnv AmazonDecisionCtx where+ environment = amazonCtx . acEnv++type MonadAmazonDecision c m =+ ( MonadAmazon c m+ , HasAmazonDecisionCtx c+ )+
src/Network/AWS/Wolf/Types/Product.hs view
@@ -13,15 +13,11 @@ -- SWF and S3 configuration parameters. -- data Conf = Conf- { _cTimeout :: Int- -- ^ SWF regular timeout.- , _cPollTimeout :: Int- -- ^ SWF polling timeout.- , _cDomain :: Text+ { _cDomain :: Text -- ^ SWF domain.- , _cBucket :: Text+ , _cBucket :: Text -- ^ S3 bucket.- , _cPrefix :: Text+ , _cPrefix :: Text -- ^ S3 prefix. } deriving (Show, Eq) @@ -42,28 +38,26 @@ -- -- Work task. ---data PlanTask = PlanTask- { _ptName :: Text+data Task = Task+ { _tName :: Text -- ^ Name of task.- , _ptVersion :: Text+ , _tVersion :: Text -- ^ Version of task.- , _ptQueue :: Text+ , _tQueue :: Text -- ^ Queue for task.- , _ptTimeout :: Text- -- ^ Timeout for task. } deriving (Show, Eq) -$(makeLenses ''PlanTask)-$(deriveJSON spinalOptions ''PlanTask)+$(makeLenses ''Task)+$(deriveJSON spinalOptions ''Task) -- | Plan -- -- Group of tasks. -- data Plan = Plan- { _pStart :: PlanTask+ { _pStart :: Task -- ^ Flow task.- , _pTasks :: [PlanTask]+ , _pTasks :: [Task] -- ^ Worker tasks. } deriving (Show, Eq)
− test/Test.hs
@@ -1,12 +0,0 @@-import BasicPrelude-import qualified Test.Network.AWS.Flow as Flow-import Test.Tasty--tests :: TestTree-tests =- testGroup "Tests"- [ Flow.tests- ]--main :: IO ()-main = defaultMain tests
− test/Test/Network/AWS/Flow.hs
@@ -1,29 +0,0 @@-module Test.Network.AWS.Flow- ( tests- ) where--import Network.AWS.Flow--import BasicPrelude-import Test.Tasty-import Test.Tasty.HUnit--assertUserError :: String -> IO a -> IO ()-assertUserError s action =- handleJust check (const $ return ()) $ do- void action- assertFailure $ "missed user error: " ++ s where- check e = guard $ userError s == e--testSelect :: TestTree-testSelect =- testGroup "Select tests"- [ testCase "No events" $- assertUserError "No Next Event" $ runDecide undefined undefined [] select- ]--tests :: TestTree-tests =- testGroup "Flow tests"- [ testSelect- ]
wolf.cabal view
@@ -1,35 +1,24 @@-name: wolf-version: 0.3.1-synopsis: Amazon Simple Workflow Service Wrapper.-homepage: https://github.com/swift-nav/wolf-license: MIT-license-file: LICENSE-author: Swift Navigation Inc.-maintainer: Mark Fine <dev@swiftnav.com>-copyright: Copyright (C) 2015 Swift Navigation, Inc.-category: Network, AWS, Cloud, Distributed Computing-build-type: Simple-extra-source-files: README.md-cabal-version: >= 1.10--description:- Wolf is a wrapper around Amazon Simple Workflow Service.+name: wolf+version: 0.3.2+synopsis: Amazon Simple Workflow Service Wrapper.+description: Wolf is a wrapper around Amazon Simple Workflow Service.+homepage: https://github.com/swift-nav/wolf+license: MIT+license-file: LICENSE+author: Swift Navigation Inc.+maintainer: Mark Fine <dev@swiftnav.com>+copyright: Copyright (C) 2015-2016 Swift Navigation, Inc.+category: Network, AWS, Cloud, Distributed Computing+build-type: Simple+cabal-version: >= 1.22 source-repository head- type: git- location: git@github.com:swift-nav/wolf.git+ type: git+ location: git@github.com:swift-nav/wolf.git library- exposed-modules: Network.AWS.Flow- , Network.AWS.Wolf- other-modules: Network.AWS.Flow.Env- , Network.AWS.Flow.Logger- , Network.AWS.Flow.Prelude- , Network.AWS.Flow.S3- , Network.AWS.Flow.SWF- , Network.AWS.Flow.Types- , Network.AWS.Flow.Uid- , Network.AWS.Wolf.Act+ exposed-modules: Network.AWS.Wolf+ other-modules: Network.AWS.Wolf.Act , Network.AWS.Wolf.Ctx , Network.AWS.Wolf.Decide , Network.AWS.Wolf.File@@ -40,177 +29,37 @@ , Network.AWS.Wolf.Types.Ctx , Network.AWS.Wolf.Types.Product , Network.AWS.Wolf.Types.Sum- , Paths_wolf default-language: Haskell2010 hs-source-dirs: src- ghc-options: -Wall -fno-warn-orphans+ ghc-options: -Wall build-depends: aeson , amazonka , amazonka-core , amazonka-s3 , amazonka-swf- , base >= 4.7 && < 5- , basic-prelude+ , base >= 4.8 && < 5 , bytestring , conduit , conduit-combinators , conduit-extra , directory , exceptions- , fast-logger , filemanip- , formatting- , http-conduit- , http-types- , lens , lifted-async- , lifted-base , monad-control- , monad-logger- , mtl- , mtl-compat , optparse-applicative , preamble , process- , regex-applicative- , regex-compat , resourcet- , safe- , template-haskell , text- , text-manipulate , time- , transformers- , transformers-base- , unordered-containers , uuid , yaml- default-extensions: OverloadedStrings- FlexibleContexts- RecordWildCards- NoImplicitPrelude -test-suite test- type: exitcode-stdio-1.0- hs-source-dirs: test- main-is: Test.hs- other-modules: Test.Network.AWS.Flow- build-depends: base- , basic-prelude- , tasty- , tasty-hunit- , wolf- ghc-options: -threaded -rtsopts -with-rtsopts=-N -Wall- default-language: Haskell2010- default-extensions: NoImplicitPrelude- OverloadedStrings--executable wolf-register- default-language: Haskell2010- main-is: Register.hs- other-modules: Options- hs-source-dirs: main- ghc-options: -Wall -main-is Register- build-depends: async- , base- , basic-prelude- , optparse-applicative- , resourcet- , text- , wolf- , yaml- default-extensions: OverloadedStrings- RecordWildCards- NoImplicitPrelude--executable wolf-execute- default-language: Haskell2010- main-is: Execute.hs- other-modules: Options- hs-source-dirs: main- ghc-options: -Wall -main-is Execute- build-depends: async- , base- , basic-prelude- , optparse-applicative- , resourcet- , text- , wolf- , yaml- default-extensions: OverloadedStrings- RecordWildCards- NoImplicitPrelude--executable wolf-decide- default-language: Haskell2010- main-is: Decide.hs- other-modules: Options- hs-source-dirs: main- ghc-options: -Wall -main-is Decide- build-depends: async- , base- , basic-prelude- , optparse-applicative- , resourcet- , text- , wolf- , yaml- default-extensions: OverloadedStrings- RecordWildCards- NoImplicitPrelude--executable wolf-act- default-language: Haskell2010- main-is: Act.hs- other-modules: Options- hs-source-dirs: main- ghc-options: -Wall -main-is Act- build-depends: aeson- , amazonka-core- , base- , basic-prelude- , bytestring- , optparse-applicative- , resourcet- , shelly- , system-filepath- , text- , transformers- , wolf- , yaml- , zlib- default-extensions: OverloadedStrings- RecordWildCards- NoImplicitPrelude--executable wolf-act2- default-language: Haskell2010- main-is: Act2.hs- other-modules: Options- hs-source-dirs: main- ghc-options: -Wall -main-is Act2- build-depends: aeson- , amazonka-core- , base- , basic-prelude- , bytestring- , optparse-applicative- , resourcet- , shelly- , system-filepath- , text- , transformers- , wolf- , yaml- , zlib- default-extensions: OverloadedStrings- RecordWildCards- NoImplicitPrelude- executable wolf-actor hs-source-dirs: main main-is: actor.hs- ghc-options: -threaded -rtsopts -with-rtsopts=-N -Wall -O2+ ghc-options: -threaded -rtsopts -with-rtsopts=-N -Wall build-depends: base , wolf , optparse-generic@@ -219,7 +68,7 @@ executable wolf-decider hs-source-dirs: main main-is: decider.hs- ghc-options: -threaded -rtsopts -with-rtsopts=-N -Wall -O2+ ghc-options: -threaded -rtsopts -with-rtsopts=-N -Wall build-depends: base , wolf , optparse-generic@@ -227,8 +76,8 @@ executable shake-wolf main-is: Shakefile.hs- ghc-options: -threaded -rtsopts -with-rtsopts=-N -Wall -O2- build-depends: base >= 4.7 && < 5+ ghc-options: -threaded -rtsopts -with-rtsopts=-N -Wall+ build-depends: base >= 4.8 && < 5 , basic-prelude , directory , shake