periodic-client-exe (empty) → 1.1.7.1
raw patch · 6 files changed
+910/−0 lines, 6 filesdep +basedep +binarydep +boxes
Dependencies added: base, binary, boxes, bytestring, data-default-class, deepseq, http-types, metro, metro-socket, metro-transport-tls, metro-transport-websockets, metro-transport-xor, periodic-client, periodic-common, process, scotty, streaming-commons, text, unix-time, unliftio, warp, websockets
Files
- LICENSE +30/−0
- README.md +3/−0
- app/periodic-http-bridge.hs +202/−0
- app/periodic-run.hs +224/−0
- app/periodic.hs +375/−0
- periodic-client-exe.cabal +76/−0
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Li Meng Jun (c) 2017++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Li Meng Jun nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ README.md view
@@ -0,0 +1,3 @@+# periodic-client-exe++Periodic task system haskell client executables.
+ app/periodic-http-bridge.hs view
@@ -0,0 +1,202 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}++module Main+ ( main+ ) where++import Control.Monad (when)+import Control.Monad.IO.Class (liftIO)+import qualified Data.ByteString.Char8 as B (pack)+import qualified Data.ByteString.Lazy as LB (empty, fromStrict,+ toStrict)+import Data.Default.Class (def)+import Data.List (isPrefixOf)+import Data.Maybe (fromMaybe)+import Data.Streaming.Network.Internal (HostPreference (Host))+import Data.String (fromString)+import Metro.Class (Transport)+import Metro.Socket (getHost, getService)+import Metro.TP.Socket (socket)+import Metro.TP.TLS (makeClientParams', tlsConfig)+import Metro.TP.WebSockets (clientConfig)+import Metro.TP.XOR (xorConfig)+import Network.HTTP.Types (status204, status500)+import Network.Wai.Handler.Warp (setHost, setPort)+import Periodic.Trans.ClientPool+import Periodic.Types.Job+import System.Environment (getArgs, lookupEnv)+import System.Exit (exitSuccess)+import Web.Scotty (ActionM, ScottyM, body, get,+ param, post, raw, rescue,+ scottyOpts, settings)+import qualified Web.Scotty as WS (status)+++data Options = Options+ { host :: String+ , xorFile :: FilePath+ , useTls :: Bool+ , useWs :: Bool+ , hostName :: String+ , certKey :: FilePath+ , cert :: FilePath+ , caStore :: FilePath+ , httpHost :: String+ , httpPort :: Int+ , poolSize :: Int+ , showHelp :: Bool+ }++options :: Maybe String -> Maybe String -> Options+options h f = Options+ { host = fromMaybe "unix:///tmp/periodic.sock" h+ , xorFile = fromMaybe "" f+ , useTls = False+ , useWs = False+ , hostName = "localhost"+ , certKey = "client-key.pem"+ , cert = "client.pem"+ , caStore = "ca.pem"+ , httpHost = "127.0.0.1"+ , httpPort = 8080+ , poolSize = 10+ , showHelp = False+ }++parseOptions :: [String] -> Options -> Options+parseOptions [] opt = opt+parseOptions ("-H":x:xs) opt = parseOptions xs opt { host = x }+parseOptions ("--host":x:xs) opt = parseOptions xs opt { host = x }+parseOptions ("--xor":x:xs) opt = parseOptions xs opt { xorFile = x }+parseOptions ("--tls":xs) opt = parseOptions xs opt { useTls = True }+parseOptions ("--ws":xs) opt = parseOptions xs opt { useWs = True }+parseOptions ("--hostname":x:xs) opt = parseOptions xs opt { hostName = x }+parseOptions ("--cert-key":x:xs) opt = parseOptions xs opt { certKey = x }+parseOptions ("--cert":x:xs) opt = parseOptions xs opt { cert = x }+parseOptions ("--ca":x:xs) opt = parseOptions xs opt { caStore = x }+parseOptions ("--http-host":x:xs) opt = parseOptions xs opt { httpHost = x }+parseOptions ("--http-port":x:xs) opt = parseOptions xs opt { httpPort = read x }+parseOptions ("--pool-size":x:xs) opt = parseOptions xs opt { poolSize = read x }+parseOptions ("--help":xs) opt = parseOptions xs opt { showHelp = True }+parseOptions ("-h":xs) opt = parseOptions xs opt { showHelp = True }+parseOptions (_:xs) opt = parseOptions xs opt++printHelp :: IO ()+printHelp = do+ putStrLn "periodic-http-bridge - Periodic task system client http bridge"+ putStrLn ""+ putStrLn "Usage: periodic [--host|-H HOST] [--http-host HOST] [--http-port PORT] [--pool-size SIZE] [--xor FILE|--ws|--tls [--hostname HOSTNAME] [--cert-key FILE] [--cert FILE] [--ca FILE]]"+ putStrLn ""+ putStrLn "Available options:"+ putStrLn " -H --host Socket path [$PERIODIC_PORT]"+ putStrLn " eg: tcp://:5000 (optional: unix:///tmp/periodic.sock) "+ putStrLn " --xor XOR Transport encode file [$XOR_FILE]"+ putStrLn " --tls Use tls transport"+ putStrLn " --ws Use websockets transport"+ putStrLn " --hostname Host name"+ putStrLn " --cert-key Private key associated"+ putStrLn " --cert Public certificate (X.509 format)"+ putStrLn " --ca trusted certificates"+ putStrLn " --http-host HTTP host (optional: 127.0.0.1)"+ putStrLn " --http-port HTTP port (optional: 8080)"+ putStrLn " --pool-size Connection pool size"+ putStrLn " -h --help Display help message"+ putStrLn ""+ putStrLn "Version: v1.1.7.1"+ putStrLn ""+ exitSuccess++main :: IO ()+main = do+ h <- lookupEnv "PERIODIC_PORT"+ f <- lookupEnv "XOR_FILE"++ opts@Options{..} <- flip parseOptions (options h f) <$> getArgs++ when showHelp printHelp++ when (not ("tcp" `isPrefixOf` host) && not ("unix" `isPrefixOf` host)) $ do+ putStrLn $ "Invalid host " ++ host+ printHelp++ run opts def+ { settings = setPort httpPort+ $ setHost (Host httpHost) (settings def)}++run Options {useTls = True, ..} sopts = do+ prms <- makeClientParams' cert [] certKey caStore (hostName, B.pack $ fromMaybe "" $ getService host)+ clientEnv <- openPool (tlsConfig prms (socket host)) poolSize+ scottyOpts sopts $ application clientEnv++run Options {useWs = True, ..} sopts = do+ clientEnv <- openPool (clientConfig (socket host) (fromMaybe "0.0.0.0" $ getHost host) (fromMaybe "" $ getService host)) poolSize+ scottyOpts sopts $ application clientEnv++run Options {xorFile = "", ..} sopts = do+ clientEnv <- openPool (socket host) poolSize+ scottyOpts sopts $ application clientEnv++run Options {..} sopts = do+ clientEnv <- openPool (xorConfig xorFile $ socket host) poolSize+ scottyOpts sopts $ application clientEnv++application :: Transport tp => ClientPoolEnv tp -> ScottyM ()+application clientEnv = do+ get "/periodic/status/" $ statusHandler clientEnv+ post "/periodic/submit/:func_name/:job_name/" $ submitJobHandler clientEnv+ post "/periodic/run/:func_name/:job_name/" $ runJobHandler clientEnv+ post "/periodic/remove/:func_name/:job_name/" $ removeJobHandler clientEnv+ post "/periodic/drop/:func_name/" $ dropFuncHandler clientEnv++paramJob :: ActionM (FuncName, JobName)+paramJob = do+ fn <- fromString <$> param "func_name"+ jn <- fromString <$> param "job_name"+ return (fn, jn)++paramJob_ :: ActionM Job+paramJob_ = do+ (fn, jn) <- paramJob+ schedAt <- param "sched_at" `rescue` const (return 0)+ timeout <- param "timeout" `rescue` const (return 0)+ wb <- Workload . LB.toStrict <$> body+ pure $ setSchedAt schedAt $ setTimeout timeout $ setWorkload wb $ initJob fn jn++sampleRet :: Bool -> ActionM ()+sampleRet r = do+ WS.status $ if r then status204 else status500+ raw LB.empty++submitJobHandler :: Transport tp => ClientPoolEnv tp -> ActionM ()+submitJobHandler clientEnv = do+ job <- paramJob_+ r <- liftIO $ runClientPoolT clientEnv $ submitJob_ job+ sampleRet r++removeJobHandler :: Transport tp => ClientPoolEnv tp -> ActionM ()+removeJobHandler clientEnv = do+ (fn, jn) <- paramJob+ r <- liftIO $ runClientPoolT clientEnv $ removeJob fn jn+ sampleRet r++runJobHandler :: Transport tp => ClientPoolEnv tp -> ActionM ()+runJobHandler clientEnv = do+ job <- paramJob_+ r <- liftIO $ runClientPoolT clientEnv $ runJob_ job+ case r of+ Nothing -> do+ WS.status status500+ raw "error"+ Just bs -> raw $ LB.fromStrict bs++dropFuncHandler :: Transport tp => ClientPoolEnv tp -> ActionM ()+dropFuncHandler clientEnv = do+ fn <- fromString <$> param "func_name"+ r <- liftIO $ runClientPoolT clientEnv $ dropFunc fn+ sampleRet r++statusHandler :: Transport tp => ClientPoolEnv tp -> ActionM ()+statusHandler clientEnv = do+ r <- liftIO $ runClientPoolT clientEnv status+ raw $ LB.fromStrict r
+ app/periodic-run.hs view
@@ -0,0 +1,224 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}++module Main+ ( main+ ) where++import Control.Concurrent (forkIO, killThread)+import Control.DeepSeq (rnf)+import Control.Monad (unless, void, when)+import Control.Monad.IO.Class (liftIO)+import qualified Data.ByteString.Char8 as B (ByteString, pack)+import qualified Data.ByteString.Lazy as LB (null, toStrict)+import qualified Data.ByteString.Lazy.Char8 as LB (hGetContents, hPut)+import Data.List (isPrefixOf)+import Data.Maybe (fromMaybe)+import qualified Data.Text as T (unpack)+import Data.Text.Encoding (decodeUtf8With)+import Data.Text.Encoding.Error (ignore)+import Metro.Class (Transport)+import Metro.Socket (getHost, getService)+import Metro.TP.Socket (socket)+import Metro.TP.TLS (makeClientParams', tlsConfig)+import Metro.TP.WebSockets (clientConfig)+import Metro.TP.XOR (xorConfig)+import Periodic.Trans.Job (JobT, name, withLock_, workDone,+ workDone_, workFail, workload)+import Periodic.Trans.Worker (WorkerT, addFunc, broadcast,+ startWorkerT, work)+import Periodic.Types (FuncName (..), LockName (..))+import System.Environment (getArgs, lookupEnv)+import System.Exit (ExitCode (..), exitSuccess)+import System.IO (hClose)+import System.Process (CreateProcess (std_in, std_out),+ StdStream (CreatePipe, Inherit),+ proc, waitForProcess,+ withCreateProcess)+import UnliftIO (MVar, SomeException, evaluate,+ mask, newEmptyMVar, onException,+ putMVar, takeMVar, throwIO, try,+ tryIO)+++data Options = Options+ { host :: String+ , xorFile :: FilePath+ , useTls :: Bool+ , useWs :: Bool+ , hostName :: String+ , certKey :: FilePath+ , cert :: FilePath+ , caStore :: FilePath+ , thread :: Int+ , lockCount :: Int+ , lockName :: Maybe LockName+ , notify :: Bool+ , useData :: Bool+ , useName :: Bool+ , showHelp :: Bool+ }++options :: Maybe Int -> Maybe String -> Maybe String -> Options+options t h f = Options+ { host = fromMaybe "unix:///tmp/periodic.sock" h+ , xorFile = fromMaybe "" f+ , useTls = False+ , useWs = False+ , hostName = "localhost"+ , certKey = "client-key.pem"+ , cert = "client.pem"+ , caStore = "ca.pem"+ , thread = fromMaybe 1 t+ , lockCount = 1+ , lockName = Nothing+ , notify = False+ , useData = False+ , useName = True+ , showHelp = False+ }++parseOptions :: [String] -> Options -> (Options, FuncName, String, [String])+parseOptions ("-H":x:xs) opt = parseOptions xs opt { host = x }+parseOptions ("--host":x:xs) opt = parseOptions xs opt { host = x }+parseOptions ("--xor":x:xs) opt = parseOptions xs opt { xorFile = x }+parseOptions ("--tls":xs) opt = parseOptions xs opt { useTls = True }+parseOptions ("--ws":xs) opt = parseOptions xs opt { useWs = True }+parseOptions ("--hostname":x:xs) opt = parseOptions xs opt { hostName = x }+parseOptions ("--cert-key":x:xs) opt = parseOptions xs opt { certKey = x }+parseOptions ("--cert":x:xs) opt = parseOptions xs opt { cert = x }+parseOptions ("--ca":x:xs) opt = parseOptions xs opt { caStore = x }+parseOptions ("--thread":x:xs) opt = parseOptions xs opt { thread = read x }+parseOptions ("--lock-count":x:xs) opt = parseOptions xs opt { lockCount = read x }+parseOptions ("--lock-name":x:xs) opt = parseOptions xs opt { lockName = Just (LockName $ B.pack x) }+parseOptions ("--help":xs) opt = parseOptions xs opt { showHelp = True }+parseOptions ("--broadcast":xs) opt = parseOptions xs opt { notify = True }+parseOptions ("--data":xs) opt = parseOptions xs opt { useData = True }+parseOptions ("--no-name":xs) opt = parseOptions xs opt { useName = False }+parseOptions ("-h":xs) opt = parseOptions xs opt { showHelp = True }+parseOptions [] opt = (opt { showHelp = True }, "", "", [])+parseOptions [_] opt = (opt { showHelp = True }, "", "", [])+parseOptions (x:y:xs) opt = (opt, FuncName $ B.pack x, y, xs)++printHelp :: IO ()+printHelp = do+ putStrLn "periodic-run - Periodic task system worker"+ putStrLn ""+ putStrLn "Usage: periodic-run [--host|-H HOST] [--xor FILE|--ws|--tls [--hostname HOSTNAME] [--cert-key FILE] [--cert FILE] [--ca FILE] [--thread THREAD] [--lock-name NAME] [--lock-count COUNT] [--broadcast] [--data] [--no-name]] funcname command [options]"+ putStrLn ""+ putStrLn "Available options:"+ putStrLn " -H --host Socket path [$PERIODIC_PORT]"+ putStrLn " Eg: tcp://:5000 (optional: unix:///tmp/periodic.sock) "+ putStrLn " --xor XOR Transport encode file [$XOR_FILE]"+ putStrLn " --tls Use tls transport"+ putStrLn " --ws Use websockets transport"+ putStrLn " --hostname Host name"+ putStrLn " --cert-key Private key associated"+ putStrLn " --cert Public certificate (X.509 format)"+ putStrLn " --ca Trusted certificates"+ putStrLn " --thread Worker thread [$THREAD]"+ putStrLn " --lock-count Max lock count (optional: 1)"+ putStrLn " --lock-name The lock name (optional: no lock)"+ putStrLn " --broadcast Is broadcast worker"+ putStrLn " --data Send work data to client"+ putStrLn " --no-name Ignore the job name"+ putStrLn " -h --help Display help message"+ putStrLn ""+ putStrLn "Version: v1.1.7.1"+ putStrLn ""+ exitSuccess++main :: IO ()+main = do+ h <- lookupEnv "PERIODIC_PORT"+ f <- lookupEnv "XOR_FILE"+ t <- fmap read <$> lookupEnv "THREAD"++ (opts@Options {..}, func, cmd, argv) <- flip parseOptions (options t h f) <$> getArgs++ when showHelp printHelp++ when (not ("tcp" `isPrefixOf` host) && not ("unix" `isPrefixOf` host)) $ do+ putStrLn $ "Invalid host " ++ host+ printHelp++ run opts func cmd argv++doWork :: Transport tp => Options -> FuncName -> String -> [String] -> WorkerT tp IO ()+doWork opts@Options{..} func cmd argv = do+ let w = processWorker opts cmd argv+ if notify then broadcast func w+ else+ case lockName of+ Nothing -> addFunc func w+ Just n -> addFunc func $ withLock_ n lockCount w+ liftIO $ putStrLn "Worker started."+ work thread++run :: Options -> FuncName -> String -> [String] -> IO ()+run opts@Options {useTls = True, ..} func cmd argv = do+ prms <- makeClientParams' cert [] certKey caStore (hostName, B.pack $ fromMaybe "" $ getService host)+ startWorkerT (tlsConfig prms (socket host)) $ doWork opts func cmd argv++run opts@Options {useWs = True, ..} func cmd argv =+ startWorkerT (clientConfig (socket host) (fromMaybe "0.0.0.0" $ getHost host) (fromMaybe "" $ getService host)) $ doWork opts func cmd argv++run opts@Options {xorFile = "", ..} func cmd argv =+ startWorkerT (socket host) $ doWork opts func cmd argv++run opts@Options {..} func cmd argv =+ startWorkerT (xorConfig xorFile $ socket host) $ doWork opts func cmd argv++processWorker :: Transport tp => Options -> String -> [String] -> JobT tp IO ()+processWorker Options{..} cmd argv = do+ n <- name+ rb <- workload+ let argv' = if useName then argv ++ [n] else argv+ cp = (proc cmd argv') {std_in = CreatePipe, std_out= if useData then CreatePipe else Inherit}++ (code, out) <- liftIO $ withCreateProcess cp $ \mb_inh mb_outh _ ph ->+ case (mb_inh, mb_outh) of+ (Nothing, _) -> error "processWorker: Failed to get a stdin handle."+ (Just inh, Nothing) -> do+ unless (LB.null rb) $ void $ tryIO $ LB.hPut inh rb+ void $ tryIO $ hClose inh+ code <- waitForProcess ph+ return (code, Nothing)++ (Just inh, Just outh) -> do+ output <- LB.hGetContents outh+ withForkWait (evaluate $ rnf output) $ \waitOut -> do+ unless (LB.null rb) $ void $ tryIO $ LB.hPut inh rb+ void $ tryIO $ hClose inh++ waitOut+ hClose outh++ code <- waitForProcess ph+ return (code, Just output)+++ case code of+ ExitFailure _ -> workFail+ ExitSuccess ->+ case out of+ Nothing -> workDone+ Just wl -> workDone_ $ LB.toStrict wl++unpackBS :: B.ByteString -> String+unpackBS = T.unpack . decodeUtf8With ignore++-- | Fork a thread while doing something else, but kill it if there's an+-- exception.+--+-- This is important in the cases above because we want to kill the thread+-- that is holding the Handle lock, because when we clean up the process we+-- try to close that handle, which could otherwise deadlock.+--+withForkWait :: IO () -> (IO () -> IO a) -> IO a+withForkWait async body = do+ waitVar <- newEmptyMVar :: IO (MVar (Either SomeException ()))+ mask $ \restore -> do+ tid <- forkIO $ try (restore async) >>= putMVar waitVar+ let wait = takeMVar waitVar >>= either throwIO return+ restore (body wait) `onException` killThread tid
+ app/periodic.hs view
@@ -0,0 +1,375 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}++module Main+ ( main+ ) where++import Control.Monad (void, when)+import Control.Monad.IO.Class (liftIO)+import Data.Binary (decodeFile, encodeFile)+import Data.ByteString (ByteString)+import qualified Data.ByteString.Char8 as B (lines, pack, putStr, readFile,+ split, unpack)+import Data.Int (Int64)+import Data.List (isPrefixOf, transpose)+import Data.Maybe (fromMaybe)+import Metro.Class (Transport)+import Metro.Socket (getHost, getService)+import Metro.TP.Socket (socket)+import Metro.TP.TLS (makeClientParams', tlsConfig)+import Metro.TP.WebSockets (clientConfig)+import Metro.TP.XOR (xorConfig)+import Periodic.Trans.Client+import Periodic.Types (Workload (..))+import System.Environment (getArgs, lookupEnv)+import System.Exit (exitSuccess)+import Text.Read (readMaybe)++import Data.String (fromString)+import Data.UnixTime+import System.IO.Unsafe (unsafePerformIO)+import qualified Text.PrettyPrint.Boxes as T+++data Command = Status+ | Submit+ | Run+ | Remove+ | Drop+ | Config+ | Shutdown+ | Dump+ | Load+ | Help+ deriving (Eq)++parseCommand :: String -> Command+parseCommand "status" = Status+parseCommand "submit" = Submit+parseCommand "run" = Run+parseCommand "remove" = Remove+parseCommand "drop" = Drop+parseCommand "config" = Config+parseCommand "shutdown" = Shutdown+parseCommand "dump" = Dump+parseCommand "load" = Load+parseCommand _ = Help++data Options = Options+ { host :: String+ , xorFile :: FilePath+ , useTls :: Bool+ , useWs :: Bool+ , hostName :: String+ , certKey :: FilePath+ , cert :: FilePath+ , caStore :: FilePath+ }++options :: Maybe String -> Maybe String -> Options+options h f = Options { host = fromMaybe "unix:///tmp/periodic.sock" h+ , xorFile = fromMaybe "" f+ , useTls = False+ , useWs = False+ , hostName = "localhost"+ , certKey = "client-key.pem"+ , cert = "client.pem"+ , caStore = "ca.pem"+ }++parseOptions :: [String] -> Options -> (Command, Options, [String])+parseOptions [] opt = (Help, opt, [])+parseOptions ("-H":x:xs) opt = parseOptions xs opt { host = x }+parseOptions ("--host":x:xs) opt = parseOptions xs opt { host = x }+parseOptions ("--xor":x:xs) opt = parseOptions xs opt { xorFile = x }+parseOptions ("--tls":xs) opt = parseOptions xs opt { useTls = True }+parseOptions ("--ws":xs) opt = parseOptions xs opt { useWs = True }+parseOptions ("--hostname":x:xs) opt = parseOptions xs opt { hostName = x }+parseOptions ("--cert-key":x:xs) opt = parseOptions xs opt { certKey = x }+parseOptions ("--cert":x:xs) opt = parseOptions xs opt { cert = x }+parseOptions ("--ca":x:xs) opt = parseOptions xs opt { caStore = x }+parseOptions (x:xs) opt = (parseCommand x, opt, xs)++printHelp :: IO ()+printHelp = do+ putStrLn "periodic - Periodic task system client"+ putStrLn ""+ putStrLn "Usage: periodic [--host|-H HOST] [--xor FILE|--ws|--tls [--hostname HOSTNAME] [--cert-key FILE] [--cert FILE] [--ca FILE]] command [options]"+ putStrLn ""+ putStrLn "Commands:"+ putStrLn " status Show status"+ putStrLn " submit Submit job"+ putStrLn " run Run job and output result"+ putStrLn " remove Remove job"+ putStrLn " drop Drop function"+ putStrLn " config Set or Get config"+ putStrLn " shutdown Shutdown periodicd"+ putStrLn " dump Dump jobs"+ putStrLn " load Load jobs"+ putStrLn " help Shows a list of commands or help for one command"+ putStrLn ""+ putStrLn "Available options:"+ putStrLn " -H --host Socket path [$PERIODIC_PORT]"+ putStrLn " eg: tcp://:5000 (optional: unix:///tmp/periodic.sock) "+ putStrLn " --xor XOR Transport encode file [$XOR_FILE]"+ putStrLn " --tls Use tls transport"+ putStrLn " --ws Use websockets transport"+ putStrLn " --hostname Host name"+ putStrLn " --cert-key Private key associated"+ putStrLn " --cert Public certificate (X.509 format)"+ putStrLn " --ca trusted certificates"+ putStrLn ""+ putStrLn "Version: v1.1.7.1"+ putStrLn ""+ exitSuccess++printWorkloadHelp :: IO ()+printWorkloadHelp = do+ putStrLn ""+ putStrLn "Available options:"+ putStrLn " -w --workload WORKLOAD or @FILE"++printSubmitHelp :: IO ()+printSubmitHelp = do+ putStrLn "periodic submit - Submit job"+ putStrLn ""+ putStrLn "Usage: periodic submit funcname jobname [-w|--workload WORKLOAD|@FILE] [--later 0]"+ printWorkloadHelp+ putStrLn " --later Sched job later"+ putStrLn ""+ exitSuccess++printRunHelp :: IO ()+printRunHelp = do+ putStrLn "periodic run - Run job and output result"+ putStrLn ""+ putStrLn "Usage: periodic run funcname jobname [-w|--workload WORKLOAD|@FILE]"+ printWorkloadHelp+ putStrLn ""+ exitSuccess++printRemoveHelp :: IO ()+printRemoveHelp = do+ putStrLn "periodic remove - Remove job"+ putStrLn ""+ putStrLn "Usage: periodic remove funcname jobname [jobname...]"+ putStrLn ""+ exitSuccess++printDropHelp :: IO ()+printDropHelp = do+ putStrLn "periodic drop - Drop function"+ putStrLn ""+ putStrLn "Usage: periodic drop funcname [funcname...]"+ putStrLn ""+ exitSuccess++printConfigHelp :: IO ()+printConfigHelp = do+ putStrLn "periodic config - Set or get config"+ putStrLn ""+ putStrLn "Usage: periodic config command [options]"+ putStrLn ""+ putStrLn "Commands:"+ putStrLn " get Get config"+ putStrLn " set Set config"+ putStrLn ""+ exitSuccess++printConfigKeyList :: IO ()+printConfigKeyList = do+ putStrLn ""+ putStrLn "Available keys:"+ putStrLn " poll-interval - poll loop every time interval"+ putStrLn " revert-interval - revert process queue loop every time interval"+ putStrLn " timeout - job process timeout"+ putStrLn " keepalive - client keepalive"+ putStrLn " max-batch-size - max poll batch size"+ putStrLn " expiration - run job result cache expiration"+ putStrLn ""+ exitSuccess++printConfigGetHelp :: IO ()+printConfigGetHelp = do+ putStrLn "periodic config get - Get config"+ putStrLn ""+ putStrLn "Usage: periodic config get key"+ printConfigKeyList++printConfigSetHelp :: IO ()+printConfigSetHelp = do+ putStrLn "periodic config set - Set config"+ putStrLn ""+ putStrLn "Usage: periodic config set key val"+ printConfigKeyList++printLoadHelp :: IO ()+printLoadHelp = do+ putStrLn "periodic load - Load jobs"+ putStrLn ""+ putStrLn "Usage: periodic load file"+ putStrLn ""+ exitSuccess++printDumpHelp :: IO ()+printDumpHelp = do+ putStrLn "periodic dump - Dump jobs"+ putStrLn ""+ putStrLn "Usage: periodic dump file"+ putStrLn ""+ exitSuccess++main :: IO ()+main = do+ h <- lookupEnv "PERIODIC_PORT"+ f <- lookupEnv "XOR_FILE"++ (cmd, opts@Options {..}, argv) <- flip parseOptions (options h f) <$> getArgs++ let argc = length argv++ when (cmd == Help) printHelp++ when (cmd == Submit && argc < 2) printSubmitHelp+ when (cmd == Run && argc < 2) printRunHelp+ when (cmd == Remove && argc < 2) printRemoveHelp+ when (cmd == Drop && argc < 1) printDropHelp+ when (cmd == Config && argc < 1) printConfigHelp+ when (cmd == Load && argc < 1) printLoadHelp+ when (cmd == Dump && argc < 1) printDumpHelp++ when (not ("tcp" `isPrefixOf` host) && not ("unix" `isPrefixOf` host)) $ do+ putStrLn $ "Invalid host " ++ host+ printHelp++ run opts cmd argv++run Options {useTls = True, ..} cmd argv = do+ prms <- makeClientParams' cert [] certKey caStore (hostName, B.pack $ fromMaybe "" $ getService host)+ clientEnv <- open (tlsConfig prms (socket host))+ runClientT clientEnv $ processCommand cmd argv++run Options {useWs = True, ..} cmd argv = do+ clientEnv <- open (clientConfig (socket host) (fromMaybe "0.0.0.0" $ getHost host) (fromMaybe "" $ getService host))+ runClientT clientEnv $ processCommand cmd argv++run Options {xorFile = "", ..} cmd argv = do+ clientEnv <- open (socket host)+ runClientT clientEnv $ processCommand cmd argv++run Options {..} cmd argv = do+ clientEnv <- open (xorConfig xorFile $ socket host)+ runClientT clientEnv $ processCommand cmd argv++processCommand :: Transport tp => Command -> [String] -> ClientT tp IO ()+processCommand Help _ = liftIO printHelp+processCommand Status _ = doStatus+processCommand Submit xs = doSubmitJob xs+processCommand Run xs = doRunJob xs+processCommand Remove xs = doRemoveJob xs+processCommand Drop xs = doDropFunc xs+processCommand Config xs = doConfig xs+processCommand Load xs = doLoad xs+processCommand Dump xs = doDump xs+processCommand Shutdown _ = shutdown++doRemoveJob :: Transport tp => [String] -> ClientT tp IO ()+doRemoveJob (x:xs) = mapM_ (removeJob (fromString x) . fromString) xs+doRemoveJob [] = liftIO printRemoveHelp++doDropFunc :: Transport tp => [String] -> ClientT tp IO ()+doDropFunc = mapM_ (dropFunc . fromString)++doConfig :: Transport tp => [String] -> ClientT tp IO ()+doConfig ["get"] = liftIO printConfigGetHelp+doConfig ["get", k] = do+ v <- configGet k+ liftIO $ print v+doConfig ("get":_:_) = liftIO printConfigGetHelp+doConfig ["set"] = liftIO printConfigSetHelp+doConfig ["set", _] = liftIO printConfigSetHelp+doConfig ["set", k, v] = void $ configSet k (read v)+doConfig ("set":_:_:_) = liftIO printConfigSetHelp+doConfig _ = liftIO printConfigHelp++doLoad :: Transport tp => [String] -> ClientT tp IO ()+doLoad [fn] = do+ jobs <- liftIO $ decodeFile fn+ void $ load jobs++doLoad _ = liftIO printLoadHelp++doDump :: Transport tp => [String] -> ClientT tp IO ()+doDump [fn] = do+ jobs <- dump+ liftIO $ encodeFile fn jobs++doDump _ = liftIO printDumpHelp++doSubmitJob :: Transport tp => [String] -> ClientT tp IO ()+doSubmitJob [] = liftIO printSubmitHelp+doSubmitJob [_] = liftIO printSubmitHelp+doSubmitJob (x:y:xs) = do+ (wl, l) <- liftIO $ go (Nothing, Nothing) xs+ void $ submitJob (fromString x) (fromString y) wl l++ where go :: (Maybe Workload, Maybe Int64) -> [String] -> IO (Maybe Workload, Maybe Int64)+ go v [] = pure v+ go (_, l0) ("-w":('@':f):ys) = do+ w <- B.readFile f+ go (Just (Workload w), l0) ys+ go (_, l0) ("--workload":('@':f):ys) = do+ w <- B.readFile f+ go (Just (Workload w), l0) ys+ go (_, l0) ("-w":w:ys) = go (Just (fromString w), l0) ys+ go (_, l0) ("--workload":w:ys) = go (Just (fromString w), l0) ys+ go (w, _) ("--later":l0:ys) = go (w, readMaybe l0) ys+ go v (_:ys) = go v ys++doRunJob :: Transport tp => [String] -> ClientT tp IO ()+doRunJob [] = liftIO printRunHelp+doRunJob [_] = liftIO printRunHelp+doRunJob (x:y:xs) =+ liftIO . putR+ =<< runJob (fromString x) (fromString y)+ =<< liftIO (go xs)++ where go :: [String] -> IO (Maybe Workload)+ go [] = pure Nothing+ go ("-w":('@':f):_) = getFile f+ go ("--workload":('@':f):_) = getFile f+ go ("-w":w:_) = pure $ Just (fromString w)+ go ("--workload":w:_) = pure $ Just (fromString w)+ go (_:ys) = go ys++ getFile :: String -> IO (Maybe Workload)+ getFile fn = Just . Workload <$> B.readFile fn++ putR :: Maybe ByteString -> IO ()+ putR Nothing = putStrLn "Error: run job failed"+ putR (Just bs) = B.putStr bs++doStatus :: Transport tp => ClientT tp IO ()+doStatus = do+ st <- map formatTime . unpackBS . map (B.split ',') . B.lines <$> status+ liftIO $ printTable (["FUNCTIONS", "WORKERS", "JOBS", "PROCESSING", "LOCKING", "SCHEDAT"]:st)++unpackBS :: [[ByteString]] -> [[String]]+unpackBS = map (map B.unpack)++formatTime :: [String] -> [String]+formatTime [] = []+formatTime [x] = [formatUnixTimeLocal x]+formatTime (x:xs) = x:formatTime xs++formatUnixTimeLocal :: String -> String+formatUnixTimeLocal = B.unpack+ . unsafePerformIO+ . formatUnixTime "%Y-%m-%d %H:%M:%S"+ . fromEpochTime+ . fromIntegral+ . read++printTable :: [[String]] -> IO ()+printTable rows = T.printBox $ T.hsep 2 T.left (map (T.vcat T.left . map T.text) (transpose rows))
+ periodic-client-exe.cabal view
@@ -0,0 +1,76 @@+name: periodic-client-exe+version: 1.1.7.1+synopsis: Periodic task system haskell client executables+description: Periodic task system haskell client executables.+homepage: https://github.com/Lupino/haskell-periodic/tree/master/periodic-client-exe#readme+license: BSD3+license-file: LICENSE+author: Li Meng Jun+maintainer: lmjubuntu@gmail.com+copyright: MIT+category: System,Web+build-type: Simple+extra-source-files: README.md+cabal-version: >=1.10++executable periodic+ hs-source-dirs: app+ main-is: periodic.hs+ ghc-options: -threaded -rtsopts -with-rtsopts=-N+ build-depends: base >= 4.7 && < 5+ , periodic-client+ , periodic-common+ , bytestring+ , boxes+ , unix-time+ , binary+ , metro+ , metro-socket+ , metro-transport-xor+ , metro-transport-tls+ , metro-transport-websockets+ default-language: Haskell2010++executable periodic-run+ hs-source-dirs: app+ main-is: periodic-run.hs+ ghc-options: -threaded -rtsopts -with-rtsopts=-N+ build-depends: base >= 4.7 && < 5+ , periodic-client+ , periodic-common+ , bytestring+ , text+ , process+ , unliftio+ , deepseq+ , metro+ , metro-socket+ , metro-transport-xor+ , metro-transport-tls+ , metro-transport-websockets+ default-language: Haskell2010++executable periodic-http-bridge+ hs-source-dirs: app+ main-is: periodic-http-bridge.hs+ ghc-options: -threaded -rtsopts -with-rtsopts=-N+ build-depends: base >= 4.7 && < 5+ , periodic-client+ , periodic-common+ , bytestring+ , scotty+ , warp+ , data-default-class+ , streaming-commons+ , http-types+ , metro+ , metro-socket+ , metro-transport-xor+ , metro-transport-tls+ , metro-transport-websockets+ , websockets+ default-language: Haskell2010++source-repository head+ type: git+ location: https://github.com/Lupino/haskell-periodic