yesod-job-queue 0.3.0.0 → 0.3.0.1
raw patch · 9 files changed
+230/−104 lines, 9 filesdep +monad-controldep +transformersdep +yesod-persistentbinary-added
Dependencies added: monad-control, transformers, yesod-persistent
Files
- README.md +103/−0
- Yesod/JobQueue.hs +84/−67
- Yesod/JobQueue/GenericConstr.hs +1/−7
- Yesod/JobQueue/Routes.hs +4/−1
- Yesod/JobQueue/Scheduler.hs +7/−7
- Yesod/JobQueue/Types.hs +4/−2
- app/dist/app.bundle.js too large to diff
- doc/yesod-job-queue-ss.png binary
- yesod-job-queue.cabal +27/−20
+ README.md view
@@ -0,0 +1,103 @@+# yesod-job-queue [](https://hackage.haskell.org/package/yesod-job-queue) [](http://stackage.org/lts/package/yesod-job-queue) [](http://stackage.org/nightly/package/yesod-job-queue) [](http://travis-ci.org/nakaji-dayo/yesod-job-queue)+Background jobs library for Yesod.++- There are API and Web UI for managing the job.+- Queue backend is Redis.+- Multithreaded.+++## Web interface++++## Usage++Prepare the JobState to manage the running jobs+``` haskell+import Yesod.JobQueue++data App = App {+ appConnPool :: ConnectionPool+ , appDBConf :: SqliteConf+ , appJobState :: JobState+ }++-- e.g. In makeFoundation+main = do+ jobState <- newJobState -- create JobState+ let app = App pool dbConf jobState+```++Make the routes for API & Manager UI+``` haskell+mkYesod "App" [parseRoutes|+/ HomeR GET+/job JobQueueR JobQueue getJobQueue -- ^ JobQueue API and Manager+|]+```++Define the job. And, to the setting of JobQueue+``` haskell+-- Your job type+data MyJobType = AggregationUser+ | PushNotification+ | HelloJob String+ deriving (Show, Read, Generic)++-- JobQueue settings+instance YesodJobQueue App where+ type JobType App = MyJobType+ getJobState = appJobState+ threadNumber _ = 2+ runDBJob action = do -- Required if you use the DB in the job+ app <- ask+ runSqlPool action $ appConnPool app+ -- Your Job code+ runJob _ AggregationUser = do+ us <- runDBJob $ selectList ([] :: [Filter Person]) []+ liftIO $ threadDelay $ 10 * 1000 * 1000+ print us+ putStrLn "complate job!"+ runJob _ PushNotification = do+ putStrLn "sent norification!"+ runJob _ (HelloJob name) = do+ putStrLn . pack $ "Hello " ++ name+```++Please see `example`++### Configration queue backend (Redis)+The default is to connect to `redis localhost:6379`.++And, you can change connect info by [queueConnectInfo](https://hackage.haskell.org/package/yesod-job-queue-0.2.0.1/docs/Yesod-JobQueue.html#v:queueConnectInfo)++## Build package+```+stack build+```+### build javascripts (required when app/* was editted)+```+cd app+npm install+npm run build-js+# need again `stack build`+```+#### developping app/*+You can edit app/app.js without re-build package+```+npm run watch-js ## watch app/*.js for changes, automatically build.+```+```+npm run bs ## serving app.bundle.js at `http://localhost:3001/dist/app.bundle.js` for test+```+in `example/Main.hs`. to use test version of app.bundle.js.+```+jobManagerJSUrl _ = "http://localhost:3001/dist/app.bundle.js"+```++## Run example+```+stack build --flag yesod-job-queue:example && stack exec yesod-job-queue-example+```+You can access Web UI: `http://localhost:3000/job/manager`+
Yesod/JobQueue.hs view
@@ -1,4 +1,9 @@-{-# LANGUAGE ImpredicativeTypes, UndecidableInstances #-}+{-# LANGUAGE ImpredicativeTypes #-}+{-# LANGUAGE QuasiQuotes #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE UndecidableInstances #-}+{-# OPTIONS_GHC -fno-warn-orphans #-}+ -- | -- -- Background jobs library for Yesod.@@ -15,32 +20,41 @@ , getJobQueue ) where -import qualified Prelude as P- import Yesod.JobQueue.Routes import Yesod.JobQueue.Types import Yesod.JobQueue.GenericConstr -import qualified Data.List as L-import ClassyPrelude.Yesod hiding (Proxy)-import Control.Concurrent+import Control.Concurrent (forkIO)+import qualified Control.Concurrent.STM as STM+import Control.Concurrent.STM (TVar) import Control.Lens ((^.))-import qualified Data.ByteString.Char8 as BSC (pack, unpack)-import qualified Data.Text as T (pack, append)-import Control.Monad.Logger-import qualified Database.Redis as R-+import Control.Monad (forever, void)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Monad.Trans.Class (lift)+import Control.Monad.Trans.Control (MonadBaseControl)+import Control.Monad.Trans.Reader (ReaderT, runReaderT)+import Data.Aeson (ToJSON(toJSON), Value(String), (.=), object)+import Data.Aeson.TH (defaultOptions, deriveToJSON)+import Data.ByteString (ByteString)+import qualified Data.ByteString.Char8 as BSC+import Data.FileEmbed (embedFile)+import Data.Foldable (forM_)+import qualified Data.List as L+import Data.Maybe (catMaybes)+import Data.Proxy (Proxy(Proxy))+import Data.Text (Text)+import qualified Data.Text as T+import Data.Time.Clock (UTCTime, getCurrentTime) import qualified Data.UUID as U import qualified Data.UUID.V4 as U-import qualified Control.Concurrent.STM as STM-import Data.Time.Clock--import Data.Aeson.TH-import GHC.Generics (Rep)-import Data.Proxy--import Data.FileEmbed (embedFile)-+import qualified Database.Redis as R+import GHC.Generics (Generic, Rep)+import Text.Read (readMaybe)+import Yesod.Core+ (HandlerT, Html, Yesod, YesodSubDispatch(yesodSubDispatch), getYesod,+ hamlet, invalidArgs, mkYesodSubDispatch, notFound, requireJsonBody,+ returnJson, sendResponse, toContent, withUrlRenderer)+import Yesod.Persist.Core (YesodPersistBackend) -- | Thread ID for convenience type ThreadNum = Int@@ -56,7 +70,7 @@ , startTime :: UTCTime } deriving (Eq) instance ToJSON U.UUID where- toJSON = String . pack . U.toString+ toJSON = String . T.pack . U.toString $(deriveToJSON defaultOptions ''RunningJob) -- | Manage the running jobs@@ -77,10 +91,10 @@ , Generic (JobType master), Constructors (Rep (JobType master)) ) => YesodJobQueue master where- + -- | Custom Job Type type JobType master- + -- | Job Handler runJob :: (MonadBaseControl IO m, MonadIO m) => master -> JobType master -> ReaderT master m ()@@ -88,7 +102,7 @@ -- | connection info for redis queueConnectInfo :: master -> R.ConnectInfo queueConnectInfo _ = R.defaultConnectInfo- + -- | queue key name for redis queueKey :: master -> ByteString queueKey _ = "yesod-job-queue"@@ -96,7 +110,7 @@ -- | The number of threads to run the job threadNumber :: master -> Int threadNumber _ = 1- + -- | runDB for job runDBJob :: (MonadBaseControl IO m, MonadIO m) => ReaderT (YesodPersistBackend master) (ReaderT master m) a@@ -127,46 +141,51 @@ forM_ [1 .. num] $ startThread m -- | start dequeue-ing job in new thread-startThread :: (YesodJobQueue master, MonadBaseControl IO m, MonadIO m) => master -> ThreadNum -> m ()-startThread m tNo = do- liftIO $ forkIO $ do- conn <- R.connect $ queueConnectInfo m- R.runRedis conn $ do- forever $ do- emr <- R.blpop [queueKey m] 600- liftIO $ case emr of- Left e -> putStrLn "[dequeue] error in at connection redis"- Right Nothing -> putStrLn "[dequeue] timeout retry"- Right (Just (k, v)) -> do- let item = readMay $ BSC.unpack v :: Maybe JobQueueItem- case readJobType m =<< queueJobType <$> item of- Just jt -> do- jid <- U.nextRandom- time <- getCurrentTime- let job = RunningJob (show jt) tNo jid time- STM.atomically- $ STM.modifyTVar (getJobState m)- (job:)- putStrLn . pack $ "dequeued: " ++ (show jt)- runReaderT (runJob m jt) m- STM.atomically- $ STM.modifyTVar (getJobState m)- (L.delete job)- Nothing -> putStrLn "[dequeue] unknown JobType"- return ()- return ()+startThread :: forall master m . (YesodJobQueue master, MonadBaseControl IO m, MonadIO m)+ => master -> ThreadNum -> m ()+startThread m tNo = void $ liftIO $ forkIO $ do+ conn <- R.connect $ queueConnectInfo m+ R.runRedis conn $ forever $ do+ result <- R.blpop [queueKey m] 600+ liftIO $ handleResultFromRedis result+ where+ handleResultFromRedis :: Either R.Reply (Maybe (ByteString, ByteString)) -> IO ()+ handleResultFromRedis (Left _) =+ putStrLn "[dequeue] error in at connection redis"+ handleResultFromRedis (Right Nothing) =+ putStrLn "[dequeue] timeout retry"+ handleResultFromRedis (Right (Just (_, redisValue))) = do+ let item = readMaybe $ BSC.unpack redisValue+ handleJob $ readJobType m =<< queueJobType <$> item + handleJob :: Maybe (JobType master) -> IO ()+ handleJob Nothing = putStrLn "[dequeue] unknown JobType"+ handleJob (Just jt) = do+ jid <- U.nextRandom+ time <- getCurrentTime+ let runningJob = RunningJob+ { jobType = (show jt)+ , threadId = tNo+ , jobId = jid+ , startTime = time+ }+ STM.atomically $ STM.modifyTVar (getJobState m) (runningJob:)+ putStrLn $ "dequeued: " ++ (show jt)+ runReaderT (runJob m jt) m+ STM.atomically $ STM.modifyTVar (getJobState m) (L.delete runningJob)+ -- | Add job to end of the queue-enqueue :: YesodJobQueue master => master -> JobType master -> IO ()-enqueue m jt = do+enqueue :: (MonadIO m, YesodJobQueue master) => master -> JobType master -> m ()+enqueue m jt = liftIO $ do time <- getCurrentTime- let item = JobQueueItem (show jt) time- conn <- liftIO $ R.connect $ queueConnectInfo m- R.runRedis conn $ do- R.rpush (queueKey m) [BSC.pack $ show item]- return ()+ let item = JobQueueItem+ { queueJobType = show jt+ , queueTime = time+ }+ conn <- R.connect $ queueConnectInfo m+ void $ R.runRedis conn $ R.rpush (queueKey m) [BSC.pack $ show item] --- | Get all job in the queue+-- | Get all jobs in the queue listQueue :: YesodJobQueue master => master -> IO (Either String [JobQueueItem]) listQueue m = do conn <- R.connect $ queueConnectInfo m@@ -175,12 +194,12 @@ case exs of Right xs -> return $ Right $ catMaybes- $ map (readMay . BSC.unpack) xs+ $ map (readMaybe . BSC.unpack) xs Left r -> return $ Left $ show r -- | read JobType from String readJobType :: YesodJobQueue master => master -> String -> Maybe (JobType master)-readJobType _ = readMay+readJobType _ = readMaybe -- | Need by 'getClassInformation' jobQueueInfo :: YesodJobQueue master => master -> JobQueueClassInfo@@ -216,7 +235,7 @@ y <- getYesod Right q <- liftIO $ listQueue y returnJson $ object ["queue" .= q]- + -- | enqueue new job postJobQueueR :: JobHandler master Value postJobQueueR = lift $ do@@ -233,7 +252,7 @@ getJobStateR = lift $ do y <- getYesod s <- liftIO $ STM.readTVarIO (getJobState y)- returnJson $ object ["running" .= s] + returnJson $ object ["running" .= s] getJobManagerR :: JobHandler master Html getJobManagerR = lift $ do@@ -277,5 +296,3 @@ getJobQueue :: a -> JobQueue getJobQueue = const JobQueue--
Yesod/JobQueue/GenericConstr.hs view
@@ -1,13 +1,8 @@-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE KindSignatures #-}-{-# LANGUAGE TypeOperators #-} module Yesod.JobQueue.GenericConstr where -import GHC.Generics import Data.Proxy import Data.Typeable+import GHC.Generics -- *Demo> genericConstructors (Proxy :: Proxy (Either Int Bool)) -- [["Left","Int"],["Right","Bool"]]@@ -49,4 +44,3 @@ instance Typeable a => Fields (K1 i a) where fields _ = [show (typeRep (Proxy :: Proxy a))]-
Yesod/JobQueue/Routes.hs view
@@ -1,7 +1,10 @@+{-# LANGUAGE QuasiQuotes #-}+{-# LANGUAGE TemplateHaskell #-}+ module Yesod.JobQueue.Routes where -import Yesod import Data.Text+import Yesod -- Subsites have foundations just like master sites. data JobQueue = JobQueue
Yesod/JobQueue/Scheduler.hs view
@@ -1,15 +1,15 @@ -- | Cron Job for Yesod module Yesod.JobQueue.Scheduler where +import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Monad.Trans.Control (MonadBaseControl)+import Data.Monoid ((<>))+import qualified Data.Text as T (pack)+import System.Cron.Schedule import Yesod.JobQueue import Yesod.JobQueue.Types-import System.Cron.Schedule -import qualified Prelude as P-import ClassyPrelude.Yesod -import qualified Data.Text as T (pack)- -- | Cron Scheduler for YesodJobQueue class (YesodJobQueue master) => YesodJobQueueScheduler master where -- | job schedules@@ -20,9 +20,9 @@ startJobSchedule master = do let add (s, jt) = addJob (enqueue master jt) s tids <- liftIO $ execSchedule $ mapM_ add $ getJobSchedules master- print tids+ liftIO $ print tids -- | Need by 'getClassInformation' schedulerInfo :: YesodJobQueueScheduler master => master -> JobQueueClassInfo schedulerInfo m = JobQueueClassInfo "Scheduler" $ map (T.pack . showSchedule) $ getJobSchedules m- where showSchedule (s, jt) = s ++ " | " ++ (show jt)+ where showSchedule (s, jt) = s <> " | " <> (show jt)
Yesod/JobQueue/Types.hs view
@@ -1,8 +1,10 @@+{-# LANGUAGE TemplateHaskell #-}+ module Yesod.JobQueue.Types where -import ClassyPrelude.Yesod import Data.Aeson.APIFieldJsonTH-import Control.Lens+import Data.Text (Text)+import Control.Lens (makeFields) data PostJobQueueRequest = PostJobQueueRequest { _postJobQueueRequestJob :: String
app/dist/app.bundle.js view
file too large to diff
+ doc/yesod-job-queue-ss.png view
binary file changed (absent → 245266 bytes)
yesod-job-queue.cabal view
@@ -1,5 +1,5 @@ name: yesod-job-queue-version: 0.3.0.0+version: 0.3.0.1 synopsis: Background jobs library for Yesod. description: Background jobs library for Yesod@@ -18,7 +18,9 @@ copyright: 2016 Daishi Nakajima category: Web build-type: Custom-extra-source-files: app/dist/app.bundle.js+extra-source-files: README.md+ app/dist/app.bundle.js+ doc/yesod-job-queue-ss.png cabal-version: >=1.10 flag example@@ -27,36 +29,41 @@ library exposed-modules: Yesod.JobQueue+ , Yesod.JobQueue.GenericConstr+ , Yesod.JobQueue.Routes , Yesod.JobQueue.Scheduler , Yesod.JobQueue.Types- , Yesod.JobQueue.Routes- , Yesod.JobQueue.GenericConstr- default-extensions: TemplateHaskell- , MultiParamTypeClasses- , FunctionalDependencies- , TypeSynonymInstances+ default-extensions: FlexibleContexts , FlexibleInstances- , QuasiQuotes- , TypeFamilies+ , FunctionalDependencies+ , KindSignatures+ , MultiParamTypeClasses , OverloadedStrings- , FlexibleContexts+ , ScopedTypeVariables+ , TypeFamilies+ , TypeOperators+ , TypeSynonymInstances , ViewPatterns build-depends: base >= 4.7 && < 5- , stm- , hedis- , uuid , aeson- , cron- , monad-logger- , bytestring- , lens- , classy-prelude-yesod , api-field-json-th- , yesod >= 1.4 && < 1.5+ , bytestring+ , cron , file-embed+ , hedis+ , lens+ , monad-control+ , monad-logger+ , stm , text , time+ , transformers+ , uuid+ , yesod >= 1.4 && < 1.5+ , yesod-core+ , yesod-persistent + ghc-options: -Wall -fwarn-tabs -O2 default-language: Haskell2010 executable yesod-job-queue-example