wolf 0.1.1 → 0.2.0
raw patch · 12 files changed
+464/−537 lines, 12 filesdep +amazonka-coredep +resourcetdep −cryptohashdep ~aesondep ~amazonkadep ~amazonka-s3
Dependencies added: amazonka-core, resourcet
Dependencies removed: cryptohash
Dependency ranges changed: aeson, amazonka, amazonka-s3, amazonka-swf, base, bytestring, conduit, conduit-extra, exceptions, fast-logger, http-conduit, lens, monad-control, monad-logger, mtl, mtl-compat, optparse-applicative, safe, shelly, text, transformers, transformers-base, unordered-containers, uuid, yaml
Files
- README.md +3/−1
- src/Network/AWS/Flow.hs +53/−50
- src/Network/AWS/Flow/Env.hs +6/−63
- src/Network/AWS/Flow/Internal.hs +0/−207
- src/Network/AWS/Flow/S3.hs +11/−12
- src/Network/AWS/Flow/SWF.hs +47/−44
- src/Network/AWS/Flow/Types.hs +203/−45
- src/main/Act.hs +66/−49
- src/main/Decide.hs +13/−16
- src/main/Execute.hs +18/−19
- src/main/Register.hs +10/−13
- wolf.cabal +34/−18
README.md view
@@ -6,7 +6,9 @@ [![Build status][4]][5] [![Dependency status][6]][7] -Wolf is a wrapper around Amazon Simple Workflow Service.+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
src/Network/AWS/Flow.hs view
@@ -1,30 +1,15 @@-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE ConstraintKinds #-}-{-# LANGUAGE RecordWildCards #-}- module Network.AWS.Flow ( register , execute , act , decide+ , flowEnv , runFlowT- , throwStringError- , hoistStringEither- , maybeToFlowError+ , maybeThrow , Uid- , Name- , Version , Queue- , Token- , Timeout , Metadata , Artifact- , FlowConfig (..)- , FlowEnv (..)- , FlowError- , FlowT- , MonadFlow , Task (..) , Timer (..) , Start (..)@@ -33,27 +18,30 @@ , Plan (..) ) where -import Control.Lens ( (^.) )-import Control.Monad ( foldM, forM_ )-import Control.Monad.Reader ( asks )-import Data.List ( find )-import Network.AWS.SWF-import Network.AWS.Flow.Internal-import Network.AWS.Flow.S3-import Network.AWS.Flow.SWF-import Network.AWS.Flow.Types-import Safe ( headMay, tailMay )+import Control.Lens+import Control.Monad.Catch+import Control.Monad.Reader+import qualified Data.HashMap.Strict as Map+import Data.List+import Data.UUID+import Data.UUID.V4+import Network.AWS.SWF+import Network.AWS.Flow.Env+import Network.AWS.Flow.S3+import Network.AWS.Flow.SWF+import Network.AWS.Flow.Types+import Safe -- Interface -register :: MonadFlow m => Plan -> m [()]+register :: MonadFlow m => Plan -> m () register Plan{..} = do r <- registerDomainAction s <- registerWorkflowTypeAction (tskName $ strtTask plnStart) (tskVersion $ strtTask plnStart) (tskTimeout $ strtTask plnStart)- foldM go [s, r] plnSpecs where+ foldM_ go [s, r] plnSpecs where go rs Work{..} = do r <- registerActivityTypeAction (tskName wrkTask)@@ -64,7 +52,7 @@ execute :: MonadFlow m => Task -> Metadata -> m () execute Task{..} input = do- uid <- newUid+ uid <- liftIO $ newUid startWorkflowExecutionAction uid tskName tskVersion tskQueue input act :: MonadFlow m => Queue -> (Uid -> Metadata -> m (Metadata, [Artifact])) -> m ()@@ -76,18 +64,26 @@ decide :: MonadFlow m => Plan -> m () decide plan@Plan{..} = do- (token', events) <- pollForDecisionTaskAction (tskQueue $ strtTask plnStart)- token <- maybeToFlowError "No Token" token'- logger <- asks feLogger- decisions <- runDecide logger plan events select- respondDecisionTaskCompletedAction token decisions+ (token', events) <- pollForDecisionTaskAction (tskQueue $ strtTask plnStart)+ token <- maybeThrow (userError "No Token") token'+ logger <- asks feLogger+ decisions <- runDecide logger plan events select+ respondDecisionTaskCompletedAction token decisions --- Helpers+-- Decisions +runDecide :: Log -> Plan -> [HistoryEvent] -> DecideT m a -> m a+runDecide logger plan events action =+ runDecideT env action 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- maybeToFlowError "No Next Event" $ flip find events $ \e ->+ maybeThrow (userError "No Next Event") $ flip find events $ \e -> e ^. heEventType `elem` ets workNext :: MonadDecide m => Name -> m (Maybe Spec)@@ -115,11 +111,11 @@ ActivityTaskCompleted -> completed event TimerFired -> timer event StartChildWorkflowExecutionInitiated -> child- _ -> throwStringError "Unknown Select Event"+ _ -> throwM (userError "Unknown Select Event") start :: MonadDecide m => HistoryEvent -> m [Decision] start event = do- input <- maybeToFlowError "No Start Information" $ do+ input <- maybeThrow (userError "No Start Information") $ do attrs <- event ^. heWorkflowExecutionStartedEventAttributes return $ attrs ^. weseaInput specs <- asks (plnSpecs . dePlan)@@ -128,7 +124,7 @@ completed :: MonadDecide m => HistoryEvent -> m [Decision] completed event = do findEvent <- asks deFindEvent- (input, name) <- maybeToFlowError "No Completed Information" $ do+ (input, name) <- maybeThrow (userError "No Completed Information") $ do attrs <- event ^. heActivityTaskCompletedEventAttributes event' <- findEvent $ attrs ^. atceaScheduledEventId attrs' <- event' ^. heActivityTaskScheduledEventAttributes@@ -139,7 +135,7 @@ timer :: MonadDecide m => HistoryEvent -> m [Decision] timer event = do findEvent <- asks deFindEvent- name <- maybeToFlowError "No Timer Information" $ do+ name <- maybeThrow (userError "No Timer Information") $ do attrs <- event ^. heTimerFiredEventAttributes event' <- findEvent $ attrs ^. tfeaStartedEventId attrs' <- event' ^. heTimerStartedEventAttributes@@ -148,11 +144,11 @@ case event' ^. heEventType of WorkflowExecutionStarted -> timerStart event' name ActivityTaskCompleted -> timerCompleted event' name- _ -> throwStringError "Unknown Timer Event"+ _ -> throwM (userError "Unknown Timer Event") timerStart :: MonadDecide m => HistoryEvent -> Name -> m [Decision] timerStart event name = do- input <- maybeToFlowError "No Timer Start Information" $ do+ input <- maybeThrow (userError "No Timer Start Information") $ do attrs <- event ^. heWorkflowExecutionStartedEventAttributes return $ attrs ^. weseaInput next <- sleepNext name@@ -160,7 +156,7 @@ timerCompleted :: MonadDecide m => HistoryEvent -> Name -> m [Decision] timerCompleted event name = do- input <- maybeToFlowError "No Timer Completed Information" $ do+ input <- maybeThrow (userError "No Timer Completed Information") $ do attrs <- event ^. heActivityTaskCompletedEventAttributes return $ attrs ^. atceaResult next <- sleepNext name@@ -171,7 +167,7 @@ scheduleSpec :: MonadDecide m => Metadata -> Spec -> m [Decision] scheduleSpec input spec = do- uid <- newUid+ uid <- liftIO $ newUid case spec of Work{..} -> return [scheduleActivityTaskDecision uid@@ -194,10 +190,10 @@ scheduleContinue :: MonadDecide m => m [Decision] scheduleContinue = do event <- nextEvent [WorkflowExecutionStarted]- input <- maybeToFlowError "No Continue Start Information" $ do+ input <- maybeThrow (userError "No Continue Start Information") $ do attrs <- event ^. heWorkflowExecutionStartedEventAttributes return $ attrs ^. weseaInput- uid <- newUid+ uid <- liftIO $ newUid task <- asks (strtTask . plnStart . dePlan) return [startChildWorkflowExecutionDecision uid (tskName task)@@ -211,19 +207,26 @@ case event ^. heEventType of WorkflowExecutionStarted -> childStart event ActivityTaskCompleted -> childCompleted event- _ -> throwStringError "Unknown Child Event"+ _ -> throwM (userError "Unknown Child Event") childStart :: MonadDecide m => HistoryEvent -> m [Decision] childStart event = do- input <- maybeToFlowError "No Child Start Information" $ do+ 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- input <- maybeToFlowError "No Child Completed Information" $ do+ 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++newUid :: IO Uid+newUid = toText <$> nextRandom
src/Network/AWS/Flow/Env.hs view
@@ -1,75 +1,18 @@-{-# LANGUAGE RecordWildCards #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE ConstraintKinds #-}-{-# LANGUAGE OverloadedStrings #-}- module Network.AWS.Flow.Env ( flowEnv ) where -import Control.Applicative ( (<$>), (<*>) )-import Control.Lens ( (.~), (<&>) )-import Control.Monad ( mzero )-import Control.Monad.Except ( runExceptT )+import Control.Lens import Control.Monad.Trans.AWS-import Data.Aeson-import Network.AWS.Flow-import Network.HTTP.Conduit ( conduitManagerSettings- , managerResponseTimeout- , newManager )-import System.Log.FastLogger ( defaultBufSize- , flushLogStr- , newStderrLoggerSet- , pushLogStr )-import System.IO ( stderr )--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"- parseJSON _ = mzero--instance FromJSON FlowConfig where- parseJSON (Object v) =- FlowConfig <$>- v .: "region" <*>- v .: "credentials" <*>- v .: "timeout" <*>- v .: "poll-timeout" <*>- v .: "domain" <*>- v .: "bucket"- parseJSON _ = mzero+import Network.AWS.Flow.Types+import System.Log.FastLogger+import System.IO flowEnv :: FlowConfig -> IO FlowEnv flowEnv FlowConfig{..} = do loggerSet <- newStderrLoggerSet defaultBufSize logger <- newLogger Info stderr- manager <- newManager (managerSettings fcTimeout)- pollManager <- newManager (managerSettings fcPollTimeout)- env <- newEnv' manager <&> envLogger .~ logger- pollEnv <- newEnv' pollManager <&> envLogger .~ logger- return $ FlowEnv (logStrLn loggerSet) env pollEnv fcDomain fcBucket where- managerSettings timeout =- conduitManagerSettings { managerResponseTimeout = Just timeout }- newEnv' m =- runExceptT (newEnv fcRegion fcCredentials m) >>= either error return+ env <- newEnv fcRegion fcCredentials <&> envLogger .~ logger+ return $ FlowEnv (logStrLn loggerSet) env (fromIntegral fcTimeout) (fromIntegral fcPollTimeout) fcDomain fcBucket fcPrefix where logStrLn ls s = pushLogStr ls s >> flushLogStr ls
− src/Network/AWS/Flow/Internal.hs
@@ -1,207 +0,0 @@-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE UndecidableInstances #-}-{-# LANGUAGE TypeFamilies #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE ConstraintKinds #-}--module Network.AWS.Flow.Internal- ( runAWS- , runFlowT- , runDecide- , throwStringError- , hoistStringEither- , maybeToFlowError- , newUid- ) where--import Control.Applicative ( (<$>), (<*>) )-import Control.Lens ( (^.) )-import Control.Monad ( msum, mzero )-import Control.Monad.Base ( MonadBase, liftBase, liftBaseDefault )-import Control.Monad.Except ( MonadError, ExceptT, runExceptT, throwError )-import Control.Monad.IO.Class ( MonadIO, liftIO )-import Control.Monad.Logger ( LogStr, runLoggingT )-import Control.Monad.Reader ( MonadReader, ReaderT, ask, asks, local, runReaderT )-import Control.Monad.Trans.AWS ( AWST, Env, Error, runAWST )-import Control.Monad.Trans.Class ( MonadTrans, lift )-import Control.Monad.Trans.Control ( MonadBaseControl- , MonadTransControl- , StM- , StT- , ComposeSt- , liftBaseWith- , liftWith- , defaultLiftBaseWith- , defaultRestoreM- , restoreM- , restoreT )-import Data.Aeson ( FromJSON, Value(..), parseJSON, (.:) )-import Data.HashMap.Strict ( fromList, lookup )-import Data.Text ( pack )-import Data.UUID ( toString )-import Data.UUID.V4 ( nextRandom )-import Network.AWS.SWF-import Network.AWS.Flow.Types-import Prelude hiding ( lookup )---- FlowT--runFlowT :: MonadIO m => FlowEnv -> FlowT m a -> m (Either FlowError a)-runFlowT e (FlowT k) = runExceptT (runReaderT (runLoggingT k l) e) where- l = const . const . const $ feLogger e--instance MonadBase b m => MonadBase b (FlowT m) where- liftBase = liftBaseDefault--instance MonadBaseControl b m => MonadBaseControl b (FlowT m) where- type StM (FlowT m) a = ComposeSt FlowT m a-- liftBaseWith = defaultLiftBaseWith-- restoreM = defaultRestoreM--instance MonadTrans FlowT where- lift = FlowT . lift . lift . lift--instance MonadTransControl FlowT where- type StT FlowT a =- StT (ExceptT FlowError) (StT (ReaderT FlowEnv) a)-- liftWith f = FlowT $- liftWith $ \g ->- liftWith $ \h ->- liftWith $ \i ->- f (i . h . g . unFlowT)-- restoreT = FlowT . restoreT . restoreT . restoreT--instance Monad m => MonadReader FlowEnv (FlowT m) where- ask = FlowT ask- local f = FlowT . local f . unFlowT---- DecideT--runDecideT :: MonadIO m => DecideEnv -> DecideT m a -> m (Either FlowError a)-runDecideT e (DecideT k) = runExceptT (runReaderT (runLoggingT k l) e) where- l = const . const . const $ deLogger e--instance MonadBase b m => MonadBase b (DecideT m) where- liftBase = liftBaseDefault--instance MonadBaseControl b m => MonadBaseControl b (DecideT m) where- type StM (DecideT m) a = ComposeSt DecideT m a-- liftBaseWith = defaultLiftBaseWith-- restoreM = defaultRestoreM--instance MonadTrans DecideT where- lift = DecideT . lift . lift . lift--instance MonadTransControl DecideT where- type StT DecideT a =- StT (ExceptT FlowError) (StT (ReaderT DecideEnv) a)-- liftWith f = DecideT $- liftWith $ \g ->- liftWith $ \h ->- liftWith $ \i ->- f (i . h . g . unDecideT)-- restoreT = DecideT . restoreT . restoreT . restoreT--instance Monad m => MonadReader DecideEnv (DecideT m) where- ask = DecideT ask- local f = DecideT . local f . unDecideT---- Planning--instance FromJSON Plan where- parseJSON (Object v) =- Plan <$>- v .: "start" <*>- v .: "specs" <*>- v .: "end"- parseJSON _ = mzero--instance FromJSON Spec where- parseJSON (Object v) =- msum- [ Work <$>- v .: "work"- , Sleep <$>- v .: "sleep"- ]- parseJSON _ =- mzero--instance FromJSON End where- parseJSON (String v)- | v == "stop" = return Stop- | v == "continue" = return Continue- | otherwise = mzero- parseJSON _ = mzero--instance FromJSON Start where- parseJSON (Object v) =- Start <$>- v .: "flow"- parseJSON _ = mzero--instance FromJSON Task where- parseJSON (Object v) =- Task <$>- v .: "name" <*>- v .: "version" <*>- v .: "queue" <*>- v .: "timeout"- parseJSON _ = mzero--instance FromJSON Timer where- parseJSON (Object v) =- Timer <$>- v .: "name" <*>- v .: "timeout"- parseJSON _ = mzero---- Helpers--hoistAWSEither :: MonadError FlowError m => Either Error a -> m a-hoistAWSEither = either (throwError . AWSError) return--runAWS :: MonadFlow m => (FlowEnv -> Env) -> AWST m a -> m a-runAWS env action = do- e <- asks env- r <- runAWST e action- hoistAWSEither r--throwStringError :: MonadError FlowError m => String -> m a-throwStringError = throwError . FlowError--hoistStringEither :: MonadError FlowError m => Either String a -> m a-hoistStringEither = either throwStringError return--runDecide :: (MonadError FlowError m, MonadIO m)- => (LogStr -> IO ()) -> Plan -> [HistoryEvent] -> DecideT m a -> m a-runDecide logger plan events action =- runDecideT env action >>= hoistFlowEither- where- env = DecideEnv logger plan events findEvent where- findEvent =- flip lookup $ fromList $ flip map events $ \e ->- (e ^. heEventId, e)- hoistFlowEither = either throwError return--maybeToEither :: e -> Maybe a -> Either e a-maybeToEither e = maybe (Left e) Right--maybeToFlowError :: MonadError FlowError m => String -> Maybe a -> m a-maybeToFlowError e = hoistStringEither . maybeToEither e--newUid :: MonadIO m => m Uid-newUid =- liftIO $ do- r <- nextRandom- return $ pack $ toString r
src/Network/AWS/Flow/S3.hs view
@@ -1,22 +1,21 @@-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE ConstraintKinds #-}- module Network.AWS.Flow.S3 ( putObjectAction ) where -import Control.Monad.Reader ( asks )-import Control.Monad.Trans.AWS ( send_, sourceBody )-import Data.Conduit.Binary ( sourceLbs )+import Network.AWS.Data.Body+import Control.Monad.Reader+import Control.Monad.Trans.AWS+import Data.Conduit.Binary+import Data.Monoid import Network.AWS.Flow.Types-import Network.AWS.Flow.Internal ( runAWS )-import Network.AWS.S3 hiding ( bucket )+import Network.AWS.S3 -- Actions putObjectAction :: MonadFlow m => Artifact -> m () putObjectAction (key, hash, size, blob) = do- bucket <- asks feBucket- runAWS feEnv $- send_ $ putObject body bucket key where- body = sourceBody hash size $ sourceLbs blob+ timeout' <- asks feTimeout+ bucket' <- asks feBucket+ prefix <- asks fePrefix+ void $ timeout timeout' $+ send $ putObject (BucketName $ bucket') (ObjectKey $ prefix <> "/" <> key) (Hashed $ hashedBody hash size $ sourceLbs blob)
src/Network/AWS/Flow/SWF.hs view
@@ -1,7 +1,3 @@-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE ConstraintKinds #-}- module Network.AWS.Flow.SWF ( registerDomainAction , registerActivityTypeAction@@ -19,90 +15,97 @@ , startChildWorkflowExecutionDecision ) where -import Control.Lens ( (^.), (.~), (&) )-import Control.Monad ( liftM )-import Control.Monad.Reader ( asks )-import Control.Monad.Trans.AWS ( paginate, send, send_ )-import Data.Conduit ( ($$) )-import Data.Conduit.List ( consume )+import Control.Lens+import Control.Monad.Reader+import Control.Monad.Trans.AWS hiding ( Metadata )+import Data.Conduit+import Data.Conduit.List hiding ( concatMap ) import Network.AWS.Flow.Types-import Network.AWS.Flow.Internal ( runAWS ) import Network.AWS.SWF-import Safe ( headMay )+import Safe -- Actions registerDomainAction :: MonadFlow m => m () registerDomainAction = do+ timeout' <- asks feTimeout domain <- asks feDomain- runAWS feEnv $- send_ $ registerDomain domain "30"+ void $ timeout timeout' $+ send $ registerDomain domain "30" registerActivityTypeAction :: MonadFlow m => Name -> Version -> Timeout -> m ()-registerActivityTypeAction name version timeout = do+registerActivityTypeAction name version t = do+ timeout' <- asks feTimeout domain <- asks feDomain- runAWS feEnv $- send_ $ registerActivityType domain name version &+ void $ timeout timeout' $+ send $ registerActivityType domain name version & ratDefaultTaskHeartbeatTimeout .~ Just "NONE" & ratDefaultTaskScheduleToCloseTimeout .~ Just "NONE" & ratDefaultTaskScheduleToStartTimeout .~ Just "60" &- ratDefaultTaskStartToCloseTimeout .~ Just timeout+ ratDefaultTaskStartToCloseTimeout .~ Just t registerWorkflowTypeAction :: MonadFlow m => Name -> Version -> Timeout -> m ()-registerWorkflowTypeAction name version timeout = do+registerWorkflowTypeAction name version t = do+ timeout' <- asks feTimeout domain <- asks feDomain- runAWS feEnv $- send_ $ registerWorkflowType domain name version &+ void $ timeout timeout' $+ send $ registerWorkflowType domain name version & rwtDefaultChildPolicy .~ Just Abandon &- rwtDefaultExecutionStartToCloseTimeout .~ Just timeout &+ 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- runAWS feEnv $- send_ $ startWorkflowExecution domain uid (workflowType name version) &- swe1TaskList .~ Just (taskList queue) &- swe1Input .~ input+ void $ timeout timeout' $+ send $ startWorkflowExecution domain uid (workflowType name version) &+ sTaskList .~ Just (taskList queue) &+ sInput .~ input pollForActivityTaskAction :: MonadFlow m => Queue -> m (Token, Uid, Metadata) pollForActivityTaskAction queue = do+ timeout' <- asks fePollTimeout domain <- asks feDomain- runAWS fePollEnv $ do+ timeout timeout' $ do r <- send $ pollForActivityTask domain (taskList queue) return- ( r ^. pfatrTaskToken- , r ^. pfatrWorkflowExecution ^. weWorkflowId- , r ^. pfatrInput )+ ( r ^. pfatrsTaskToken+ , r ^. pfatrsWorkflowExecution ^. weWorkflowId+ , r ^. pfatrsInput ) respondActivityTaskCompletedAction :: MonadFlow m => Token -> Metadata -> m ()-respondActivityTaskCompletedAction token result =- runAWS feEnv $- send_ $ respondActivityTaskCompleted token &+respondActivityTaskCompletedAction token result = do+ timeout' <- asks feTimeout+ void $ timeout timeout' $+ send $ respondActivityTaskCompleted token & ratcResult .~ result respondActivityTaskFailedAction :: MonadFlow m => Token -> m ()-respondActivityTaskFailedAction token =- runAWS feEnv $- send_ $ respondActivityTaskFailed token+respondActivityTaskFailedAction token = do+ timeout' <- asks feTimeout+ void $ timeout timeout' $+ send $ respondActivityTaskFailed token pollForDecisionTaskAction :: MonadFlow m => Queue -> m (Maybe Token, [HistoryEvent]) pollForDecisionTaskAction queue = do+ timeout' <- asks fePollTimeout domain <- asks feDomain- runAWS fePollEnv $ do+ timeout timeout' $ do rs <- paginate (pollForDecisionTask domain (taskList queue) & pfdtReverseOrder .~ Just True & pfdtMaximumPageSize .~ Just 100) $$ consume return- ( liftM (^. pfdtrTaskToken) (headMay rs)- , concatMap (^. pfdtrEvents) rs)+ ( liftM (^. pfdtrsTaskToken) (headMay rs)+ , concatMap (^. pfdtrsEvents) rs) respondDecisionTaskCompletedAction :: MonadFlow m => Token -> [Decision] -> m ()-respondDecisionTaskCompletedAction token decisions =- runAWS feEnv $- send_ $ respondDecisionTaskCompleted token &+respondDecisionTaskCompletedAction token decisions = do+ timeout' <- asks fePollTimeout+ void $ timeout timeout' $+ send $ respondDecisionTaskCompleted token & rdtcDecisions .~ decisions -- Decisions@@ -123,10 +126,10 @@ cwedaResult .~ result startTimerDecision :: Uid -> Name -> Timeout -> Decision-startTimerDecision uid name timeout =+startTimerDecision uid name t = decision StartTimer & dStartTimerDecisionAttributes .~ Just attrs where- attrs = startTimerDecisionAttributes uid timeout &+ attrs = startTimerDecisionAttributes uid t & stdaControl .~ Just name continueAsNewWorkflowExecutionDecision :: Version -> Queue -> Metadata -> Decision
src/Network/AWS/Flow/Types.hs view
@@ -1,21 +1,27 @@ {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE TypeFamilies #-} module Network.AWS.Flow.Types where -import Control.Applicative ( Applicative )-import Control.Monad.Catch ( MonadCatch, MonadThrow )-import Control.Monad.IO.Class ( MonadIO )-import Control.Monad.Except ( ExceptT, MonadError )-import Control.Monad.Logger ( LoggingT, MonadLogger, LogStr )-import Control.Monad.Reader ( ReaderT, MonadReader )-import Control.Monad.Trans.AWS ( Credentials, Env, Error, Region )-import Control.Monad.Trans.Control ( MonadBaseControl )-import Crypto.Hash ( Digest, SHA256 )-import Data.ByteString.Lazy ( ByteString )-import Data.Int ( Int64 )-import Data.Text ( Text )-import Network.AWS.SWF.Types ( HistoryEvent )+import Control.Lens+import Control.Monad.Base+import Control.Monad.Catch+import Control.Monad.Except+import Control.Monad.Logger+import Control.Monad.Reader+import Control.Monad.Trans.Control+import Control.Monad.Trans.Resource+import Control.Monad.Trans.AWS+import Data.Aeson+import Data.ByteString.Lazy+import Data.Conduit.Lazy+import Data.Text+import Network.AWS.Data.Crypto+import Network.AWS.SWF.Types type Uid = Text type Name = Text@@ -24,7 +30,8 @@ type Token = Text type Timeout = Text type Metadata = Maybe Text-type Artifact = (Text, Digest SHA256, Int64, ByteString)+type Artifact = (Text, Digest SHA256, Integer, ByteString)+type Log = LogStr -> IO () data FlowConfig = FlowConfig { fcRegion :: Region@@ -33,70 +40,172 @@ , 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 :: LogStr -> IO ()- , feEnv :: Env- , fePollEnv :: Env- , feDomain :: Text- , feBucket :: Text+ { feLogger :: Log+ , feEnv :: Env+ , feTimeout :: Seconds+ , fePollTimeout :: Seconds+ , feDomain :: Text+ , feBucket :: Text+ , fePrefix :: Text } -data FlowError- = FlowError String- | AWSError Error- deriving ( Show )- newtype FlowT m a = FlowT- { unFlowT :: LoggingT (ReaderT FlowEnv (ExceptT FlowError m)) a+ { unFlowT :: AWST' FlowEnv m a } deriving ( Functor , Applicative , Monad , MonadIO- , MonadLogger- , MonadCatch- , MonadThrow- , MonadError FlowError+ , MonadActive+ , MonadTrans ) type MonadFlow m =- ( MonadBaseControl IO m- , MonadCatch m- , MonadIO m- , MonadLogger m+ ( MonadCatch m+ , MonadThrow m+ , MonadResource m , MonadReader FlowEnv m- , MonadError FlowError 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 MonadTransControl FlowT where+ type StT FlowT a = StT (ReaderT FlowEnv) a++ liftWith = defaultLiftWith FlowT unFlowT+ restoreT = defaultRestoreT FlowT++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 MonadError e m => MonadError e (FlowT m) where+ throwError = lift . throwError+ catchError m f = FlowT (catchError (unFlowT m) (unFlowT . f))++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 m+ data DecideEnv = DecideEnv- { deLogger :: LogStr -> IO ()+ { deLogger :: Log , dePlan :: Plan , deEvents :: [HistoryEvent] , deFindEvent :: Integer -> Maybe HistoryEvent } newtype DecideT m a = DecideT- { unDecideT :: LoggingT (ReaderT DecideEnv (ExceptT FlowError m)) a+ { unDecideT :: ReaderT DecideEnv m a } deriving ( Functor , Applicative , Monad , MonadIO- , MonadLogger- , MonadCatch- , MonadThrow- , MonadError FlowError+ , MonadActive+ , MonadTrans ) type MonadDecide m =- ( MonadBaseControl IO m- , MonadCatch m- , MonadIO m- , MonadLogger m+ ( MonadCatch m+ , MonadThrow m+ , MonadResource m , MonadReader DecideEnv m- , MonadError FlowError m ) +instance MonadThrow m => MonadThrow (DecideT m) where+ throwM = lift . throwM++instance MonadCatch m => MonadCatch (DecideT m) where+ catch (DecideT m) f = DecideT (catch m (unDecideT . f))++instance MonadBase b m => MonadBase b (DecideT m) where+ liftBase = liftBaseDefault++instance MonadTransControl DecideT where+ type StT DecideT a = StT (ReaderT DecideEnv) a++ liftWith = defaultLiftWith DecideT unDecideT+ restoreT = defaultRestoreT DecideT++instance MonadBaseControl b m => MonadBaseControl b (DecideT m) where+ type StM (DecideT m) a = ComposeSt DecideT m a++ liftBaseWith = defaultLiftBaseWith+ restoreM = defaultRestoreM++instance MonadResource m => MonadResource (DecideT m) where+ liftResourceT = lift . liftResourceT++instance MonadError e m => MonadError e (DecideT m) where+ throwError = lift . throwError+ catchError m f = DecideT (catchError (unDecideT m) (unDecideT . f))++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 m e+ data Task = Task { tskName :: Name , tskVersion :: Version@@ -104,15 +213,37 @@ , 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@@ -121,13 +252,40 @@ { 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/main/Act.hs view
@@ -1,38 +1,25 @@-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-} -module Act ( main ) where+module Act+ ( main+ ) where -import Control.Exception ( SomeException )-import Control.Monad ( forever, forM, mzero, liftM )-import Control.Monad.IO.Class ( MonadIO )-import Crypto.Hash ( hash )-import Data.ByteString ( length )-import Data.ByteString.Lazy ( fromStrict )-import Data.Text ( Text, pack, append, words )+import Control.Exception+import Control.Monad+import Control.Monad.Trans.Resource+import Control.Monad.IO.Class+import Data.Aeson.Encode+import Data.ByteString ( length )+import Data.ByteString.Lazy ( fromStrict )+import Data.Text ( Text, pack, append, words, strip )+import Data.Text.Lazy ( toStrict )+import Data.Text.Lazy.Builder import Data.Yaml-import Network.AWS.Flow ( Artifact, Metadata, Queue, Uid, runFlowT, act )-import Network.AWS.Flow.Env ( flowEnv )+import Network.AWS.Data.Crypto+import Network.AWS.Flow import Options.Applicative hiding ( action )-import Shelly hiding ( FilePath )-import Prelude hiding ( length, readFile, words, writeFile )--data Container = Container- { cImage :: Text- , cCommand :: Text- , cVolumes :: [Text]- , cDevices :: [Text]- } deriving ( Eq, Read, Show )--instance FromJSON Container where- parseJSON (Object v) =- Container <$>- v .: "image" <*>- v .: "command" <*>- v .:? "volumes" .!= [] <*>- v .:? "devices" .!= []- parseJSON _ = mzero+import Shelly hiding ( FilePath )+import Prelude hiding ( length, readFile, words, writeFile ) data Args = Args { aConfig :: FilePath@@ -68,9 +55,41 @@ , aContainer = container } +data Container = Container+ { cImage :: Text+ , cCommand :: Text+ , cVolumes :: [Text]+ , cDevices :: [Text]+ , cEnvironment :: [Text]+ } deriving ( Eq, Read, Show )++instance FromJSON Container where+ parseJSON (Object v) =+ Container <$>+ v .: "image" <*>+ v .: "command" <*>+ v .:? "volumes" .!= [] <*>+ v .:? "devices" .!= [] <*>+ v .:? "environment" .!= []+ 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+ exec :: MonadIO m => Container -> Uid -> Metadata -> m (Metadata, [Artifact]) exec container uid metadata = shelly $ withDir $ \dataDir storeDir -> do+ control dataDir $ encodeText $ Control uid input dataDir metadata docker dataDir storeDir container result <- output dataDir@@ -81,12 +100,14 @@ mkdir $ dir </> pack "data" mkdir $ dir </> pack "store" action (dir </> pack "data") (dir </> pack "store")+ control dir =+ writefile (dir </> pack "control.json") input dir =- maybe_ $ writefile $ dir </> pack "input.json" where+ maybe_ (writefile $ dir </> pack "input.json") where maybe_ = maybe (return ()) output dir =- catch_sh_maybe $ readfile $ dir </> pack "output.json" where+ catch_sh_maybe (readfile $ dir </> pack "output.json") where catch_sh_maybe action = catch_sh (liftM Just action) $ \(_ :: SomeException) -> return Nothing store dir = do@@ -99,31 +120,27 @@ , fromIntegral $ length blob , fromStrict blob )- docker dataDir storeDir Container{..} =+ docker dataDir storeDir Container{..} = do+ devices <- forM cDevices $ \device ->+ liftM strip $ run "readlink" ["-f", device] run_ "docker" $ concat [["run"]- , devices- , volumes+ , concatMap (("--device" :) . return) devices+ , concatMap (("--env" :) . return) cEnvironment+ , concatMap (("--volume" :) . return) $+ append (toTextIgnore dataDir) ":/app/data" :+ append (toTextIgnore storeDir) ":/app/store" : cVolumes , [cImage] , words cCommand- ] where- devices =- concatMap (("--device" :) . return) cDevices- volumes =- concatMap (("--volume" :) . return) $- append (toTextIgnore dataDir) ":/app/data" :- append (toTextIgnore storeDir) ":/app/store" : cVolumes+ ] main :: IO () main = execParser argsPI >>= call where call Args{..} = do- config <- decodeFile aConfig >>= hoistMaybe "Bad Config"- container <- decodeFile aContainer >>= hoistMaybe "Bad Container"+ config <- decodeFile aConfig >>= maybeThrow (userError "Bad Config")+ container <- decodeFile aContainer >>= maybeThrow (userError "Bad Container") env <- flowEnv config- forever $ do- r <- runFlowT env $+ forever $+ runResourceT $ runFlowT env $ act aQueue $ exec container- print r where- hoistMaybe s =- maybe (error s) return
src/main/Decide.hs view
@@ -1,13 +1,13 @@-{-# LANGUAGE RecordWildCards #-}--module Decide ( main ) where+module Decide+ ( main+ ) where -import Control.Monad ( forever )-import Data.Yaml ( decodeFile )-import Network.AWS.Flow ( runFlowT, decide )-import Network.AWS.Flow.Env ( flowEnv )+import Control.Monad+import Control.Monad.Trans.Resource+import Data.Yaml+import Network.AWS.Flow import Options.Applicative-import Prelude hiding ( readFile )+import Prelude hiding ( readFile ) data Args = Args { aConfig :: FilePath@@ -37,15 +37,12 @@ } main :: IO ()-main =+main = do execParser argsPI >>= call where call Args{..} = do- config <- decodeFile aConfig >>= hoistMaybe "Bad Config"- plan <- decodeFile aPlan >>= hoistMaybe "Bad Plan"+ config <- decodeFile aConfig >>= maybeThrow (userError "Bad Config")+ plan <- decodeFile aPlan >>= maybeThrow (userError "Bad Plan") env <- flowEnv config- forever $ do- r <- runFlowT env $+ forever $+ runResourceT $ runFlowT env $ decide plan- print r where- hoistMaybe s =- maybe (error s) return
src/main/Execute.hs view
@@ -1,18 +1,18 @@-{-# LANGUAGE RecordWildCards #-}--module Execute ( main ) where+module Execute+ ( main+ ) where -import Data.Text.IO ( readFile )-import Data.Yaml ( decodeFile )-import Network.AWS.Flow ( Plan (..), Start (..), runFlowT, execute )-import Network.AWS.Flow.Env ( flowEnv )+import Control.Monad.Trans.Resource+import Data.Text.IO+import Data.Yaml+import Network.AWS.Flow import Options.Applicative-import Prelude hiding ( readFile )+import Prelude hiding ( readFile ) data Args = Args { aConfig :: FilePath , aPlan :: FilePath- , aInput :: FilePath+ , aInput :: Maybe FilePath } deriving ( Eq, Read, Show ) argsPI :: ParserInfo Args@@ -32,11 +32,11 @@ <> short 'p' <> metavar "FILE" <> help "AWS SWF Service Flow plan" )- <*> strOption+ <*> optional ( strOption ( long "input" <> short 'i' <> metavar "FILE"- <> help "AWS SWF Service Flow input" ) where+ <> help "AWS SWF Service Flow input" ) ) where args config plan input = Args { aConfig = config , aPlan = plan@@ -47,12 +47,11 @@ main = execParser argsPI >>= call where call Args{..} = do- config <- decodeFile aConfig >>= hoistMaybe "Bad Config"- plan <- decodeFile aPlan >>= hoistMaybe "Bad Plan"- input <- readFile aInput+ config <- decodeFile aConfig >>= maybeThrow (userError "Bad Config")+ plan <- decodeFile aPlan >>= maybeThrow (userError "Bad Plan")+ input <- readFileMaybe aInput env <- flowEnv config- r <- runFlowT env $- execute (strtTask $ plnStart plan) (Just input)- print r where- hoistMaybe s =- maybe (error s) return+ runResourceT $ runFlowT env $+ execute (strtTask $ plnStart plan) input where+ readFileMaybe =+ maybe (return Nothing) ((>>= return . Just) . readFile)
src/main/Register.hs view
@@ -1,10 +1,10 @@-{-# LANGUAGE RecordWildCards #-}--module Register ( main ) where+module Register+ ( main+ ) where -import Data.Yaml ( decodeFile )-import Network.AWS.Flow ( runFlowT, register )-import Network.AWS.Flow.Env ( flowEnv )+import Control.Monad.Trans.Resource hiding ( register )+import Data.Yaml+import Network.AWS.Flow import Options.Applicative data Args = Args@@ -35,14 +35,11 @@ } main :: IO ()-main =+main = do execParser argsPI >>= call where call Args{..} = do- config <- decodeFile aConfig >>= hoistMaybe "Bad Config"- plan <- decodeFile aPlan >>= hoistMaybe "Bad Plan"+ config <- decodeFile aConfig >>= maybeThrow (userError "Bad Config")+ plan <- decodeFile aPlan >>= maybeThrow (userError "Bad Plan") env <- flowEnv config- r <- runFlowT env $+ runResourceT $ runFlowT env $ register plan- print r where- hoistMaybe s =- maybe (error s) return
wolf.cabal view
@@ -1,5 +1,5 @@ name: wolf-version: 0.1.1+version: 0.2.0 synopsis: Amazon Simple Workflow Service Wrapper. homepage: https://github.com/swift-nav/wolf license: MIT@@ -21,8 +21,7 @@ library exposed-modules: Network.AWS.Flow- , Network.AWS.Flow.Env- other-modules: Network.AWS.Flow.Internal+ other-modules: Network.AWS.Flow.Env , Network.AWS.Flow.S3 , Network.AWS.Flow.SWF , Network.AWS.Flow.Types@@ -31,13 +30,13 @@ ghc-options: -Wall -fno-warn-orphans build-depends: aeson , amazonka+ , amazonka-core , amazonka-s3 , amazonka-swf- , base >= 4.7 && < 4.8+ , base >= 4.7 && < 5 , bytestring , conduit , conduit-extra- , cryptohash , exceptions , fast-logger , http-conduit@@ -47,6 +46,7 @@ , mtl , mtl-compat , optparse-applicative+ , resourcet , safe , text , transformers@@ -54,51 +54,67 @@ , unordered-containers , uuid , yaml+ default-extensions: OverloadedStrings+ FlexibleContexts+ RecordWildCards executable wolf-register default-language: Haskell2010 main-is: Register.hs hs-source-dirs: src/main- ghc-options: -Wall -fno-warn-orphans -main-is Register- build-depends: base >= 4.7 && < 4.8- , wolf+ ghc-options: -Wall -main-is Register+ build-depends: base , optparse-applicative+ , resourcet , text+ , wolf , yaml+ default-extensions: OverloadedStrings+ RecordWildCards executable wolf-execute default-language: Haskell2010 main-is: Execute.hs hs-source-dirs: src/main- ghc-options: -Wall -fno-warn-orphans -main-is Execute- build-depends: base >= 4.7 && < 4.8- , wolf+ ghc-options: -Wall -main-is Execute+ build-depends: base , optparse-applicative+ , resourcet , text+ , wolf , yaml+ default-extensions: OverloadedStrings+ RecordWildCards executable wolf-decide default-language: Haskell2010 main-is: Decide.hs hs-source-dirs: src/main- ghc-options: -Wall -fno-warn-orphans -main-is Decide- build-depends: base >= 4.7 && < 4.8- , wolf+ ghc-options: -Wall -main-is Decide+ build-depends: base , optparse-applicative+ , resourcet , text+ , wolf , yaml+ default-extensions: OverloadedStrings+ RecordWildCards executable wolf-act default-language: Haskell2010 main-is: Act.hs hs-source-dirs: src/main- ghc-options: -Wall -fno-warn-orphans -main-is Act- build-depends: base >= 4.7 && < 4.8- , wolf+ ghc-options: -Wall -main-is Act+ build-depends: aeson+ , amazonka-core+ , base , bytestring- , cryptohash , optparse-applicative , shelly+ , resourcet , text , transformers+ , wolf , yaml+ default-extensions: OverloadedStrings+ RecordWildCards