wolf (empty) → 0.1.0
raw patch · 14 files changed
+1227/−0 lines, 14 filesdep +aesondep +amazonkadep +amazonka-s3setup-changed
Dependencies added: aeson, amazonka, amazonka-s3, amazonka-swf, base, bytestring, conduit, conduit-extra, cryptohash, exceptions, fast-logger, http-conduit, lens, monad-control, monad-logger, mtl, optparse-applicative, safe, shelly, text, transformers, transformers-base, unordered-containers, uuid, wolf, yaml
Files
- LICENSE +20/−0
- README.md +3/−0
- Setup.hs +2/−0
- src/Network/AWS/Flow.hs +229/−0
- src/Network/AWS/Flow/Env.hs +75/−0
- src/Network/AWS/Flow/Internal.hs +207/−0
- src/Network/AWS/Flow/S3.hs +22/−0
- src/Network/AWS/Flow/SWF.hs +147/−0
- src/Network/AWS/Flow/Types.hs +133/−0
- src/main/Act.hs +129/−0
- src/main/Decide.hs +51/−0
- src/main/Execute.hs +58/−0
- src/main/Register.hs +48/−0
- wolf.cabal +103/−0
+ LICENSE view
@@ -0,0 +1,20 @@+Copyright (C) 2015 Swift Navigation Inc.++Permission is hereby granted, free of charge, to any person obtaining+a copy of this software and associated documentation files (the+"Software"), to deal in the Software without restriction, including+without limitation the rights to use, copy, modify, merge, publish,+distribute, sublicense, and/or sell copies of the Software, and to+permit persons to whom the Software is furnished to do so, subject to+the following conditions:++The above copyright notice and this permission notice shall be included+in all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.+IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY+CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,+TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE+SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ README.md view
@@ -0,0 +1,3 @@+# Wolf++Wolf is a wrapper around Amazon Simple Workflow Service.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ src/Network/AWS/Flow.hs view
@@ -0,0 +1,229 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE RecordWildCards #-}++module Network.AWS.Flow+ ( register+ , execute+ , act+ , decide+ , runFlowT+ , throwStringError+ , hoistStringEither+ , maybeToFlowError+ , Uid+ , Name+ , Version+ , Queue+ , Token+ , Timeout+ , Metadata+ , Artifact+ , FlowConfig (..)+ , FlowEnv (..)+ , FlowError+ , FlowT+ , MonadFlow+ , Task (..)+ , Timer (..)+ , Start (..)+ , Spec (..)+ , End (..)+ , 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 )++-- Interface++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+ go rs Work{..} = do+ r <- registerActivityTypeAction+ (tskName wrkTask)+ (tskVersion wrkTask)+ (tskTimeout wrkTask)+ return (r : rs)+ go rs Sleep{..} = return rs++execute :: MonadFlow m => Task -> Metadata -> m ()+execute Task{..} input = do+ uid <- newUid+ startWorkflowExecutionAction uid tskName tskVersion tskQueue input++act :: MonadFlow m => Queue -> (Uid -> Metadata -> m (Metadata, [Artifact])) -> m ()+act queue action = do+ (token, uid, input) <- pollForActivityTaskAction queue+ (output, artifacts) <- action uid input+ forM_ artifacts putObjectAction+ respondActivityTaskCompletedAction token output++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++-- Helpers++nextEvent :: MonadDecide m => [EventType] -> m HistoryEvent+nextEvent ets = do+ events <- asks deEvents+ maybeToFlowError "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+ , TimerFired+ , StartChildWorkflowExecutionInitiated ]+ case event ^. heEventType of+ WorkflowExecutionStarted -> start event+ ActivityTaskCompleted -> completed event+ TimerFired -> timer event+ StartChildWorkflowExecutionInitiated -> child+ _ -> throwStringError "Unknown Select Event"++start :: MonadDecide m => HistoryEvent -> m [Decision]+start event = do+ input <- maybeToFlowError "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+ findEvent <- asks deFindEvent+ (input, name) <- maybeToFlowError "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++timer :: MonadDecide m => HistoryEvent -> m [Decision]+timer event = do+ findEvent <- asks deFindEvent+ name <- maybeToFlowError "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+ _ -> throwStringError "Unknown Timer Event"++timerStart :: MonadDecide m => HistoryEvent -> Name -> m [Decision]+timerStart event name = do+ input <- maybeToFlowError "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+ input <- maybeToFlowError "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+ 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+ end <- asks (plnEnd . dePlan)+ case end of+ Stop -> return [completeWorkflowExecutionDecision input]+ Continue -> scheduleContinue++scheduleContinue :: MonadDecide m => m [Decision]+scheduleContinue = do+ event <- nextEvent [WorkflowExecutionStarted]+ input <- maybeToFlowError "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+ _ -> throwStringError "Unknown Child Event"++childStart :: MonadDecide m => HistoryEvent -> m [Decision]+childStart event = do+ input <- maybeToFlowError "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+ attrs <- event ^. heActivityTaskCompletedEventAttributes+ return $ attrs ^. atceaResult+ return [completeWorkflowExecutionDecision input]+
+ src/Network/AWS/Flow/Env.hs view
@@ -0,0 +1,75 @@+{-# 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.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++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+ logStrLn ls s =+ pushLogStr ls s >> flushLogStr ls
+ src/Network/AWS/Flow/Internal.hs view
@@ -0,0 +1,207 @@+{-# 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
@@ -0,0 +1,22 @@+{-# 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.Flow.Types+import Network.AWS.Flow.Internal ( runAWS )+import Network.AWS.S3 hiding ( bucket )++-- 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
+ src/Network/AWS/Flow/SWF.hs view
@@ -0,0 +1,147 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE ConstraintKinds #-}++module Network.AWS.Flow.SWF+ ( registerDomainAction+ , registerActivityTypeAction+ , registerWorkflowTypeAction+ , startWorkflowExecutionAction+ , pollForActivityTaskAction+ , respondActivityTaskCompletedAction+ , respondActivityTaskFailedAction+ , pollForDecisionTaskAction+ , respondDecisionTaskCompletedAction+ , scheduleActivityTaskDecision+ , completeWorkflowExecutionDecision+ , startTimerDecision+ , continueAsNewWorkflowExecutionDecision+ , 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 Network.AWS.Flow.Types+import Network.AWS.Flow.Internal ( runAWS )+import Network.AWS.SWF+import Safe ( headMay )++-- Actions++registerDomainAction :: MonadFlow m => m ()+registerDomainAction = do+ domain <- asks feDomain+ runAWS feEnv $+ send_ $ registerDomain domain "30"++registerActivityTypeAction :: MonadFlow m => Name -> Version -> Timeout -> m ()+registerActivityTypeAction name version timeout = do+ domain <- asks feDomain+ runAWS feEnv $+ send_ $ registerActivityType domain name version &+ ratDefaultTaskHeartbeatTimeout .~ Just "NONE" &+ ratDefaultTaskScheduleToCloseTimeout .~ Just "NONE" &+ ratDefaultTaskScheduleToStartTimeout .~ Just "60" &+ ratDefaultTaskStartToCloseTimeout .~ Just timeout++registerWorkflowTypeAction :: MonadFlow m => Name -> Version -> Timeout -> m ()+registerWorkflowTypeAction name version timeout = do+ domain <- asks feDomain+ runAWS feEnv $+ send_ $ registerWorkflowType domain name version &+ rwtDefaultChildPolicy .~ Just Abandon &+ rwtDefaultExecutionStartToCloseTimeout .~ Just timeout &+ rwtDefaultTaskStartToCloseTimeout .~ Just "60"++startWorkflowExecutionAction :: MonadFlow m+ => Uid -> Name -> Version -> Queue -> Metadata -> m ()+startWorkflowExecutionAction uid name version queue input = do+ domain <- asks feDomain+ runAWS feEnv $+ send_ $ startWorkflowExecution domain uid (workflowType name version) &+ swe1TaskList .~ Just (taskList queue) &+ swe1Input .~ input++pollForActivityTaskAction :: MonadFlow m => Queue -> m (Token, Uid, Metadata)+pollForActivityTaskAction queue = do+ domain <- asks feDomain+ runAWS fePollEnv $ do+ r <- send $ pollForActivityTask domain (taskList queue)+ return+ ( r ^. pfatrTaskToken+ , r ^. pfatrWorkflowExecution ^. weWorkflowId+ , r ^. pfatrInput )++respondActivityTaskCompletedAction :: MonadFlow m => Token -> Metadata -> m ()+respondActivityTaskCompletedAction token result =+ runAWS feEnv $+ send_ $ respondActivityTaskCompleted token &+ ratcResult .~ result++respondActivityTaskFailedAction :: MonadFlow m => Token -> m ()+respondActivityTaskFailedAction token =+ runAWS feEnv $+ send_ $ respondActivityTaskFailed token++pollForDecisionTaskAction :: MonadFlow m => Queue -> m (Maybe Token, [HistoryEvent])+pollForDecisionTaskAction queue = do+ domain <- asks feDomain+ runAWS fePollEnv $ do+ rs <- paginate (pollForDecisionTask domain (taskList queue) &+ pfdtReverseOrder .~ Just True &+ pfdtMaximumPageSize .~ Just 100)+ $$ consume+ return+ ( liftM (^. pfdtrTaskToken) (headMay rs)+ , concatMap (^. pfdtrEvents) rs)++respondDecisionTaskCompletedAction :: MonadFlow m => Token -> [Decision] -> m ()+respondDecisionTaskCompletedAction token decisions =+ runAWS feEnv $+ 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++startTimerDecision :: Uid -> Name -> Timeout -> Decision+startTimerDecision uid name timeout =+ decision StartTimer &+ dStartTimerDecisionAttributes .~ Just attrs where+ attrs = startTimerDecisionAttributes uid timeout &+ 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 view
@@ -0,0 +1,133 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE ConstraintKinds #-}++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 )++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, Int64, ByteString)++data FlowConfig = FlowConfig+ { fcRegion :: Region+ , fcCredentials :: Credentials+ , fcTimeout :: Int+ , fcPollTimeout :: Int+ , fcDomain :: Text+ , fcBucket :: Text+ }++data FlowEnv = FlowEnv+ { feLogger :: LogStr -> IO ()+ , feEnv :: Env+ , fePollEnv :: Env+ , feDomain :: Text+ , feBucket :: Text+ }++data FlowError+ = FlowError String+ | AWSError Error+ deriving ( Show )++newtype FlowT m a = FlowT+ { unFlowT :: LoggingT (ReaderT FlowEnv (ExceptT FlowError m)) a+ } deriving ( Functor+ , Applicative+ , Monad+ , MonadIO+ , MonadLogger+ , MonadCatch+ , MonadThrow+ , MonadError FlowError+ )++type MonadFlow m =+ ( MonadBaseControl IO m+ , MonadCatch m+ , MonadIO m+ , MonadLogger m+ , MonadReader FlowEnv m+ , MonadError FlowError m+ )++data DecideEnv = DecideEnv+ { deLogger :: LogStr -> IO ()+ , dePlan :: Plan+ , deEvents :: [HistoryEvent]+ , deFindEvent :: Integer -> Maybe HistoryEvent+ }++newtype DecideT m a = DecideT+ { unDecideT :: LoggingT (ReaderT DecideEnv (ExceptT FlowError m)) a+ } deriving ( Functor+ , Applicative+ , Monad+ , MonadIO+ , MonadLogger+ , MonadCatch+ , MonadThrow+ , MonadError FlowError+ )++type MonadDecide m =+ ( MonadBaseControl IO m+ , MonadCatch m+ , MonadIO m+ , MonadLogger m+ , MonadReader DecideEnv m+ , MonadError FlowError m+ )++data Task = Task+ { tskName :: Name+ , tskVersion :: Version+ , tskQueue :: Queue+ , tskTimeout :: Timeout+ } deriving ( Eq, Read, Show )++data Timer = Timer+ { tmrName :: Name+ , tmrTimeout :: Timeout+ } deriving ( Eq, Read, Show )++data Start = Start+ { strtTask :: Task+ } deriving ( Eq, Read, Show )++data Spec+ = Work+ { wrkTask :: Task+ }+ | Sleep+ { slpTimer :: Timer+ } deriving ( Eq, Read, Show )++data End+ = Stop+ | Continue+ deriving ( Eq, Read, Show )++data Plan = Plan+ { plnStart :: Start+ , plnSpecs :: [Spec]+ , plnEnd :: End+ } deriving ( Eq, Read, Show )
+ src/main/Act.hs view
@@ -0,0 +1,129 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}++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 Data.Yaml+import Network.AWS.Flow ( Artifact, Metadata, Queue, Uid, runFlowT, act )+import Network.AWS.Flow.Env ( flowEnv )+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++data Args = Args+ { aConfig :: FilePath+ , aQueue :: Queue+ , aContainer :: FilePath+ } deriving ( Eq, Read, Show )++argsPI :: ParserInfo Args+argsPI =+ info ( helper <*> argsP )+ ( fullDesc+ <> header "act: Workflow activity"+ <> progDesc "Workflow activity" ) where+ argsP = args+ <$> strOption+ ( long "config"+ <> short 'c'+ <> metavar "FILE"+ <> help "AWS SWF Service Flow config" )+ <*> strOption+ ( long "queue"+ <> short 'q'+ <> metavar "NAME"+ <> help "AWS SWF Service Flow queue" )+ <*> strOption+ ( long "container"+ <> short 'x'+ <> metavar "FILE"+ <> help "AWS SWF Service Flow worker container" ) where+ args config queue container = Args+ { aConfig = config+ , aQueue = pack queue+ , aContainer = container+ }++exec :: MonadIO m => Container -> Uid -> Metadata -> m (Metadata, [Artifact])+exec container uid metadata =+ shelly $ withDir $ \dataDir storeDir -> do+ input dataDir metadata+ docker dataDir storeDir container+ result <- output dataDir+ artifacts <- store storeDir+ return (result, artifacts) where+ withDir action =+ withTmpDir $ \dir -> do+ mkdir $ dir </> pack "data"+ mkdir $ dir </> pack "store"+ action (dir </> pack "data") (dir </> pack "store")+ input dir =+ maybe_ $ writefile $ dir </> pack "input.json" where+ maybe_ =+ maybe (return ())+ output dir =+ catch_sh_maybe $ readfile $ dir </> pack "output.json" where+ catch_sh_maybe action =+ catch_sh (liftM Just action) $ \(_ :: SomeException) -> return Nothing+ store dir = do+ artifacts <- findWhen test_f dir+ forM artifacts $ \artifact -> do+ key <- relativeTo dir artifact+ blob <- readBinary artifact+ return ( toTextIgnore $ uid </> key+ , hash blob+ , fromIntegral $ length blob+ , fromStrict blob+ )+ docker dataDir storeDir Container{..} =+ run_ "docker" $ concat+ [["run"]+ , devices+ , volumes+ , [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"+ env <- flowEnv config+ forever $ do+ r <- runFlowT env $+ act aQueue $ exec container+ print r where+ hoistMaybe s =+ maybe (error s) return
+ src/main/Decide.hs view
@@ -0,0 +1,51 @@+{-# LANGUAGE RecordWildCards #-}++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 Options.Applicative+import Prelude hiding ( readFile )++data Args = Args+ { aConfig :: FilePath+ , aPlan :: FilePath+ } deriving ( Eq, Read, Show )++argsPI :: ParserInfo Args+argsPI =+ info ( helper <*> argsP )+ ( fullDesc+ <> header "decide: Decide a workflow"+ <> progDesc "Decide a workflow" ) where+ argsP = args+ <$> strOption+ ( long "config"+ <> short 'c'+ <> metavar "FILE"+ <> help "AWS SWF Service Flow config" )+ <*> strOption+ ( long "plan"+ <> short 'p'+ <> metavar "FILE"+ <> help "AWS SWF Service Flow plan" ) where+ args config plan = Args+ { aConfig = config+ , aPlan = plan+ }++main :: IO ()+main =+ execParser argsPI >>= call where+ call Args{..} = do+ config <- decodeFile aConfig >>= hoistMaybe "Bad Config"+ plan <- decodeFile aPlan >>= hoistMaybe "Bad Plan"+ env <- flowEnv config+ forever $ do+ r <- runFlowT env $+ decide plan+ print r where+ hoistMaybe s =+ maybe (error s) return
+ src/main/Execute.hs view
@@ -0,0 +1,58 @@+{-# LANGUAGE RecordWildCards #-}++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 Options.Applicative+import Prelude hiding ( readFile )++data Args = Args+ { aConfig :: FilePath+ , aPlan :: FilePath+ , aInput :: FilePath+ } deriving ( Eq, Read, Show )++argsPI :: ParserInfo Args+argsPI =+ info ( helper <*> argsP )+ ( fullDesc+ <> header "execute: Execute a workflow"+ <> progDesc "Execute a workflow" ) where+ argsP = args+ <$> strOption+ ( long "config"+ <> short 'c'+ <> metavar "FILE"+ <> help "AWS SWF Service Flow config" )+ <*> strOption+ ( long "plan"+ <> short 'p'+ <> metavar "FILE"+ <> help "AWS SWF Service Flow plan" )+ <*> strOption+ ( long "input"+ <> short 'i'+ <> metavar "FILE"+ <> help "AWS SWF Service Flow input" ) where+ args config plan input = Args+ { aConfig = config+ , aPlan = plan+ , aInput = input+ }++main :: IO ()+main =+ execParser argsPI >>= call where+ call Args{..} = do+ config <- decodeFile aConfig >>= hoistMaybe "Bad Config"+ plan <- decodeFile aPlan >>= hoistMaybe "Bad Plan"+ input <- readFile aInput+ env <- flowEnv config+ r <- runFlowT env $+ execute (strtTask $ plnStart plan) (Just input)+ print r where+ hoistMaybe s =+ maybe (error s) return
+ src/main/Register.hs view
@@ -0,0 +1,48 @@+{-# LANGUAGE RecordWildCards #-}++module Register ( main ) where++import Data.Yaml ( decodeFile )+import Network.AWS.Flow ( runFlowT, register )+import Network.AWS.Flow.Env ( flowEnv )+import Options.Applicative++data Args = Args+ { aConfig :: FilePath+ , aPlan :: FilePath+ } deriving ( Eq, Read, Show )++argsPI :: ParserInfo Args+argsPI =+ info ( helper <*> argsP )+ ( fullDesc+ <> header "register: Register a workflow"+ <> progDesc "Register a workflow" ) where+ argsP = args+ <$> strOption+ ( long "config"+ <> short 'c'+ <> metavar "FILE"+ <> help "AWS SWF Service Flow config" )+ <*> strOption+ ( long "plan"+ <> short 'p'+ <> metavar "FILE"+ <> help "AWS SWF Service Flow plan" ) where+ args config plan = Args+ { aConfig = config+ , aPlan = plan+ }++main :: IO ()+main =+ execParser argsPI >>= call where+ call Args{..} = do+ config <- decodeFile aConfig >>= hoistMaybe "Bad Config"+ plan <- decodeFile aPlan >>= hoistMaybe "Bad Plan"+ env <- flowEnv config+ r <- runFlowT env $+ register plan+ print r where+ hoistMaybe s =+ maybe (error s) return
+ wolf.cabal view
@@ -0,0 +1,103 @@+name: wolf+version: 0.1.0+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.++source-repository head+ type: git+ location: git@github.com:swift-nav/wolf.git++library+ exposed-modules: Network.AWS.Flow+ , Network.AWS.Flow.Env+ other-modules: Network.AWS.Flow.Internal+ , Network.AWS.Flow.S3+ , Network.AWS.Flow.SWF+ , Network.AWS.Flow.Types+ default-language: Haskell2010+ hs-source-dirs: src+ ghc-options: -Wall -fno-warn-orphans+ build-depends: aeson >= 0.8.0+ , amazonka >= 0.3.4+ , amazonka-s3 >= 0.3.4+ , amazonka-swf >= 0.3.4+ , base >= 4.7 && < 4.8+ , bytestring >= 0.10.4+ , conduit >= 1.2.0+ , conduit-extra >= 1.1.7+ , cryptohash >= 0.11.6+ , exceptions >= 0.8.0+ , fast-logger >= 2.3.1+ , http-conduit >= 2.1.0+ , lens >= 4.9.0+ , monad-control >= 1.0.0+ , monad-logger >= 0.3.0+ , mtl >= 2.2.0+ , optparse-applicative >= 0.11.0+ , safe >= 0.3.0+ , text >= 1.2.0+ , transformers >= 0.4.0+ , transformers-base >= 0.4.0+ , unordered-containers >= 0.2.5+ , uuid >= 1.3.10+ , yaml >= 0.8.11++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+ , optparse-applicative >= 0.11.0+ , text >= 1.2.0+ , yaml >= 0.8.0++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+ , optparse-applicative >= 0.11.0+ , text >= 1.2.0+ , yaml >= 0.8.0++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+ , optparse-applicative >= 0.11.0+ , text >= 1.2.0+ , yaml >= 0.8.0++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+ , bytestring >= 0.10.4+ , cryptohash >= 0.11.6+ , optparse-applicative >= 0.11.0+ , shelly >= 1.6.1+ , text >= 1.2.0+ , transformers >= 0.4.0+ , yaml >= 0.8.0