diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,103 @@ +# yesod-job-queue [![Hackage](https://img.shields.io/hackage/v/yesod-job-queue.svg?maxAge=25920)](https://hackage.haskell.org/package/yesod-job-queue) [![Stackage LTS](http://stackage.org/package/yesod-job-queue/badge/lts)](http://stackage.org/lts/package/yesod-job-queue) [![Stackage Nightly](http://stackage.org/package/yesod-job-queue/badge/nightly)](http://stackage.org/nightly/package/yesod-job-queue) [![Build Status](https://secure.travis-ci.org/nakaji-dayo/yesod-job-queue.svg)](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 + +![web interface screenshot](https://raw.githubusercontent.com/nakaji-dayo/yesod-job-queue/master/doc/yesod-job-queue-ss.png) + +## 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` + diff --git a/Yesod/JobQueue.hs b/Yesod/JobQueue.hs --- a/Yesod/JobQueue.hs +++ b/Yesod/JobQueue.hs @@ -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 - - diff --git a/Yesod/JobQueue/GenericConstr.hs b/Yesod/JobQueue/GenericConstr.hs --- a/Yesod/JobQueue/GenericConstr.hs +++ b/Yesod/JobQueue/GenericConstr.hs @@ -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))] - diff --git a/Yesod/JobQueue/Routes.hs b/Yesod/JobQueue/Routes.hs --- a/Yesod/JobQueue/Routes.hs +++ b/Yesod/JobQueue/Routes.hs @@ -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 diff --git a/Yesod/JobQueue/Scheduler.hs b/Yesod/JobQueue/Scheduler.hs --- a/Yesod/JobQueue/Scheduler.hs +++ b/Yesod/JobQueue/Scheduler.hs @@ -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) diff --git a/Yesod/JobQueue/Types.hs b/Yesod/JobQueue/Types.hs --- a/Yesod/JobQueue/Types.hs +++ b/Yesod/JobQueue/Types.hs @@ -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 diff --git a/app/dist/app.bundle.js b/app/dist/app.bundle.js # file too large to diff: app/dist/app.bundle.js diff --git a/doc/yesod-job-queue-ss.png b/doc/yesod-job-queue-ss.png new file mode 100644 Binary files /dev/null and b/doc/yesod-job-queue-ss.png differ diff --git a/yesod-job-queue.cabal b/yesod-job-queue.cabal --- a/yesod-job-queue.cabal +++ b/yesod-job-queue.cabal @@ -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