Copyright	(c) Gargantext 2024-Present
License	AGPL
Maintainer	gargantext@iscpif.fr
Stability	experimental
Portability	POSIX
Safe Haskell	Safe-Inferred
Language	Haskell2010

Async.Worker.Types

Contents

Main worker state
Job wrapped in metadata
Events emitted during job lifetime
Other useful types and functions

Description

Types for worker.

Synopsis

data State b a = State {
- broker :: Broker b (Job a)
- queueName :: Queue
- name :: String
- performAction :: PerformAction b a
- onMessageReceived :: WorkerJobEvent b a
- onJobFinish :: WorkerJobEvent b a
- onJobTimeout :: WorkerJobEvent b a
- onJobError :: WorkerJobErrorEvent b a
- onWorkerKilledSafely :: WorkerMJobEvent b a
}
data Job a = Job {
- job :: a
- metadata :: JobMetadata
}
data JobMetadata = JobMetadata {
- archiveStrategy :: ArchiveStrategy
- errorStrategy :: ErrorStrategy
- timeoutStrategy :: TimeoutStrategy
- timeout :: Timeout
- readCount :: ReadCount
- resendWhenWorkerKilled :: Bool
}
defaultMetadata :: JobMetadata
data ArchiveStrategy
- = ASDelete
- | ASArchive
data ErrorStrategy
- = ESDelete
- | ESArchive
- | ESRepeatNElseArchive Int
data TimeoutStrategy
- = TSDelete
- | TSArchive
- | TSRepeat
- | TSRepeatNElseArchive Int
- | TSRepeatNElseDelete Int
getJob :: Job a -> a
jobTimeout :: Job a -> Timeout
type Timeout = Int
runAction :: State b a -> BrokerMessage b (Job a) -> IO ()
type PerformAction b a = State b a -> BrokerMessage b (Job a) -> IO ()
type WorkerJobEvent b a = Maybe (State b a -> BrokerMessage b (Job a) -> IO ())
type WorkerJobErrorEvent b a = Maybe (State b a -> BrokerMessage b (Job a) -> SomeException -> IO ())
type WorkerMJobEvent b a = Maybe (State b a -> Maybe (BrokerMessage b (Job a)) -> IO ())
type HasWorkerBroker b a = (MessageBroker b (Job a), Typeable a, Typeable b, Show a)
formatStr :: State b a -> String -> String
data JobTimeout b a = JobTimeout {
- jtBMessage :: BrokerMessage b (Job a)
- jtTimeout :: Timeout
}

Main worker state

data State b a Source #

Main state for a running worker (b is Broker, a is the underlying message).

a is the underlying message specification and is implementation-dependent, e.g. can be of JSON form

{"function": ..., "arguments": [...]}

Job a is worker's wrapper around that message with metadata (corresponds to broker Message b a). BrokerMessage 'b a, on the other hand, is what we get when the broker reads that message).

Note that our underlying Broker handles messages of type Job a.

Constructors

State

Fields

broker :: Broker b (Job a)
queueName :: Queue
Queue associated with this worker. If you want to support more queues, spawn more workers.
name :: String
Name of this worker (useful for debugging).
performAction :: PerformAction b a
Actions that will be performed when message a arrives.
onMessageReceived :: WorkerJobEvent b a
Event emitted after job was received from broker
onJobFinish :: WorkerJobEvent b a
Event emitted after job was finished successfully
onJobTimeout :: WorkerJobEvent b a
Event emitted after job timed out
onJobError :: WorkerJobErrorEvent b a
Event emitted after job ended with error
onWorkerKilledSafely :: WorkerMJobEvent b a
Event emitted when worker is safely killed (don't overuse it)

Job wrapped in metadata

data Job a Source #

Worker Job is a (defining action to call via $sel:performAction:State) together with associated JobMetadata.

Constructors

Job
Fields job :: a metadata :: JobMetadata

Instances

Instances details

FromJSON a => FromJSON (Job a) Source #
Instance details Defined in Async.Worker.Types Methods parseJSON :: Value -> Parser (Job a) # parseJSONList :: Value -> Parser [Job a] # omittedField :: Maybe (Job a) #
ToJSON a => ToJSON (Job a) Source #
Instance details Defined in Async.Worker.Types Methods toJSON :: Job a -> Value # toEncoding :: Job a -> Encoding # toJSONList :: [Job a] -> Value # toEncodingList :: [Job a] -> Encoding # omitField :: Job a -> Bool #
Show a => Show (Job a) Source #
Instance details Defined in Async.Worker.Types Methods showsPrec :: Int -> Job a -> ShowS # show :: Job a -> String # showList :: [Job a] -> ShowS #
Eq a => Eq (Job a) Source #
Instance details Defined in Async.Worker.Types Methods (==) :: Job a -> Job a -> Bool # (/=) :: Job a -> Job a -> Bool #

Job metadata

data JobMetadata Source #

Metadata associated with a job.

Constructors

JobMetadata

Fields

archiveStrategy :: ArchiveStrategy
What to do after job is finished successfully
errorStrategy :: ErrorStrategy
What to do when a job ends with error
timeoutStrategy :: TimeoutStrategy
What to do when a job ends with timeout
timeout :: Timeout
Time after which the job is considered to time-out (in seconds).
readCount :: ReadCount
Read count so we know how many times this message was processed
resendWhenWorkerKilled :: Bool
A worker might have processed a task and be killed. If $sel:resendWhenWorkerKilled:JobMetadata is True, this job will be resent to broker and picked up later. Otherwise it will be discarded.

Instances

Instances details

FromJSON JobMetadata Source #
Instance details Defined in Async.Worker.Types Methods parseJSON :: Value -> Parser JobMetadata # parseJSONList :: Value -> Parser [JobMetadata] # omittedField :: Maybe JobMetadata #
ToJSON JobMetadata Source #
Instance details Defined in Async.Worker.Types Methods toJSON :: JobMetadata -> Value # toEncoding :: JobMetadata -> Encoding # toJSONList :: [JobMetadata] -> Value # toEncodingList :: [JobMetadata] -> Encoding # omitField :: JobMetadata -> Bool #
Show JobMetadata Source #
Instance details Defined in Async.Worker.Types Methods showsPrec :: Int -> JobMetadata -> ShowS # show :: JobMetadata -> String # showList :: [JobMetadata] -> ShowS #
Eq JobMetadata Source #
Instance details Defined in Async.Worker.Types Methods (==) :: JobMetadata -> JobMetadata -> Bool # (/=) :: JobMetadata -> JobMetadata -> Bool #

defaultMetadata :: JobMetadata Source #

For a typical Job it's probably sane to just archive it no matter how it finished.

Strategies for handling finished, errored, timed-out jobs

data ArchiveStrategy Source #

Strategy for archiving finished jobs

Constructors

ASDelete	Delete message when it's done
ASArchive	Archive message when it's done

Instances

Instances details

FromJSON ArchiveStrategy Source #
Instance details Defined in Async.Worker.Types Methods parseJSON :: Value -> Parser ArchiveStrategy # parseJSONList :: Value -> Parser [ArchiveStrategy] # omittedField :: Maybe ArchiveStrategy #
ToJSON ArchiveStrategy Source #
Instance details Defined in Async.Worker.Types Methods toJSON :: ArchiveStrategy -> Value # toEncoding :: ArchiveStrategy -> Encoding # toJSONList :: [ArchiveStrategy] -> Value # toEncodingList :: [ArchiveStrategy] -> Encoding # omitField :: ArchiveStrategy -> Bool #
Show ArchiveStrategy Source #
Instance details Defined in Async.Worker.Types Methods showsPrec :: Int -> ArchiveStrategy -> ShowS # show :: ArchiveStrategy -> String # showList :: [ArchiveStrategy] -> ShowS #
Eq ArchiveStrategy Source #
Instance details Defined in Async.Worker.Types Methods (==) :: ArchiveStrategy -> ArchiveStrategy -> Bool # (/=) :: ArchiveStrategy -> ArchiveStrategy -> Bool #

data ErrorStrategy Source #

Strategy for handling jobs with errors

Constructors

ESDelete	Delete job when it threw an error
ESArchive	Archive job when it threw an error
ESRepeatNElseArchive Int	Try to repeat the job when an error ocurred (at most N times), otherwise archive the job.

Instances

Instances details

FromJSON ErrorStrategy Source #
Instance details Defined in Async.Worker.Types Methods parseJSON :: Value -> Parser ErrorStrategy # parseJSONList :: Value -> Parser [ErrorStrategy] # omittedField :: Maybe ErrorStrategy #
ToJSON ErrorStrategy Source #
Instance details Defined in Async.Worker.Types Methods toJSON :: ErrorStrategy -> Value # toEncoding :: ErrorStrategy -> Encoding # toJSONList :: [ErrorStrategy] -> Value # toEncodingList :: [ErrorStrategy] -> Encoding # omitField :: ErrorStrategy -> Bool #
Show ErrorStrategy Source #
Instance details Defined in Async.Worker.Types Methods showsPrec :: Int -> ErrorStrategy -> ShowS # show :: ErrorStrategy -> String # showList :: [ErrorStrategy] -> ShowS #
Eq ErrorStrategy Source #
Instance details Defined in Async.Worker.Types Methods (==) :: ErrorStrategy -> ErrorStrategy -> Bool # (/=) :: ErrorStrategy -> ErrorStrategy -> Bool #

data TimeoutStrategy Source #

Strategy for handling job timeouts

Constructors

TSDelete	Delete job when it timed out
TSArchive	Archive job when it timed out
TSRepeat	Repeat job when it timed out (infinitely)
TSRepeatNElseArchive Int	Repeat job when it timed out (at most N times), otherwise archive it
TSRepeatNElseDelete Int	Repeat job when it timed out (at most N times), otherwise delete it

Instances

Instances details

FromJSON TimeoutStrategy Source #
Instance details Defined in Async.Worker.Types Methods parseJSON :: Value -> Parser TimeoutStrategy # parseJSONList :: Value -> Parser [TimeoutStrategy] # omittedField :: Maybe TimeoutStrategy #
ToJSON TimeoutStrategy Source #
Instance details Defined in Async.Worker.Types Methods toJSON :: TimeoutStrategy -> Value # toEncoding :: TimeoutStrategy -> Encoding # toJSONList :: [TimeoutStrategy] -> Value # toEncodingList :: [TimeoutStrategy] -> Encoding # omitField :: TimeoutStrategy -> Bool #
Show TimeoutStrategy Source #
Instance details Defined in Async.Worker.Types Methods showsPrec :: Int -> TimeoutStrategy -> ShowS # show :: TimeoutStrategy -> String # showList :: [TimeoutStrategy] -> ShowS #
Eq TimeoutStrategy Source #
Instance details Defined in Async.Worker.Types Methods (==) :: TimeoutStrategy -> TimeoutStrategy -> Bool # (/=) :: TimeoutStrategy -> TimeoutStrategy -> Bool #

Job utility functions

getJob :: Job a -> a Source #

Given worker Job a, return the a part

jobTimeout :: Job a -> Timeout Source #

For a given Job, return it's $sel:timeout:JobMetadata value from JobMetadata.

type Timeout = Int Source #

runAction :: State b a -> BrokerMessage b (Job a) -> IO () Source #

Helper function to call an action for given worker, for a BrokerMessage.

Actions that the worker will perform

type PerformAction b a = State b a -> BrokerMessage b (Job a) -> IO () Source #

Callback definition (what to execute when a message arrives)

Events emitted during job lifetime

type WorkerJobEvent b a = Maybe (State b a -> BrokerMessage b (Job a) -> IO ()) Source #

type WorkerJobErrorEvent b a = Maybe (State b a -> BrokerMessage b (Job a) -> SomeException -> IO ()) Source #

type WorkerMJobEvent b a = Maybe (State b a -> Maybe (BrokerMessage b (Job a)) -> IO ()) Source #

Other useful types and functions

type HasWorkerBroker b a = (MessageBroker b (Job a), Typeable a, Typeable b, Show a) Source #

TODO Show a could be removed. Any logging can be done as part of on events in State.

formatStr :: State b a -> String -> String Source #

Helper function to format a string with worker name (for logging)

data JobTimeout b a Source #

An exception, thrown when job times out

Constructors

JobTimeout
Fields jtBMessage :: BrokerMessage b (Job a) jtTimeout :: Timeout

Instances

Instances details

(Show (BrokerMessage b (Job a)), Typeable a, Typeable b) => Exception (JobTimeout b a) Source #
Instance details Defined in Async.Worker.Types Methods toException :: JobTimeout b a -> SomeException # fromException :: SomeException -> Maybe (JobTimeout b a) # displayException :: JobTimeout b a -> String #
Show (BrokerMessage b (Job a)) => Show (JobTimeout b a) Source #
Instance details Defined in Async.Worker.Types Methods showsPrec :: Int -> JobTimeout b a -> ShowS # show :: JobTimeout b a -> String # showList :: [JobTimeout b a] -> ShowS #