# Home ## A protocol-based queueing system for ColdBox Queues allow you to push work to the background, schedule work to be done later, or even process work on many different machines. It runs on a provider-based system allowing a unified API to talk with many different queue backends. ### Where to go next? * [Installation](/2.0.0/getting-started/installation) * [Walkthrough](/2.0.0/getting-started/walkthrough) * CFCasts Series (Coming Soon) * API Docs # What's New? ## v2.0.5 **DBProvider**: Disable `forceRun` because it is causing ColdBox Futures to lose mappings. ## v**2.0.4** Reload module mappings in an attempt to work around ColdBox Async losing them. ## **v2.0.3** **SyncProvider:** Add pool to releaseJob call ## v2.0.2 Fix moduleSettings missing a queryOptions key for failed jobs ## v2.0.1 ColdBoxAsyncProvider now correctly respects Worker Pool conifguration, including queues. ## v2.0.0 ### BREAKING CHANGES #### Worker Pools can only define a single queue to work In order to work with new Queue Providers, the Worker Pools need to be updated to only work a specific queue. This is because many future Queue Providers like RabbitMQ and Amazon SQS only support listening to a single queue in a consumer. If you previously had multiple queues defined in a Worker Pool, you will need to define multiple Worker Pool instances, one for each of the queues. ```cfscript // Old newWorkerPool( "default" ) .forConnection( "default" ) .onQueues( [ "priority", "default" ] ); // New newWorkerPool( "default" ) .forConnection( "default" ) .onQueue( "priority" ); newWorkerPool( "default" ) .forConnection( "default" ) .onQueue( "default" ); ``` Notice that the method has been renamed from `onQueues` to `onQueue`. Additionally, there are no more wildcard queues. Every queue you publish to must have a WorkerPool defined in order for that Job to be worked. Finally, queue priorities are defined by the number of workers (`quantity`) you define for the WorkerPool. WorkerPools can no longer share workers across queues. # Installation ## Requirements cbq requires the following: * Adobe ColdFusion (ACF) 2018+ **OR** Lucee 5+ * ColdBox 6+ The different [Queue Providers](/2.0.0/configuration/providers) each have their own requirements that are listed on their individual pages. ## Install via ForgeBox with CommandBox cbq is installed via [ForgeBox](https://forgebox.io) with [CommandBox](https://www.ortussolutions.com/products/commandbox). You can install the latest version using the command: ```shell install cbq ``` ## Load Java Libraries When using batches, cbq utilizes additional Java libraries included in the `lib/` folder. These need to be added to your Application's `javaSettings` in `Application.cfc`. ```cfscript // Java Integration this.javaSettings = { loadPaths: [ expandPath( "./modules/cbq/lib" ) ], loadColdFusionClassPath: true, reloadOnChange: false }; ``` {% hint style="info" %} Feel free to use mappings to point to the cbq path, if needed. {% endhint %} ## Additional Provider Installation Steps The [Queue Provider ](/2.0.0/configuration/providers)you use may have additional installation steps. Check out the individual provider pages for more details. # Walkthrough ## Install cbq Refer to the [Installation](/2.0.0/getting-started/installation) section for general installation instructions as well as instructions for your specific [provider](/2.0.0/configuration/providers). ## Create a Queue Connection Open up your `config/cbq.cfc` file and create your first Queue Connection: ```cfscript component { function configure() { newConnection( "default" ) .setProvider( "ColdBoxAsyncProvider@cbq" ); } } ``` ## Create a Worker Pool Next, create a Worker Pool for your new Queue Connection. This allows our application to work the Jobs we will dispatch. ```cfscript component { function configure() { newConnection( "default" ) .setProvider( "ColdBoxAsyncProvider@cbq" ); newWorkerPool( "default" ) .forConnection( "default" ); } } ``` ## Define your first Job A Job is a CFC that extends `cbq.models.Jobs.AbstractJob`. It can live anywhere in your application. ```cfscript // models/jobs/emails/SendWelcomeEmailJob.cfc component extends="cbq.models.Jobs.AbstractJob" { property name="userId"; function handle() { log.info( "Sending a Welcome email to User ###getUserId()#" ); /* sample code var user = getInstance( "User" ).findOrFail( getUserId() ); getInstance( "MailService@cbmailservices" ) .newMail( from = "no-reply@example.com", to = user.getEmail(), subject = "Welcome!", type = "html" ) .setView( "/_emails/users/welcome" ) .setBodyTokens( { "firstName" : user.getFirstName(), "lastName" : user.getLastName() } ) .send(); */ } } ``` ## Create an instance of your Job You can create an instance of your Job anywhere in your code — handlers, services, models, etc. Populate it with the specific data needed for this instance. ```cfscript var job = getInstance( "SendWelcomeEmailJob" ); job.setUserId( newUser.getId() ); ``` ## Dispatch your Job Once your Job is created and configured, `dispatch` it to the Queue Connection. ```cfscript job.dispatch(); ``` ## Watch your job get executed Check out LogBox to see your Job being executed. Congratulations! You've dispatched your first background Job using cbq! # Module Settings ## Full Module Settings ```cfscript settings = { // The path the custom config file to register connections and worker pools "configPath" : "config.cbq", // Flag if workers should be registered. // If your application only pushes to the queues, you can set this to `false`. "registerWorkers" : getSystemSetting( "CBQ_REGISTER_WORKERS", true ), // The interval to poll for changes to the worker pool scaling. // Defaults to 0 which turns off the scheduled scaling feature. "scaleInterval" : 0, // The default amount of time, in seconds, to delay a job. // Used if the connection and job doesn't define their own. "defaultWorkerBackoff" : 0, // The default amount of time, in seconds, to wait before timing out a job. // Used if the connection and job doesn't define their own. "defaultWorkerTimeout" : 60, // The default amount of attempts to try before failing a job. // Used if the connection and job doesn't define their own. "defaultWorkerMaxAttempts" : 1, // Datasource information for tracking batches. "batchRepositoryProperties" : { "tableName" : "cbq_batches", "datasource" : "", // `datasource` can also be a struct "queryOptions" : {} // The sibling `datasource` property overrides any defined datasource in queryOptions. }, // Flag to turn on logging failed jobs to a database table. "logFailedJobs" : false, // Datasource information for loggin failed jobs. "logFailedJobsProperties" : { "tableName" : "cbq_failed_jobs", "datasource" : "", // `datasource` can also be a struct. "queryOptions" : {} // The sibling `datasource` property overrides any defined datasource in `queryOptions`. } }; ``` ### configPath The configPath is a dot-delimited path to your cbq Config Component. By convention, this should be placed in your application's `config/` folder alongside other config files like `ColdBox.cfc` and `WireBox.cfc`. ### registerWorkers This flag is responsible for spinning up Worker Pools when the application starts. If a particular instance of your application should **not** work jobs as well as dispatch them, then this setting should be set to `false`. To make this easy, you can set the `CBQ_REGISTER_WORKERS` environment variable and it will be picked up. (If you override this setting in your own `moduleSettings` it will still take precedence over the environment variable.) ### scaleInterval {% hint style="danger" %} This feature is not implemented yet. {% endhint %} This is the interval in seconds that the scale job is ran in the background. The scale job enables you to scale Worker Pools up or down based on any other factors in your application. **Setting this value to `0` disables the job entirely.** ### defaultWorkerBackoff This setting provides a default backoff value for a Worker Pool. This can still be overridden on a configured Worker Pool in your cbq config file or on an individual Job or `dispatch` call. ### defaultWorkerTimeout This setting provides a default timeout value for a Worker Pool. This can still be overridden on a configured Worker Pool in your cbq config file or on an individual Job or `dispatch` call. ### defaultWorkerMaxAttempts This setting provides a default max attempts value for a Worker Pool. This can still be overridden on a configured Worker Pool in your cbq config file or on an individual Job or `dispatch` call. ### batchRepositoryProperties A struct of configuration properties for a Batch Repository. This is only needed if you dispatch any batches. #### tableName The name of the batch table. The default is `cbq_batches`. #### datasource The datasource to use to interact with the batch table. If no datasource is provided, it uses the default application datasource. A `struct` can also be provided if your CFML engine supports it. #### queryOptions A struct of query options to pass to the `queryExecute` call when interacting with the batch table. If a `datasource` is defined above, it will override any `datasource` key inside the `queryOptions`. ### logFailedJobs This flag will send failed jobs to a configured database table when true. In some systems, this is called a Dead Letter Queue or DLQ. ### logFailedJobsProperties #### tableName The name of the failed jobs table. The default is `cbq_failed_jobs`. #### datasource The datasource to use to interact with the failed jobs table. If no datasource is provided, it uses the default application datasource. A `struct` can also be provided if your CFML engine supports it. #### queryOptions A struct of query options to pass to the `queryExecute` call when interacting with the failed jobs table. If a `datasource` is defined above, it will override any `datasource` key inside the `queryOptions`. # Config File The cbq config file is where you define Queue Connections as well as Worker Pools. These Connections and Worker Pools can also be added, removed, or modified based on the current environment. When you want to change where your queued Jobs are sent or how they are worked, this is the file you will modify. ## Definitions ### Queue Connections Also referred to as "Connection." A Queue Connection defines where Jobs are serialized and the default settings that apply. ## configure The `configure` method is where the production Queue Connection and Worker Pools definitions are constructed. ### newConnection This command creates a new `QueueConnectionDefinition` builder component. It requires a unique name that will define this Connection and that will be used when defining Worker Pools.
ArgumentsTypeRequiredDefaultDescription
namestringtrueThe unique name for the new Queue Connection.
```cfscript component { function configure() { newConnection( "default" ); } } ``` A `QueueConnectionDefinition` has various methods to configure the Queue Connection. {% content-ref url="/pages/pghfGWdoAmKqeFiwqKtz" %} [Queue Connection](/2.0.0/configuration/config-file/queue-connection) {% endcontent-ref %} ### newWorkerPool This command creates a new `WorkerPoolDefinition` builder component. It requires a unique `name` that will define this Worker Pool and a `connectionName` that points to an already created Connection.
ArgumentsTypeRequiredDefaultDescription
namestringtrueThe unique name for the new Worker Pool.
connectionNamestringfalseA reference to an existing Connection. If not passed in here, the forConnection method must be called to define the Connection.
quantitynumericfalse1The number of workers to spin up for this Worker Pool.
queuesarrayfalse[ * ]An array of queues that this Worker Pool will work. A queue of * refers to all queues. Queues will be worked in the order provided.
forcebooleanfalsefalseIf false, an exception will be thrown if the Worker Pool name has already been registered. If true, the new definition will override the existing definition.
component {

    function configure() {
        newConnection( "default" );
        
        newWorkerPool( "default" ).forConnection( "default" );
    }

}
A `WorkerPoolDefinition` has various methods to configure the Worker Pool. {% content-ref url="/pages/FRcimvpoy6mO5qGgTvgb" %} [Worker Pool](/2.0.0/configuration/config-file/worker-pool) {% endcontent-ref %} ### reset Removes all Connections and Worker Pools ```cfscript getInstance( "Config@cbq" ).reset(); ``` ### Environment Overrides Just as in your `config/ColdBox.cfc` file or in `ModuleConfig.cfc` files, you can add, remove, or modify Queue Connection or Worker Pool definitions per environment. You do this by defining a method on your Config component matching the environment name you want to override. ```cfscript component { function configure() { newConnection( "default" ) .provider( "DBProvider@cbq" ); newWorkerPool( "default" ) .forConnection( "default" ) .quantity( 3 ) .timeout( 15 ); } /** * This method will be called after `configure` * and only if the current environment is `development`. */ function development() { withConnection( "default" ) .provider( "SyncProvider@cbq" ); withWorkerPool( "default" ) .quantity( 1 ) .timeout( 60 ); } } ``` ### withConnection Retrieves an already defined Queue Connection Definition for overriding. All the same methods are available. {% content-ref url="/pages/pghfGWdoAmKqeFiwqKtz" %} [Queue Connection](/2.0.0/configuration/config-file/queue-connection) {% endcontent-ref %} #### delete {% hint style="danger" %} This method is not yet implemented. {% endhint %} ### withWorkerPool Retrieves an already defined Worker Pool Definition for overriding. All the same methods are available. {% content-ref url="/pages/FRcimvpoy6mO5qGgTvgb" %} [Worker Pool](/2.0.0/configuration/config-file/worker-pool) {% endcontent-ref %} #### delete {% hint style="danger" %} This method is not yet implemented. {% endhint %} # Queue Connection A Queue Connection is defined inside your cbq config file using a `QueueConnectionDefinition` builder component. You can create one of these builder components using the [`newConnection`](/2.0.0/configuration/config-file#newconnection) method. ### QueueConnectionDefinition Methods #### provider Sets the provider for the Queue Connection. This can be any valid WireBox mapping and should implement the `IQueueProvider` interface. (There is no need to use the `implements` keyword.)
ArgumentsTypeRequiredDefaultDescription
providerstringtrueA valid WireBox mapping to the desired Provider.
#### setProvider Alias for [`provider`](#provider). #### setProperties Accepts a struct of properties to configure the Queue Connection. The available properties are usually defined by the Queue Provider being used. #### setDefaultQueue Sets the default queue to use for jobs dispatched on this Queue Connection. #### markAsDefault Marks this Queue Connection as the default Queue Connection for jobs dispatched without specifying a Queue Connection. #### setMakeDefault Alias for [`markAsDefault`](#markasdefault). # Worker Pool A Queue Connection is defined inside your cbq config file using a `QueueConnectionDefinition` builder component. You can create one of these builder components using the [`newConnection`](/2.0.0/configuration/config-file#newconnection) method. ### WorkerPoolDefinition Methods #### setName Sets the name for the Worker Pool. Usually not called directly as the `newWorkerPool` method requires a `name`. ```cfscript newWorkerPool( "default" ) .setName( "not-default" ); ``` #### forConnection Sets the name of the associated Connection for the Worker Pool. This must reference an already registered Connection. ```cfscript newConnection( "db" ) .provider( "DBProvider@cbq" ); newWorkerPool( "db-worker" ) .forConnection( "db" ); ``` #### setConnectionName Alias for [forConnection](#forconnection). #### quantity Sets the quantity of workers for this Worker Pool. ```cfscript newWorkerPool( "default" ) .forConnection( "db" ) .quantity( 3 ); ``` #### setQuantity Alias for [quantity](#quantity). #### onQueue The name of a queue to work. The default queue is named `default`. ```cfscript newWorkerPool( "premium-only" ) .forConnection( "db" ) .onQueue( "premium" ); newWorkerPool( "priority" ) .forConnection( "db" ) .onQueue( "priority" ) .quantity( 4 ); newWorkerPool( "default" ) .forConnection( "db" ); // uses `default` queue ``` #### setQueue Alias for [onQueue](#onqueues). #### backoff Sets the backoff time amount, in seconds. ```cfscript newWorkerPool( "default" ) .forConnection( "db" ) .backoff( 30 ); ``` #### setBackoff Alias for [backoff](#backoff). #### timeout Sets the timeout time amount, in seconds. ```cfscript newWorkerPool( "default" ) .forConnection( "db" ) .timeout( 60 ); ``` #### setTimeout Alias for [timeout](#timeout). #### maxAttempts Sets the max number of attempts. ```cfscript newWorkerPool( "default" ) .forConnection( "db" ) .maxAttempts( 5 ); ``` #### setMaxAttempts Alias for [maxAttempts](#maxattempts). # Providers A Queue Provider provides a way to serialize jobs to a Queue Connection and to work jobs off a Queue Connection. The three built-in cbq providers are: * [SyncProvider@cbq](/2.0.0/configuration/providers/syncprovider) * [ColdBoxAsyncProvider@cbq](/2.0.0/configuration/providers/coldboxasyncprovider) * [DBProvider@cbq](/2.0.0/configuration/providers/dbprovider) ### Writing your own Custom Provider All Queue Providers extend the `AbstractQueueProvider` base component. They must implement three abstract methods: ```cfscript /** * Persists a serialized job to the Queue Connection * * @queueName The queue name for the job. * @payload The serialized job string. * @delay The delay (in seconds) before working the job. * @attempts The current attempt number. * * @return AbstractQueueProvider */ public any function push( required string queueName, required string payload, numeric delay = 0, numeric attempts = 0 ); /** * Starts a worker for a Worker Pool on this Queue Connection. * * @pool The Worker Pool that is working this Queue Connection. * * @return A function that when called will stop this worker. */ public function function startWorker( required WorkerPool pool ); /** * Starts any background processes needed for the Worker Pool. * * @pool The Worker Pool that is working this Queue Connection. * * @return AbstractQueueProvider */ public any function listen( required WorkerPool pool ); ``` # SyncProvider {% hint style="danger" %} This provider is not **durable**. (Pending jobs are not saved and may be lost if there are server issues between dispatching the job and working the job.) {% endhint %} The SyncProvider runs any jobs dispatched during a request in the same request synchronously. This can be especially useful in development when debugging jobs. If you job would throw an exception, you will see it with any chosen error handler and tools you use to debug local exceptions. # ColdBoxAsyncProvider {% hint style="danger" %} This provider is not **durable**. (Pending jobs are not persisted and may be lost if there are server issues between dispatching the job and working the job.) {% endhint %} The ColdBoxAsyncProvider runs any jobs dispatched on a background thread using ColdBox's AsyncManager. Jobs will be worked by the same server that dispatched them. The number of workers specified translates to the number of threads dedicated to working the jobs. # DBProvider {% hint style="success" %} This provider is **durable**. (Pending jobs are persisted. If there are server issues, the jobs will be worked when the issues are resolved. Multiple different servers can dispatch to this Queue Connection and multiple different Worker Pools can work these jobs.) {% endhint %} You may use a database as the backing engine for your Queue Connection using the DBProvider. All database grammars that are [supported by qb](https://qb.ortusbooks.com/v/9.0.0/installation-and-usage) are supported. ### Configuration The DBProvider has three optional arguments. The are presented below with their default values. ```cfscript { "tableName": "cbq_jobs", "datasource": null, "queryOptions": {} } ``` #### tableName The name of the table to use when managing jobs. This table should have the structure provided by the provided database migration file. #### datasource The name of the datasource to use when managing jobs. This overrides any datasource provided in the `queryOptions`. #### queryOptions A struct of options that will be passed to [`queryExecute`](https://cfdocs.org/queryexecute) when managing jobs. ### Jobs Table Structure The `cbq_jobs` table must have a specific structure. It is provided in the form of a migration file called `2000_01_01_000000_create_cbq_jobs_table.cfc`. This migration can be ran via [CommandBox Migrations](https://forgebox.io/view/commandbox-migrations) or [CFMigrations](https://forgebox.io/view/cfmigrations). If you wish, you can generate the table in other ways, so long as it matches the structure provided in the migration file. If you choose to use the migration file in your application, copy the file out to your own migrations folder first. ```cfscript schema.create( "cbq_jobs", function ( t ) { t.bigIncrements( "id" ); t.string( "queue" ); t.longText( "payload" ); t.unsignedTinyInteger( "attempts" ); t.unsignedInteger( "reservedDate" ).nullable(); t.unsignedInteger( "availableDate" ); t.unsignedInteger( "createdDate" ); t.index( "queue" ); } ); ``` # Defining a Job A Job represents both the object to be serialized to a [Connection](/2.0.0/configuration/config-file#connection) to eventually work as well as the code to execute when working a Job. ## Extending from AbstractJob The first step to define a Job is to extend from `AbstractJob`. This provides the necessary helper methods to run a Job through the cbq pipeline. ```cfscript component name="SendWelcomeEmailJob" extends="cbq.models.Jobs.AbstractJob" { } ``` ## Properties Defining properties on your job allow you to see at a glance what data your Job expects when constructing it. These properties and their values will be serialized to a [Connection](/2.0.0/configuration/config-file#connection) when dispatching a Job. {% hint style="info" %} Defining the properties is not strictly necessary, but your future self will thank you when you try to remember what properties your job is using. {% endhint %} ```cfscript component name="SendWelcomeEmailJob" extends="cbq.models.Jobs.AbstractJob" { property name="email"; property name="greeting"; } ``` ## The handle method The `handle` method is called when a Job is worked. Before being called, a Job will be reconstructed with the serialized data from the Connection. This method is the only required method when defining a Job. ```cfscript component name="SendWelcomeEmailJob" extends="cbq.models.Jobs.AbstractJob" { property name="mailService" inject="provider:MailService@cbmailservices"; property name="email"; property name="greeting"; function handle() { variables.mailService.newMail( to = variables.email, from = "noreply@example.com", subject = "Welcome!", type = "html", bodyTokens = { "greeting": variables.greeting "link": getInstance( "coldbox:requestContext" ) .buildLink( "home" ) } ) .setView( "_emails/welcome" ) .send(); } } ``` {% hint style="info" %} Prefer using `provider:` injections or inline `getInstance` calls for logic in your `handle` method. The `Job` component is created both when dispatching and when working your job, so utilizing these tools will reduce unnecessary processing time when dispatching your job. {% endhint %} ## Job Execution Properties A job can define several execution properties on the job itself. If defined, these values override the module, connection, or worker defaults. It can still be overridden using the job methods when creating a job. ```cfscript component extends="cbq.models.Jobs.AbstractJob" { // Name of the connection to dispatch this job on. variables.connection = "db"; // Name of the queue to use for this job. variables.queue = "priority"; // Time, in seconds, between job attempts, including the initial attempt. variables.backoff = 15; // Time, in seconds, to let a job run before failing it. variables.timeout = 30; // Max number of attempts before marking a job as failed. // Use 0 for unlimited retries. variables.maxAttempts = 9; } ``` # Creating a Job After you define your job, you need to create an instance of your job and set the properties before you dispatch it onto your [Queue Connection](/2.0.0/configuration/config-file#connection). You can do this is a few different ways. ## Creating a Job Instance You can create a Job instance using WireBox anywhere in your ColdBox application. ```cfscript component { function index( event, rc, prc ) { var job = getInstance( "SendWelcomeEmailJob" ); } } ``` {% hint style="warning" %} Keep in mind that Job instances are transient. Do not inject a Job instance into a Component or Scope that is not transient. For instance, do not inject a Job instance as a property in a Handler. {% endhint %} ### Setting Job Properties Once you have a Job instance, you can set the data for the particular Job by calling the setter methods. ```cfscript component { function index( event, rc, prc ) { var job = getInstance( "SendWelcomeEmailJob" ); job.setEmail( "john@example.com" ); job.setGreeting( "Welcome!" ); } } ``` ### setProperties You can also set all the properties at once using the `setProperties` method. {% hint style="warning" %} Using this method will overwrite any previously set properties. {% endhint %} ```cfscript component { function index( event, rc, prc ) { var job = getInstance( "SendWelcomeEmailJob" ); job.setProperties( { "email": "john@example.com", "greeting": "Welcome!" } ); } } ``` ### onConnection You can override the connection for a job by calling the `onConnection` method with the desired connection name. ### setConnection See [onConnection](#onconnection). ### onQueue The queue name to use for the job. Queue names can be any string you choose, but to be worked a `WorkerPool` must be defined working the same connection and queue name. ### setQueue See [onQueue](#onqueue). ### setBackoff Sets the amount of time, in seconds, to wait in-between Job attempts — including the initial attempt. ### setDelay An alias for [setBackoff](#setbackoff). ### setTimeout Sets the amount of time, in seconds, to let a Job run on a worker before marking it as a failed attempt. ### setMaxAttempts Sets the maximum number of attempts before a Job is marked as failed. ### chain Sets an array of Jobs to be executed, in order, after this Job successfully executes. ### getMemento getMemento ## [cbq.job()](/2.0.0/cbq-model#job) You can also create a Job instance using the `cbq` model. (This can be injected into your components using the `cbq@cbq` DSL.) {% hint style="success" %} The cbq model is a singleton, so feel free to inject it into any component. {% endhint %} ```cfscript component { property name="cbq" inject="cbq@cbq"; function index( event, rc, prc ) { var job = cbq.job( job = "SendWelcomeEmailJob", properties = { "email": "john@example.com", "greeting": "Welcome!" }, chain = [], queue = "priority", connection = "db", backoff = 10, timeout = 30, maxAttempts = 3 ); } } ``` # Dispatching a Job ## Dispatching Creating a Job is all well and good, but the Job will not be executed until it is dispatched on to a [Queue Connection](/2.0.0/configuration/config-file#connection). If you have a Job instance, you can dispatch it by calling the `dispatch` method. ```cfscript component { function index( event, rc, prc ) { getInstance( "SendWelcomeEmailJob" ); .setEmail( "john@example.com" ); .setGreeting( "Welcome!" ) .dispatch(); } } ``` You can also create and dispatch a job in the same call using the `cbq.dispatch()` method. ```cfscript component { property name="cbq" inject="cbq@cbq"; function index( event, rc, prc ) { cbq.dispatch( job = "SendWelcomeEmailJob", properties = { "email": "john@example.com", "greeting": "Welcome!" }, chain = [], queue = "priority", connection = "db", backoff = 10, timeout = 30, maxAttempts = 3 ); } } ``` ## Interception Points ### onCBQJobAdded This is fired before serializing a Job and sending it to a connection. The data includes the `job` to be serialized and dispatched and the connection the Job is being dispatched on. ```cfscript variables.interceptorService.announce( "onCBQJobAdded", { "job" : job, "connection" : connection } ); ``` # Working a Job Dispatched jobs will get picked up by Worker Pools set up for the same connection and queue. They are picked up in order by created date (FIFO). If you have a valid Worker Pool, then you don't need to do anything else to have your job picked up. If you want to see all the details of cbq dispatching and marshalling jobs, enable `debug` logging in LogBox for `cbq`. When a Job is worked, the Job instance is created via WireBox and the memento hydrated into the Job component. Then the `handle` method is called. When a Job is completed, the Job will be removed from the persistent storage of the Queue Provider. If the Job attempt fails, it will be sent back to the Queue Provider to be retried, up to the `maxAttempts` amount. Once a Job has failed the `maxAttempts` amount it will be removed and marked as failed. If you have `logFailedJobs` enabled, it will be sent to the [failed jobs table](/2.0.0/jobs/failed-jobs#logging-failed-jobs). ## Inside \`handle\` ### release You can choose to manually release a Job back to a queue with an optional delay (in seconds) using the `release` function. This is useful when you need to delay processing for a Job, perhaps due to rate limiting or licensing constraints. {% hint style="info" %} Calling `release` does not stop processing the `handle` method, so make sure you `return` if you don't want to keep executing your `handle` method. {% endhint %} ```cfscript component { function handle() { this.release( 60 ); // in seconds return; } } ``` ## Interception Points cbq fires a number of interception points during a Job's lifecycle. ### onCBQJobMarshalled This is fired before the `handle` method is called on a Job. The data includes the `job` to be executed. ```cfscript variables.interceptorService.announce( "onCBQJobMarshalled", { "job" : job } ); ``` ### onCBQJobComplete This is fired after a Job completes successfully. The data includes the `job` to be executed as potentially a `result` returned from the `Job`'s handle method. ```cfscript variables.interceptorService.announce( "onCBQJobComplete", { "job" : job, "result" : isNull( result ) ? javacast( "null", "" ) : result } ); ``` ## Tips ### Need up to date data? Could your data change between when you dispatched a job and when you work the job? If so, consider using the key as a property and fetching the data from inside the job. ### Need to use historical data? In some cases, you want to work with the data that existed at the time the Job was dispatched. In these cases, make sure to include all the needed data in the Job instance. ### Check if you need to work the Job still Things might have changed since the Job was dispatched. Most Queue Providers do not support querying the current Jobs queue or interacting with it in advance, so check in your `dispatch` method if the Job still needs to be worked. ### Need to try the Job later? In some cases, you need to retry your Job later. In these cases, use the `release` method to send the Job back to the queue. It also takes an optional `delay` which sets the `backoff` time for the next Job attempt. You can combine this with setting the `maxAttempts` of a Job to `0` to retry it indefinitely. Remember that a Job is only reattempted if it fails. If you decide that a Job is finished, just `return`. ### Jobs can dispatch other Jobs You can dispatch other Jobs from your Job. This can be different Jobs or even a similar instance of the same Job. ## Why isn't my Job getting picked up? To work a cbq job, you need a [defined Worker Pool in your cbq Config file.](/2.0.0/configuration/config-file#newworkerpool) In order to work a Job, the Worker Pool must: 1. Be working the same [connection](/2.0.0/configuration/config-file/worker-pool#forconnection) as the Job. 2. Be working the same [queue](/2.0.0/configuration/config-file/worker-pool#onqueues) as the Job. 3. Have at least one active worker (defined with [`setQuantity`](/2.0.0/configuration/config-file/worker-pool#setquantity)). If these requirements are not met, the job will stay in the queue storage, unable to get picked up. # Failed Jobs ## Failed Attempt vs Failed Job An important distinction to make is between failed Job attempts and failed Jobs. A Job is attempted up to the defined `maxAttempts`. If a Job fails all of its `maxAttempts` then it is marked as a failed Job and removed from the queue. Otherwise, the Job is dispatched back to the queue with the current execution count increased. ## LogBox Failed Job attempts are logged to LogBox. You will see the exception as well as the serialized Job memento in the `extraInfo`. ## Interceptors ### onCBQJobException This interception point is fired for every **Job Attempt** failure. The data includes the `job` component and the `exception`. ```cfscript variables.interceptorService.announce( "onCBQJobException", { "job" : job, "exception" : e } ); ``` ### onCBQJobFailed This interception point is fired when the Job is marked as failed — when the Job has failed all attempts up to the configured `maxAttempts`. The data includes the `job` component and the `exception`. ```cfscript variables.interceptorService.announce( "onCBQJobFailed", { "job" : job, "exception" : e } ); ``` ## onFailure Job Method You can define an `onFailure` method on your Job component. It will be called with the `exception`. ```cfscript component extends="cbq.models.Jobs.AbstractJob" { function handle() { // ... } function onFailure( e ) { log.error( "Bad things happened: #e.message#" ); } } ``` ## Logging Failed Jobs ## Failed Jobs Table cbq includes an interceptor to log failed jobs to a database. You can enable this in your module settings: ```cfscript moduleSettings = { "cbq": { // Flag to turn on logging failed jobs to a database table. "logFailedJobs" : false, // Datasource information for loggin failed jobs. "logFailedJobsProperties" : { "tableName" : "cbq_failed_jobs", "datasource" : "", // `datasource` can also be a struct. "queryOptions" : {} // The sibling `datasource` property overrides any defined datasource in `queryOptions`. } } }; ``` Also included is a migration file to add the needed table to your database. You can find it in `resources/database/migrations/2000_01_01_000002_create_cbq_failed_jobs_table.cfc`. It is also included below. ```cfscript component { function up( schema ) { schema.create( "cbq_failed_jobs", function ( t ) { t.bigIncrements( "id" ); t.string( "connection" ); t.string( "queue" ); t.string( "mapping" ); t.longText( "memento" ); t.longText( "properties" ); t.string( "exceptionType" ).nullable(); t.string( "exceptionMessage" ); t.string( "exceptionDetail" ).nullable(); t.longText( "exceptionExtendedInfo" ).nullable(); t.longText( "exceptionStackTrace" ); t.longText( "exception" ); t.timestamp( "failedDate" ).withCurrent(); } ); } function down( schema ) { schema.dropIfExists( "cbq_failed_jobs" ); } } ``` {% code title="MySQL" %} ```sql CREATE TABLE `cbq_failed_jobs` ( `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT, `connection` NVARCHAR(255) NOT NULL, `queue` NVARCHAR(255) NOT NULL, `mapping` NVARCHAR(255) NOT NULL, `memento` LONGTEXT NOT NULL, `properties` LONGTEXT NOT NULL, `exceptionType` NVARCHAR(255), `exceptionMessage` NVARCHAR(255) NOT NULL, `exceptionDetail` NVARCHAR(255), `exceptionExtendedInfo` LONGTEXT, `exceptionStackTrace` LONGTEXT NOT NULL, `exception` LONGTEXT NOT NULL, `failedDate` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ) ``` {% endcode %} {% code title="SQL Server" %} ```sql CREATE TABLE [dbo].[cbq_failed_jobs]( [id] BIGINT NOT NULL IDENTITY, [connection] NVARCHAR(255) NOT NULL, [queue] NVARCHAR(255) NOT NULL, [mapping] NVARCHAR(255) NOT NULL, [memento] NVARCHAR(MAX) NOT NULL, [properties] NVARCHAR(MAX) NOT NULL, [exceptionType] NVARCHAR(255), [exceptionMessage] NVARCHAR(255) NOT NULL, [exceptionDetail] NVARCHAR(255), [exceptionExtendedInfo] NVARCHAR(MAX), [exceptionStackTrace] NVARCHAR(MAX) NOT NULL, [exception] NVARCHAR(MAX) NOT NULL, [failedDate] DATETIME2 NOT NULL CONSTRAINT [df_cbq_failed_jobs_failedDate] DEFAULT CURRENT_TIMESTAMP ) ``` {% endcode %} ## Retrying a Failed Job ## Configuring Job Backoff # Chained Jobs Job chains mean that after a Job completes successfully it dispatches the next Job in the chain. If a Job fails, no more Jobs in the chain are dispatched. ## Dispatching Jobs from another Job One way to make a Job chain is to dispatch a Job from inside another Job. This gives you consistency in your Job chain, and a logical code path to follow. ```cfscript // FulfillOrderJob.cfc component extends="cbq.models.Jobs.AbstractJob" { function handle() { var productId = getProperties().productId; processPayment( productId ); getInstance( "SendProductLinkEmail" ) .onConnection( "fulfillment" ) .setProperties( { "productId" : productId, "userId" : getProperties().userId } ) .dispatch(); } } ``` ## Creating a Chain Another way to make a Job chain is to create is when dispatching the Job. This gives you flexibility to compose Job chains at runtime. ```cfscript // handlers/Main.cfc component { property name="cbq" inject="cbq@cbq"; function create() { cbq.chain( [ cbq.job( "FulfillOrderJob", { "productId": rc.productId } ), cbq.job( job = "SendProductLinkEmail", properties = { "productId": rc.productId, "userId": auth().getUserId() }, connection = "fulfillment" ), ] ).dispatch(); } } ``` ## Dispatching Jobs on Failure Utilizing the `onFailure` method of a Job, we can dispatch another Job if our Job fails. ```cfscript // FulfillOrderJob.cfc component extends="cbq.models.Jobs.AbstractJob" { function handle() { // ... } function onFailure( e ) { getInstance( "SendOrderProblemEmail" ) .onConnection( "fulfillment" ) .setProperties( { "productId" : productId, "userId" : getProperties().userId, "message" : e.message } ) .dispatch(); } } ``` # Batched Jobs cbq allows for tracking jobs as part of a batch. The main use case for batched jobs is to dispatch additional jobs when a batch has completed successfully, completed with failures, or completed in any fashion. (Think `try`, `catch`, `finally`.) ## Setup To utilize batches, you first need to set up a Batch repository. Batches are tracked separate from jobs. cbq utilizes a database repository to track batches regardless of what provider your Job's connection uses. A database migration is provided in `resources/database/migrations/2000_01_01_000001_create_cbq_batches_table.cfc`. You can copy this to your own project to run via [cfmigrations](https://forgebox.io/view/cfmigrations) or you can reference the migration or SQL scripts [below](#migrations-and-sql-scripts). You can customize the batch repository using the `batchRepositoryProperties` of cbq's `moduleSettings`. This is a struct with two properties you can set: #### tableName This defaults to `cbq_batches`. You can set this to any unique table name in your datasource. #### queryOptions This is a struct of options that will be passed to `queryExecute`. The most common use case for this property is to specify a specific datasource to use for the batches table. This defaults to an empty struct (`{}`). ## Defining Batches To create a batch, you can get an instance of `PendingBatch@cbq` from WireBox or use the [`cbq.batch()`](/2.0.0/cbq-model#batch) helper method. Once you have a `PendingBatch` instance, you can add jobs to the batch using the `add` method. ### add Adds a single Job or an array of Jobs to a PendingBatch.
ArgumentsTypeRequiredDefaultDescription
jobstring | Job | array<Job>trueThe Job WireBox id, Job instance, or array of Job instances to add to the PendingBatch.
propertiesStructfalse{}A struct of properties for the Job instance. (Only used when providing a Job WireBox id.)
chainarray<Job>false[]An array of Jobs to chain after this Job. (Only used when providing a Job WireBox id.)
queuestringfalseThe queue the Job belongs to. (Only used when providing a Job WireBox id.)
connectionstringfalseThe Connection to dispatch the Job on. (Only used when providing a Job WireBox id.)
backoffnumericfalseThe amount of time, in seconds, to wait between attempting Jobs. (Only used when providing a Job WireBox id.)
timeoutnumericfalseThe amount of time, in seconds, to wait before treating a Job as erroring. (Only used when providing a Job WireBox id.)
maxAttemptsnumericfalseThe maximum amount of attempts of a Job before treating the Job as failed. (Only used when providing a Job WireBox id.)
**Return:** The PendingBatch to be dispatched. ```cfscript batch .add( cbq.job( "ImportCsvJob", { "start": 1, "end": 100 } ) ) .add( "ImportCsvJob", { "start": 101, "end": 200 } ) .add( [ cbq.job( "ImportCsvJob", { "start": 201, "end": 300 } ), cbq.job( "ImportCsvJob", { "start": 301, "end": 400 } ), cbq.job( "ImportCsvJob", { "start": 401, "end": 500 } ) ] ); ``` The primary use case of batches is to dispatch lifecycle jobs when the batch has completed and if the batch completed successfully or completed with failures. These jobs can be configured using the `then`, `catch`, and `finally` methods. ### then Defines a Job to be dispatched when all the jobs in the batch finishes successfully.
ArgumentsTypeRequiredDefaultDescription
jobstring | JobtrueThe Job WireBox id or Job instance to execute if all the jobs in the Batch complete successfully.
propertiesStructfalse{}A struct of properties for the Job instance. (Only used when providing a Job WireBox id.)
chainarray<Job>false[]An array of Jobs to chain after this Job. (Only used when providing a Job WireBox id.)
queuestringfalseThe queue the Job belongs to. (Only used when providing a Job WireBox id.)
connectionstringfalseThe Connection to dispatch the Job on. (Only used when providing a Job WireBox id.)
backoffnumericfalseThe amount of time, in seconds, to wait between attempting Jobs. (Only used when providing a Job WireBox id.)
timeoutnumericfalseThe amount of time, in seconds, to wait before treating a Job as erroring. (Only used when providing a Job WireBox id.)
maxAttemptsnumericfalseThe maximum amount of attempts of a Job before treating the Job as failed. (Only used when providing a Job WireBox id.)
**Return:** The `PendingBatch` instance. ```cfscript cbq.batch( [ cbq.job( "ImportCsvJob", { "start": 1, "end": 100 } ), cbq.job( "ImportCsvJob", { "start": 101, "end": 200 } ), cbq.job( "ImportCsvJob", { "start": 201, "end": 300 } ), cbq.job( "ImportCsvJob", { "start": 301, "end": 400 } ), cbq.job( "ImportCsvJob", { "start": 401, "end": 500 } ) ] ) .then( cbq.job( "ImportCsvSuccessfulJob" ) ); ``` ### catch Defines a Job to be dispatched the first time a job in the Batch fails.
ArgumentsTypeRequiredDefaultDescription
jobstring | JobtrueThe Job WireBox id or Job instance to execute the first time a job in the Batch fails.
propertiesStructfalse{}A struct of properties for the Job instance. (Only used when providing a Job WireBox id.)
chainarray<Job>false[]An array of Jobs to chain after this Job. (Only used when providing a Job WireBox id.)
queuestringfalseThe queue the Job belongs to. (Only used when providing a Job WireBox id.)
connectionstringfalseThe Connection to dispatch the Job on. (Only used when providing a Job WireBox id.)
backoffnumericfalseThe amount of time, in seconds, to wait between attempting Jobs. (Only used when providing a Job WireBox id.)
timeoutnumericfalseThe amount of time, in seconds, to wait before treating a Job as erroring. (Only used when providing a Job WireBox id.)
maxAttemptsnumericfalseThe maximum amount of attempts of a Job before treating the Job as failed. (Only used when providing a Job WireBox id.)
**Return:** The `PendingBatch` instance. ```cfscript cbq.batch( [ cbq.job( "ImportCsvJob", { "start": 1, "end": 100 } ), cbq.job( "ImportCsvJob", { "start": 101, "end": 200 } ), cbq.job( "ImportCsvJob", { "start": 201, "end": 300 } ), cbq.job( "ImportCsvJob", { "start": 301, "end": 400 } ), cbq.job( "ImportCsvJob", { "start": 401, "end": 500 } ) ] ) .catch( cbq.job( "ImportCsvFailedJob" ) ); ``` ### finally Defines a Job to be dispatched after all the jobs in the Batch have executed successfully or failed.
ArgumentsTypeRequiredDefaultDescription
jobstring | JobtrueThe Job WireBox id or Job instance to execute after all the jobs in the Batch have executed successfully or failed.
propertiesStructfalse{}A struct of properties for the Job instance. (Only used when providing a Job WireBox id.)
chainarray<Job>false[]An array of Jobs to chain after this Job. (Only used when providing a Job WireBox id.)
queuestringfalseThe queue the Job belongs to. (Only used when providing a Job WireBox id.)
connectionstringfalseThe Connection to dispatch the Job on. (Only used when providing a Job WireBox id.)
backoffnumericfalseThe amount of time, in seconds, to wait between attempting Jobs. (Only used when providing a Job WireBox id.)
timeoutnumericfalseThe amount of time, in seconds, to wait before treating a Job as erroring. (Only used when providing a Job WireBox id.)
maxAttemptsnumericfalseThe maximum amount of attempts of a Job before treating the Job as failed. (Only used when providing a Job WireBox id.)
**Return:** The `PendingBatch` instance. ```cfscript cbq.batch( [ cbq.job( "ImportCsvJob", { "start": 1, "end": 100 } ), cbq.job( "ImportCsvJob", { "start": 101, "end": 200 } ), cbq.job( "ImportCsvJob", { "start": 201, "end": 300 } ), cbq.job( "ImportCsvJob", { "start": 301, "end": 400 } ), cbq.job( "ImportCsvJob", { "start": 401, "end": 500 } ) ] ) .finally( cbq.job( "ImportCsvCompletedJob" ) ); ``` ## Dispatching Batches A `PendingBatch` must be dispatched before any of the jobs contained in it are dispatched. This is done using the `dispatch` method on the `PendingBatch`. ### dispatch Dispatches a PendingBatch and all the Jobs it contains.
ArgumentsTypeRequiredDefaultDescription
No arguments
**Return:** A `Batch` instance created from this `PendingBatch`. ## Interacting with Batches ## Migrations and SQL Scripts ### cfmigrations ```cfscript schema.create( "cbq_batches", function ( t ) { t.string( "id" ).primaryKey(); t.string( "name" ); t.unsignedInteger( "totalJobs" ); t.unsignedInteger( "pendingJobs" ); t.unsignedInteger( "failedJobs" ); t.text( "failedJobIds" ); // JSON column t.text( "options" ).nullable(); // JSON column t.datetime( "createdDate" ); t.datetime( "cancelledDate" ).nullable(); t.datetime( "completedDate" ).nullable(); } ); ``` ### MySQL ```sql CREATE TABLE `cbq_batches` ( `id` VARCHAR(255) NOT NULL, `name` VARCHAR(255) NOT NULL, `totalJobs` INTEGER UNSIGNED NOT NULL, `pendingJobs` INTEGER UNSIGNED NOT NULL, `failedJobs` INTEGER UNSIGNED NOT NULL, `failedJobIds` TEXT NOT NULL, `options` TEXT, `createdDate` DATETIME NOT NULL, `cancelledDate` DATETIME, `completedDate` DATETIME, CONSTRAINT `pk_cbq_batches_id` PRIMARY KEY (`id`) ) ``` ### SQL Server ```sql CREATE TABLE [cbq_batches] ( [id] VARCHAR(255) NOT NULL, [name] VARCHAR(255) NOT NULL, [totalJobs] INTEGER NOT NULL, [pendingJobs] INTEGER NOT NULL, [failedJobs] INTEGER NOT NULL, [failedJobIds] VARCHAR(MAX) NOT NULL, [options] VARCHAR(MAX), [createdDate] DATETIME2 NOT NULL, [cancelledDate] DATETIME2, [completedDate] DATETIME2, CONSTRAINT [pk_cbq_batches_id] PRIMARY KEY ([id]) ) ``` ### Postgres ```sql CREATE TABLE "cbq_batches" ( "id" VARCHAR(255) NOT NULL, "name" VARCHAR(255) NOT NULL, "totalJobs" INTEGER NOT NULL, "pendingJobs" INTEGER NOT NULL, "failedJobs" INTEGER NOT NULL, "failedJobIds" TEXT NOT NULL, "options" TEXT, "createdDate" TIMESTAMP NOT NULL, "cancelledDate" TIMESTAMP, "completedDate" TIMESTAMP, CONSTRAINT "pk_cbq_batches_id" PRIMARY KEY ("id") ) ``` ### Oracle ```sql CREATE TABLE "CBQ_BATCHES" ( "ID" VARCHAR2(255) NOT NULL, "NAME" VARCHAR2(255) NOT NULL, "TOTALJOBS" NUMBER(10, 0) NOT NULL, "PENDINGJOBS" NUMBER(10, 0) NOT NULL, "FAILEDJOBS" NUMBER(10, 0) NOT NULL, "FAILEDJOBIDS" CLOB NOT NULL, "OPTIONS" CLOB, "CREATEDDATE" DATE NOT NULL, "CANCELLEDDATE" DATE, "COMPLETEDDATE" DATE, CONSTRAINT "PK_CBQ_BATCHES_ID" PRIMARY KEY ("ID") ) ``` ### SQLite ```sql CREATE TABLE "cbq_batches" ( "id" TEXT NOT NULL, "name" TEXT NOT NULL, "totalJobs" INTEGER NOT NULL, "pendingJobs" INTEGER NOT NULL, "failedJobs" INTEGER NOT NULL, "failedJobIds" TEXT NOT NULL, "options" TEXT, "createdDate" DATETIME NOT NULL, "cancelledDate" DATETIME, "completedDate" DATETIME, PRIMARY KEY ("id") ) ``` # cbq Model This model is provided to make certain tasks easier when defining and dispatching jobs, chains, and batches. ### dispatch Dispatches a job or chain of jobs.
ArgumentsTypeRequiredDefaultDescription
jobstring | Job | array<Job>trueA Job instance or a WireBox ID of a Job instance. If the WireBox ID doesn't exist, a NonExecutableJob instance will be used instead. This allows you to dispatch a Job from a server where that Job component is not defined. If an array is passed, a Job Chain is created and dispatched.
propertiesstructfalse{}A struct of properties for the Job instance.
chainarray<Job>false[]An array of Jobs to chain after this Job.
queuestringfalseThe queue the Job belongs to.
connectionstringfalseThe Connection to dispatch the Job on.
backoffnumericfalseThe amount of time, in seconds, to wait between attempting Jobs.
timeoutnumericfalseThe amount of time, in seconds, to wait before treating a Job as erroring.
maxAttemptsnumericfalseThe maximum amount of attempts of a Job before treating the Job as failed.
**Return:** The dispatched Job instance. ```cfscript cbq.dispatch( job = "SendWelcomeEmailJob", properties = { "body": "first body" }, queue = "default" ); ``` ### job Creates a job or chain of jobs to be dispatched.
ArgumentsTypeRequiredDefaultDescription
jobstring | Job | array<Job>trueA Job instance or a WireBox ID of a Job instance. If the WireBox ID doesn't exist, a NonExecutableJob instance will be used instead. This allows you to dispatch a Job from a server where that Job component is not defined. If an array is passed, a Job Chain is created and dispatched.
propertiesstructfalse{}A struct of properties for the Job instance.
chainarray<Job>false[]An array of Jobs to chain after this Job.
queuestringfalseThe queue the Job belongs to.
connectionstringfalseThe Connection to dispatch the Job on.
backoffnumericfalseThe amount of time, in seconds, to wait between attempting Jobs.
timeoutnumericfalseThe amount of time, in seconds, to wait before treating a Job as erroring.
maxAttemptsnumericfalseThe maximum amount of attempts of a Job before treating the Job as failed.
**Return:** The new Job instance. ```cfscript cbq.job( "SendWelcomeEmailJob" ) .setProperties( { "body": "first body" } ) .onQueue( "default" ) .dispatch(); ``` ### chain Creates a chain of jobs to be ran. Alias for calling `firstJob.chain( otherJobs )`.
ArgumentsTypeRequiredDefaultDescription
jobsarray<Job>trueThe array of jobs to run in order in a chain.
**Return:** The first job of the chain with the chained jobs configured to be dispatched. ```cfscript cbq.chain( [ cbq.job( "SendWelcomeEmailJob", { "body": "One" }, [], "default" ), cbq.job( "SendWelcomeEmailJob", { "body": "Two" }, [], "default", "sync" ), cbq.job( "SendWelcomeEmailJob", { "body": "Three" }, [], "default" ) ] ) .dispatch() ``` ### batch Creates a PendingBatch from the Jobs provided. {% hint style="warning" %} To use batches, you must first configure a `BatchRepository`. Learn more in the [Batched Jobs documentation](/2.0.0/jobs/batched-jobs). {% endhint %}
ArgumentsTypeRequiredDefaultDescription
jobsarray<Job>trueAn array of jobs to batch together.
**Return:** The PendingBatch to be dispatched. ```cfscript var batch = cbq .batch( [ cbq.job( "ImportCsvJob", { "start": 1, "end": 100 } ), cbq.job( "ImportCsvJob", { "start": 101, "end": 200 } ), cbq.job( "ImportCsvJob", { "start": 201, "end": 300 } ), cbq.job( "ImportCsvJob", { "start": 301, "end": 400 } ), cbq.job( "ImportCsvJob", { "start": 401, "end": 500 } ) ] ) .then( cbq.job( "ImportCsvSuccessfulJob" ) ) .catch( cbq.job( "ImportCsvFailedJob" ) ) .finally( cbq.job( "ImportCsvCompletedJob" ) ) .dispatch(); ``` # Contributing # Contributors # Prior Art Initial ideas behind cbq were derived from [Laravel](https://laravel.com). # Dedication # Home ## A protocol-based queueing system for ColdBox Queues allow you to push work to the background, schedule work to be done later, or even process work on many different machines. It runs on a provider-based system allowing a unified API to talk with many different queue backends. ### Where to go next? * [Installation](/2.1.0/getting-started/installation) * [Walkthrough](/2.1.0/getting-started/walkthrough) * CFCasts Series (Coming Soon) * API Docs # What's New? ## v2.1.0 Add back ability to [work on multiple queues](/2.1.0/configuration/config-file/worker-pool#onqueue) on a per-Provider basis. Currently only the `DBProvider` supports it. Add support for `before` and `after` [lifecycle methods](/2.1.0/jobs/defining-a-job#lifecycle-methods) on a Job instance. Add ability to [restrict interceptor execution ](/2.1.0/interceptors#jobpattern-annotation)with a `jobPattern` annotation. (This is similar to the `eventPattern` annotation [provided by ColdBox](https://coldbox.ortusbooks.com/the-basics/interceptors/restricting-execution).) ## v2.0.5 **DBProvider**: Disable `forceRun` because it is causing ColdBox Futures to lose mappings. ## v**2.0.4** Reload module mappings in an attempt to work around ColdBox Async losing them. ## **v2.0.3** **SyncProvider:** Add pool to releaseJob call ## v2.0.2 Fix moduleSettings missing a queryOptions key for failed jobs ## v2.0.1 ColdBoxAsyncProvider now correctly respects Worker Pool conifguration, including queues. ## v2.0.0 ### BREAKING CHANGES #### Worker Pools can only define a single queue to work In order to work with new Queue Providers, the Worker Pools need to be updated to only work a specific queue. This is because many future Queue Providers like RabbitMQ and Amazon SQS only support listening to a single queue in a consumer. If you previously had multiple queues defined in a Worker Pool, you will need to define multiple Worker Pool instances, one for each of the queues. ```cfscript // Old newWorkerPool( "default" ) .forConnection( "default" ) .onQueues( [ "priority", "default" ] ); // New newWorkerPool( "default" ) .forConnection( "default" ) .onQueue( "priority" ); newWorkerPool( "default" ) .forConnection( "default" ) .onQueue( "default" ); ``` Notice that the method has been renamed from `onQueues` to `onQueue`. Additionally, there are no more wildcard queues. Every queue you publish to must have a WorkerPool defined in order for that Job to be worked. Finally, queue priorities are defined by the number of workers (`quantity`) you define for the WorkerPool. WorkerPools can no longer share workers across queues. # Installation ## Requirements cbq requires the following: * Adobe ColdFusion (ACF) 2018+ **OR** Lucee 5+ * ColdBox 6+ The different [Queue Providers](/2.1.0/configuration/providers) each have their own requirements that are listed on their individual pages. ## Install via ForgeBox with CommandBox cbq is installed via [ForgeBox](https://forgebox.io) with [CommandBox](https://www.ortussolutions.com/products/commandbox). You can install the latest version using the command: ```shell install cbq ``` ## Load Java Libraries When using batches, cbq utilizes additional Java libraries included in the `lib/` folder. These need to be added to your Application's `javaSettings` in `Application.cfc`. ```cfscript // Java Integration this.javaSettings = { loadPaths: [ expandPath( "./modules/cbq/lib" ) ], loadColdFusionClassPath: true, reloadOnChange: false }; ``` {% hint style="info" %} Feel free to use mappings to point to the cbq path, if needed. {% endhint %} ## Additional Provider Installation Steps The [Queue Provider ](/2.1.0/configuration/providers)you use may have additional installation steps. Check out the individual provider pages for more details. # Walkthrough ## Install cbq Refer to the [Installation](/2.1.0/getting-started/installation) section for general installation instructions as well as instructions for your specific [provider](/2.1.0/configuration/providers). ## Create a Queue Connection Open up your `config/cbq.cfc` file and create your first Queue Connection: ```cfscript component { function configure() { newConnection( "default" ) .setProvider( "ColdBoxAsyncProvider@cbq" ); } } ``` ## Create a Worker Pool Next, create a Worker Pool for your new Queue Connection. This allows our application to work the Jobs we will dispatch. ```cfscript component { function configure() { newConnection( "default" ) .setProvider( "ColdBoxAsyncProvider@cbq" ); newWorkerPool( "default" ) .forConnection( "default" ); } } ``` ## Define your first Job A Job is a CFC that extends `cbq.models.Jobs.AbstractJob`. It can live anywhere in your application. ```cfscript // models/jobs/emails/SendWelcomeEmailJob.cfc component extends="cbq.models.Jobs.AbstractJob" { property name="userId"; function handle() { log.info( "Sending a Welcome email to User ###getUserId()#" ); /* sample code var user = getInstance( "User" ).findOrFail( getUserId() ); getInstance( "MailService@cbmailservices" ) .newMail( from = "no-reply@example.com", to = user.getEmail(), subject = "Welcome!", type = "html" ) .setView( "/_emails/users/welcome" ) .setBodyTokens( { "firstName" : user.getFirstName(), "lastName" : user.getLastName() } ) .send(); */ } } ``` ## Create an instance of your Job You can create an instance of your Job anywhere in your code — handlers, services, models, etc. Populate it with the specific data needed for this instance. ```cfscript var job = getInstance( "SendWelcomeEmailJob" ); job.setUserId( newUser.getId() ); ``` ## Dispatch your Job Once your Job is created and configured, `dispatch` it to the Queue Connection. ```cfscript job.dispatch(); ``` ## Watch your job get executed Check out LogBox to see your Job being executed. Congratulations! You've dispatched your first background Job using cbq! # Module Settings ## Full Module Settings ```cfscript settings = { // The path the custom config file to register connections and worker pools "configPath" : "config.cbq", // Flag if workers should be registered. // If your application only pushes to the queues, you can set this to `false`. "registerWorkers" : getSystemSetting( "CBQ_REGISTER_WORKERS", true ), // The interval to poll for changes to the worker pool scaling. // Defaults to 0 which turns off the scheduled scaling feature. "scaleInterval" : 0, // The default amount of time, in seconds, to delay a job. // Used if the connection and job doesn't define their own. "defaultWorkerBackoff" : 0, // The default amount of time, in seconds, to wait before timing out a job. // Used if the connection and job doesn't define their own. "defaultWorkerTimeout" : 60, // The default amount of attempts to try before failing a job. // Used if the connection and job doesn't define their own. "defaultWorkerMaxAttempts" : 1, // Datasource information for tracking batches. "batchRepositoryProperties" : { "tableName" : "cbq_batches", "datasource" : "", // `datasource` can also be a struct "queryOptions" : {} // The sibling `datasource` property overrides any defined datasource in queryOptions. }, // Flag to turn on logging failed jobs to a database table. "logFailedJobs" : false, // Datasource information for loggin failed jobs. "logFailedJobsProperties" : { "tableName" : "cbq_failed_jobs", "datasource" : "", // `datasource` can also be a struct. "queryOptions" : {} // The sibling `datasource` property overrides any defined datasource in `queryOptions`. }, // Flag to allow restricting Job interceptor execution using a `jobPattern` annotation. "registerJobInterceptorRestrictionAspect" : false }; ``` ### configPath The configPath is a dot-delimited path to your cbq Config Component. By convention, this should be placed in your application's `config/` folder alongside other config files like `ColdBox.cfc` and `WireBox.cfc`. ### registerWorkers This flag is responsible for spinning up Worker Pools when the application starts. If a particular instance of your application should **not** work jobs as well as dispatch them, then this setting should be set to `false`. To make this easy, you can set the `CBQ_REGISTER_WORKERS` environment variable and it will be picked up. (If you override this setting in your own `moduleSettings` it will still take precedence over the environment variable.) ### scaleInterval {% hint style="danger" %} This feature is not implemented yet. {% endhint %} This is the interval in seconds that the scale job is ran in the background. The scale job enables you to scale Worker Pools up or down based on any other factors in your application. **Setting this value to `0` disables the job entirely.** ### defaultWorkerBackoff This setting provides a default backoff value for a Worker Pool. This can still be overridden on a configured Worker Pool in your cbq config file or on an individual Job or `dispatch` call. ### defaultWorkerTimeout This setting provides a default timeout value for a Worker Pool. This can still be overridden on a configured Worker Pool in your cbq config file or on an individual Job or `dispatch` call. ### defaultWorkerMaxAttempts This setting provides a default max attempts value for a Worker Pool. This can still be overridden on a configured Worker Pool in your cbq config file or on an individual Job or `dispatch` call. ### batchRepositoryProperties A struct of configuration properties for a Batch Repository. This is only needed if you dispatch any batches. #### tableName The name of the batch table. The default is `cbq_batches`. #### datasource The datasource to use to interact with the batch table. If no datasource is provided, it uses the default application datasource. A `struct` can also be provided if your CFML engine supports it. #### queryOptions A struct of query options to pass to the `queryExecute` call when interacting with the batch table. If a `datasource` is defined above, it will override any `datasource` key inside the `queryOptions`. ### logFailedJobs This flag will send failed jobs to a configured database table when true. In some systems, this is called a Dead Letter Queue or DLQ. ### logFailedJobsProperties #### tableName The name of the failed jobs table. The default is `cbq_failed_jobs`. #### datasource The datasource to use to interact with the failed jobs table. If no datasource is provided, it uses the default application datasource. A `struct` can also be provided if your CFML engine supports it. #### queryOptions A struct of query options to pass to the `queryExecute` call when interacting with the failed jobs table. If a `datasource` is defined above, it will override any `datasource` key inside the `queryOptions`. ### registerJobInterceptorRestrictionAspect Flag to allow [restricting Job interceptor execution](/2.1.0/interceptors#jobpattern-annotation) using a `jobPattern` annotation. # Config File The cbq config file is where you define Queue Connections as well as Worker Pools. These Connections and Worker Pools can also be added, removed, or modified based on the current environment. When you want to change where your queued Jobs are sent or how they are worked, this is the file you will modify. ## Definitions ### Queue Connections Also referred to as "Connection." A Queue Connection defines where Jobs are serialized and the default settings that apply. ## configure The `configure` method is where the production Queue Connection and Worker Pools definitions are constructed. ### newConnection This command creates a new `QueueConnectionDefinition` builder component. It requires a unique name that will define this Connection and that will be used when defining Worker Pools.
ArgumentsTypeRequiredDefaultDescription
namestringtrueThe unique name for the new Queue Connection.
```cfscript component { function configure() { newConnection( "default" ); } } ``` A `QueueConnectionDefinition` has various methods to configure the Queue Connection. {% content-ref url="/pages/pghfGWdoAmKqeFiwqKtz" %} [Queue Connection](/2.1.0/configuration/config-file/queue-connection) {% endcontent-ref %} ### newWorkerPool This command creates a new `WorkerPoolDefinition` builder component. It requires a unique `name` that will define this Worker Pool and a `connectionName` that points to an already created Connection.
ArgumentsTypeRequiredDefaultDescription
namestringtrueThe unique name for the new Worker Pool.
connectionNamestringfalseA reference to an existing Connection. If not passed in here, the forConnection method must be called to define the Connection.
quantitynumericfalse1The number of workers to spin up for this Worker Pool.
queuesarrayfalse[ * ]An array of queues that this Worker Pool will work. A queue of * refers to all queues. Queues will be worked in the order provided.
forcebooleanfalsefalseIf false, an exception will be thrown if the Worker Pool name has already been registered. If true, the new definition will override the existing definition.
component {

    function configure() {
        newConnection( "default" );
        
        newWorkerPool( "default" ).forConnection( "default" );
    }

}
A `WorkerPoolDefinition` has various methods to configure the Worker Pool. {% content-ref url="/pages/FRcimvpoy6mO5qGgTvgb" %} [Worker Pool](/2.1.0/configuration/config-file/worker-pool) {% endcontent-ref %} ### reset Removes all Connections and Worker Pools ```cfscript getInstance( "Config@cbq" ).reset(); ``` ### Environment Overrides Just as in your `config/ColdBox.cfc` file or in `ModuleConfig.cfc` files, you can add, remove, or modify Queue Connection or Worker Pool definitions per environment. You do this by defining a method on your Config component matching the environment name you want to override. ```cfscript component { function configure() { newConnection( "default" ) .provider( "DBProvider@cbq" ); newWorkerPool( "default" ) .forConnection( "default" ) .quantity( 3 ) .timeout( 15 ); } /** * This method will be called after `configure` * and only if the current environment is `development`. */ function development() { withConnection( "default" ) .provider( "SyncProvider@cbq" ); withWorkerPool( "default" ) .quantity( 1 ) .timeout( 60 ); } } ``` ### withConnection Retrieves an already defined Queue Connection Definition for overriding. All the same methods are available. {% content-ref url="/pages/pghfGWdoAmKqeFiwqKtz" %} [Queue Connection](/2.1.0/configuration/config-file/queue-connection) {% endcontent-ref %} #### delete {% hint style="danger" %} This method is not yet implemented. {% endhint %} ### withWorkerPool Retrieves an already defined Worker Pool Definition for overriding. All the same methods are available. {% content-ref url="/pages/FRcimvpoy6mO5qGgTvgb" %} [Worker Pool](/2.1.0/configuration/config-file/worker-pool) {% endcontent-ref %} #### delete {% hint style="danger" %} This method is not yet implemented. {% endhint %} # Queue Connection A Queue Connection is defined inside your cbq config file using a `QueueConnectionDefinition` builder component. You can create one of these builder components using the [`newConnection`](/2.1.0/configuration/config-file#newconnection) method. ### QueueConnectionDefinition Methods #### provider Sets the provider for the Queue Connection. This can be any valid WireBox mapping and should implement the `IQueueProvider` interface. (There is no need to use the `implements` keyword.)
ArgumentsTypeRequiredDefaultDescription
providerstringtrueA valid WireBox mapping to the desired Provider.
#### setProvider Alias for [`provider`](#provider). #### setProperties Accepts a struct of properties to configure the Queue Connection. The available properties are usually defined by the Queue Provider being used. #### setDefaultQueue Sets the default queue to use for jobs dispatched on this Queue Connection. #### markAsDefault Marks this Queue Connection as the default Queue Connection for jobs dispatched without specifying a Queue Connection. #### setMakeDefault Alias for [`markAsDefault`](#markasdefault). # Worker Pool A Queue Connection is defined inside your cbq config file using a `QueueConnectionDefinition` builder component. You can create one of these builder components using the [`newConnection`](/2.1.0/configuration/config-file#newconnection) method. ### WorkerPoolDefinition Methods #### setName Sets the name for the Worker Pool. Usually not called directly as the `newWorkerPool` method requires a `name`. ```cfscript newWorkerPool( "default" ) .setName( "not-default" ); ``` #### forConnection Sets the name of the associated Connection for the Worker Pool. This must reference an already registered Connection. ```cfscript newConnection( "db" ) .provider( "DBProvider@cbq" ); newWorkerPool( "db-worker" ) .forConnection( "db" ); ``` #### setConnectionName Alias for [forConnection](#forconnection). #### quantity Sets the quantity of workers for this Worker Pool. ```cfscript newWorkerPool( "default" ) .forConnection( "db" ) .quantity( 3 ); ``` #### setQuantity Alias for [quantity](#quantity). #### onQueue The name of a queue to work. The default queue is named `default`. ```cfscript newWorkerPool( "premium-only" ) .forConnection( "db" ) .onQueue( "premium" ); newWorkerPool( "priority" ) .forConnection( "db" ) .onQueue( "priority" ) .quantity( 4 ); newWorkerPool( "default" ) .forConnection( "db" ); // uses `default` queue ``` Some Providers allow for working a priority of queues, such as the `DBProvider`. In these cases, you can pass an array of queues, in priority order, using the asterisk (`*`) as a wildcard character. ```cfscript newWorkerPool( "premium-only" ) .forConnection( "db" ) .onQueue( "premium" ); newWorkerPool( "default" ) .forConnection( "db" ) .onQueue( [ "priority", "*" ] ); ``` {% hint style="danger" %} **Throws:** `cbq.WorkerPool.MultipleQueuesNotSupported` {% endhint %} #### setQueue Alias for [onQueue](#onqueues). #### backoff Sets the backoff time amount, in seconds. ```cfscript newWorkerPool( "default" ) .forConnection( "db" ) .backoff( 30 ); ``` #### setBackoff Alias for [backoff](#backoff). #### timeout Sets the timeout time amount, in seconds. ```cfscript newWorkerPool( "default" ) .forConnection( "db" ) .timeout( 60 ); ``` #### setTimeout Alias for [timeout](#timeout). #### maxAttempts Sets the max number of attempts. ```cfscript newWorkerPool( "default" ) .forConnection( "db" ) .maxAttempts( 5 ); ``` #### setMaxAttempts Alias for [maxAttempts](#maxattempts). # Providers A Queue Provider provides a way to serialize jobs to a Queue Connection and to work jobs off a Queue Connection. The three built-in cbq providers are: * [SyncProvider@cbq](/2.1.0/configuration/providers/syncprovider) * [ColdBoxAsyncProvider@cbq](/2.1.0/configuration/providers/coldboxasyncprovider) * [DBProvider@cbq](/2.1.0/configuration/providers/dbprovider) ### Writing your own Custom Provider All Queue Providers extend the `AbstractQueueProvider` base component. They must implement three abstract methods: ```cfscript /** * Persists a serialized job to the Queue Connection * * @queueName The queue name for the job. * @payload The serialized job string. * @delay The delay (in seconds) before working the job. * @attempts The current attempt number. * * @return AbstractQueueProvider */ public any function push( required string queueName, required string payload, numeric delay = 0, numeric attempts = 0 ); /** * Starts a worker for a Worker Pool on this Queue Connection. * * @pool The Worker Pool that is working this Queue Connection. * * @return A function that when called will stop this worker. */ public function function startWorker( required WorkerPool pool ); /** * Starts any background processes needed for the Worker Pool. * * @pool The Worker Pool that is working this Queue Connection. * * @return AbstractQueueProvider */ public any function listen( required WorkerPool pool ); ``` # SyncProvider {% hint style="danger" %} This provider is not **durable**. (Pending jobs are not saved and may be lost if there are server issues between dispatching the job and working the job.) {% endhint %} {% hint style="danger" %} This provider **does not support multiple queues**. (Only a single queue can be provided to a Worker Pool instance. Additional queues to be worked should be registered as separate Worker Pool instances.) {% endhint %} The SyncProvider runs any jobs dispatched during a request in the same request synchronously. This can be especially useful in development when debugging jobs. If you job would throw an exception, you will see it with any chosen error handler and tools you use to debug local exceptions. # ColdBoxAsyncProvider {% hint style="danger" %} This provider is not **durable**. (Pending jobs are not persisted and may be lost if there are server issues between dispatching the job and working the job.) {% endhint %} {% hint style="danger" %} This provider **does not support multiple queues**. (Only a single queue can be provided to a Worker Pool instance. Additional queues to be worked should be registered as separate Worker Pool instances.) {% endhint %} The ColdBoxAsyncProvider runs any jobs dispatched on a background thread using ColdBox's AsyncManager. Jobs will be worked by the same server that dispatched them. The number of workers specified translates to the number of threads dedicated to working the jobs. # DBProvider {% hint style="success" %} This provider is **durable**. (Pending jobs are persisted. If there are server issues, the jobs will be worked when the issues are resolved. Multiple different servers can dispatch to this Queue Connection and multiple different Worker Pools can work these jobs.) {% endhint %} {% hint style="success" %} This provider **supports multiple queues**. (An array of prioritized queues can be assigned to Worker Pools worked by this provider.) {% endhint %} You may use a database as the backing engine for your Queue Connection using the DBProvider. All database grammars that are [supported by qb](https://qb.ortusbooks.com/v/9.0.0/installation-and-usage) are supported. ### Configuration The DBProvider has three optional arguments. The are presented below with their default values. ```cfscript { "tableName": "cbq_jobs", "datasource": null, "queryOptions": {} } ``` #### tableName The name of the table to use when managing jobs. This table should have the structure provided by the provided database migration file. #### datasource The name of the datasource to use when managing jobs. This overrides any datasource provided in the `queryOptions`. #### queryOptions A struct of options that will be passed to [`queryExecute`](https://cfdocs.org/queryexecute) when managing jobs. ### Jobs Table Structure The `cbq_jobs` table must have a specific structure. It is provided in the form of a migration file called `2000_01_01_000000_create_cbq_jobs_table.cfc`. This migration can be ran via [CommandBox Migrations](https://forgebox.io/view/commandbox-migrations) or [CFMigrations](https://forgebox.io/view/cfmigrations). If you wish, you can generate the table in other ways, so long as it matches the structure provided in the migration file. If you choose to use the migration file in your application, copy the file out to your own migrations folder first. ```cfscript schema.create( "cbq_jobs", function ( t ) { t.bigIncrements( "id" ); t.string( "queue" ); t.longText( "payload" ); t.unsignedTinyInteger( "attempts" ); t.unsignedInteger( "reservedDate" ).nullable(); t.unsignedInteger( "availableDate" ); t.unsignedInteger( "createdDate" ); t.index( "queue" ); } ); ``` # Defining a Job A Job represents both the object to be serialized to a [Connection](/2.1.0/configuration/config-file#connection) to eventually work as well as the code to execute when working a Job. ## Extending from AbstractJob The first step to define a Job is to extend from `AbstractJob`. This provides the necessary helper methods to run a Job through the cbq pipeline. ```cfscript component name="SendWelcomeEmailJob" extends="cbq.models.Jobs.AbstractJob" { } ``` ## Properties Defining properties on your job allow you to see at a glance what data your Job expects when constructing it. These properties and their values will be serialized to a [Connection](/2.1.0/configuration/config-file#connection) when dispatching a Job. {% hint style="info" %} Defining the properties is not strictly necessary, but your future self will thank you when you try to remember what properties your job is using. {% endhint %} ```cfscript component name="SendWelcomeEmailJob" extends="cbq.models.Jobs.AbstractJob" { property name="email"; property name="greeting"; } ``` ## The handle method The `handle` method is called when a Job is worked. Before being called, a Job will be reconstructed with the serialized data from the Connection. This method is the only required method when defining a Job. ```cfscript component name="SendWelcomeEmailJob" extends="cbq.models.Jobs.AbstractJob" { property name="mailService" inject="provider:MailService@cbmailservices"; property name="email"; property name="greeting"; function handle() { variables.mailService.newMail( to = variables.email, from = "noreply@example.com", subject = "Welcome!", type = "html", bodyTokens = { "greeting": variables.greeting "link": getInstance( "coldbox:requestContext" ) .buildLink( "home" ) } ) .setView( "_emails/welcome" ) .send(); } } ``` {% hint style="info" %} Prefer using `provider:` injections or inline `getInstance` calls for logic in your `handle` method. The `Job` component is created both when dispatching and when working your job, so utilizing these tools will reduce unnecessary processing time when dispatching your job. {% endhint %} ## Job Execution Properties A job can define several execution properties on the job itself. If defined, these values override the module, connection, or worker defaults. It can still be overridden using the job methods when creating a job. 
component extends="cbq.models.Jobs.AbstractJob" {

    // Name of the connection to dispatch this job on.
    variables.connection = "db";
    
    // Name of the queue to use for this job.
    variables.queue = "priority";
    
    // Time, in seconds, between job attempts, including the initial attempt.
    variables.backoff = 15;
    
    // Time, in seconds, to let a job run before failing it.
    variables.timeout = 30;
    
    // Max number of attempts before marking a job as failed.
    // Use 0 for unlimited retries.
    variables.maxAttempts = 9;

}
## Lifecycle Methods A Job can define a `before` or `after` method that will be called as part of the Job lifecycle. 
component
    name="SendWelcomeEmailJob"
    extends="cbq.models.Jobs.AbstractJob"
{

    property name="mailService" inject="provider:MailService@cbmailservices";
    
    property name="email";
    property name="greeting";
    
    function handle() {
        variables.mailService.newMail( 
            to = variables.email,
	    from = "noreply@example.com",
	    subject = "Welcome!",
            type = "html",
	    bodyTokens = { 
		"greeting": variables.greeting
		"link": getInstance( "coldbox:requestContext" )
		    .buildLink( "home" )
	    }
        )
        .setView( "_emails/welcome" )
        .send();
    }
    
    function before() {
        log.debug( "About to execute SendWelcomeEmailJob" );
    }
    
    function after() {
        log.debug( "Finished executing SendWelcomeEmailJob" );
    }

}
# Creating a Job After you define your job, you need to create an instance of your job and set the properties before you dispatch it onto your [Queue Connection](/2.1.0/configuration/config-file#connection). You can do this is a few different ways. ## Creating a Job Instance You can create a Job instance using WireBox anywhere in your ColdBox application. ```cfscript component { function index( event, rc, prc ) { var job = getInstance( "SendWelcomeEmailJob" ); } } ``` {% hint style="warning" %} Keep in mind that Job instances are transient. Do not inject a Job instance into a Component or Scope that is not transient. For instance, do not inject a Job instance as a property in a Handler. {% endhint %} ### Setting Job Properties Once you have a Job instance, you can set the data for the particular Job by calling the setter methods. ```cfscript component { function index( event, rc, prc ) { var job = getInstance( "SendWelcomeEmailJob" ); job.setEmail( "john@example.com" ); job.setGreeting( "Welcome!" ); } } ``` ### setProperties You can also set all the properties at once using the `setProperties` method. {% hint style="warning" %} Using this method will overwrite any previously set properties. {% endhint %} ```cfscript component { function index( event, rc, prc ) { var job = getInstance( "SendWelcomeEmailJob" ); job.setProperties( { "email": "john@example.com", "greeting": "Welcome!" } ); } } ``` ### onConnection You can override the connection for a job by calling the `onConnection` method with the desired connection name. ### setConnection See [onConnection](#onconnection). ### onQueue The queue name to use for the job. Queue names can be any string you choose, but to be worked a `WorkerPool` must be defined working the same connection and queue name. ### setQueue See [onQueue](#onqueue). ### setBackoff Sets the amount of time, in seconds, to wait in-between Job attempts — including the initial attempt. ### setDelay An alias for [setBackoff](#setbackoff). ### setTimeout Sets the amount of time, in seconds, to let a Job run on a worker before marking it as a failed attempt. ### setMaxAttempts Sets the maximum number of attempts before a Job is marked as failed. ### chain Sets an array of Jobs to be executed, in order, after this Job successfully executes. ### getMemento getMemento ## [cbq.job()](/2.1.0/cbq-model#job) You can also create a Job instance using the `cbq` model. (This can be injected into your components using the `cbq@cbq` DSL.) {% hint style="success" %} The cbq model is a singleton, so feel free to inject it into any component. {% endhint %} ```cfscript component { property name="cbq" inject="cbq@cbq"; function index( event, rc, prc ) { var job = cbq.job( job = "SendWelcomeEmailJob", properties = { "email": "john@example.com", "greeting": "Welcome!" }, chain = [], queue = "priority", connection = "db", backoff = 10, timeout = 30, maxAttempts = 3 ); } } ``` # Dispatching a Job ## Dispatching Creating a Job is all well and good, but the Job will not be executed until it is dispatched on to a [Queue Connection](/2.1.0/configuration/config-file#connection). If you have a Job instance, you can dispatch it by calling the `dispatch` method. ```cfscript component { function index( event, rc, prc ) { getInstance( "SendWelcomeEmailJob" ); .setEmail( "john@example.com" ); .setGreeting( "Welcome!" ) .dispatch(); } } ``` You can also create and dispatch a job in the same call using the `cbq.dispatch()` method. ```cfscript component { property name="cbq" inject="cbq@cbq"; function index( event, rc, prc ) { cbq.dispatch( job = "SendWelcomeEmailJob", properties = { "email": "john@example.com", "greeting": "Welcome!" }, chain = [], queue = "priority", connection = "db", backoff = 10, timeout = 30, maxAttempts = 3 ); } } ``` ## Interception Points ### onCBQJobAdded This is fired before serializing a Job and sending it to a connection. The data includes the `job` to be serialized and dispatched and the connection the Job is being dispatched on. ```cfscript variables.interceptorService.announce( "onCBQJobAdded", { "job" : job, "connection" : connection } ); ``` # Working a Job Dispatched jobs will get picked up by Worker Pools set up for the same connection and queue. They are picked up in order by created date (FIFO). If you have a valid Worker Pool, then you don't need to do anything else to have your job picked up. If you want to see all the details of cbq dispatching and marshalling jobs, enable `debug` logging in LogBox for `cbq`. When a Job is worked, the Job instance is created via WireBox and the memento hydrated into the Job component. Then the `handle` method is called. When a Job is completed, the Job will be removed from the persistent storage of the Queue Provider. If the Job attempt fails, it will be sent back to the Queue Provider to be retried, up to the `maxAttempts` amount. Once a Job has failed the `maxAttempts` amount it will be removed and marked as failed. If you have `logFailedJobs` enabled, it will be sent to the [failed jobs table](/2.1.0/jobs/failed-jobs#logging-failed-jobs). ## Inside \`handle\` ### release You can choose to manually release a Job back to a queue with an optional delay (in seconds) using the `release` function. This is useful when you need to delay processing for a Job, perhaps due to rate limiting or licensing constraints. {% hint style="info" %} Calling `release` does not stop processing the `handle` method, so make sure you `return` if you don't want to keep executing your `handle` method. {% endhint %} ```cfscript component { function handle() { this.release( 60 ); // in seconds return; } } ``` ## Interception Points cbq fires a number of interception points during a Job's lifecycle. ### onCBQJobMarshalled This is fired before the `handle` method is called on a Job. The data includes the `job` to be executed. ```cfscript variables.interceptorService.announce( "onCBQJobMarshalled", { "job" : job } ); ``` ### onCBQJobComplete This is fired after a Job completes successfully. The data includes the `job` to be executed as potentially a `result` returned from the `Job`'s handle method. ```cfscript variables.interceptorService.announce( "onCBQJobComplete", { "job" : job, "result" : isNull( result ) ? javacast( "null", "" ) : result } ); ``` ## Tips ### Need up to date data? Could your data change between when you dispatched a job and when you work the job? If so, consider using the key as a property and fetching the data from inside the job. ### Need to use historical data? In some cases, you want to work with the data that existed at the time the Job was dispatched. In these cases, make sure to include all the needed data in the Job instance. ### Check if you need to work the Job still Things might have changed since the Job was dispatched. Most Queue Providers do not support querying the current Jobs queue or interacting with it in advance, so check in your `dispatch` method if the Job still needs to be worked. ### Need to try the Job later? In some cases, you need to retry your Job later. In these cases, use the `release` method to send the Job back to the queue. It also takes an optional `delay` which sets the `backoff` time for the next Job attempt. You can combine this with setting the `maxAttempts` of a Job to `0` to retry it indefinitely. Remember that a Job is only reattempted if it fails. If you decide that a Job is finished, just `return`. ### Jobs can dispatch other Jobs You can dispatch other Jobs from your Job. This can be different Jobs or even a similar instance of the same Job. ## Why isn't my Job getting picked up? To work a cbq job, you need a [defined Worker Pool in your cbq Config file.](/2.1.0/configuration/config-file#newworkerpool) In order to work a Job, the Worker Pool must: 1. Be working the same [connection](/2.1.0/configuration/config-file/worker-pool#forconnection) as the Job. 2. Be working the same [queue](/2.1.0/configuration/config-file/worker-pool#onqueues) as the Job. 3. Have at least one active worker (defined with [`setQuantity`](/2.1.0/configuration/config-file/worker-pool#setquantity)). If these requirements are not met, the job will stay in the queue storage, unable to get picked up. # Failed Jobs ## Failed Attempt vs Failed Job An important distinction to make is between failed Job attempts and failed Jobs. A Job is attempted up to the defined `maxAttempts`. If a Job fails all of its `maxAttempts` then it is marked as a failed Job and removed from the queue. Otherwise, the Job is dispatched back to the queue with the current execution count increased. ## LogBox Failed Job attempts are logged to LogBox. You will see the exception as well as the serialized Job memento in the `extraInfo`. ## Interceptors ### onCBQJobException This interception point is fired for every **Job Attempt** failure. The data includes the `job` component and the `exception`. ```cfscript variables.interceptorService.announce( "onCBQJobException", { "job" : job, "exception" : e } ); ``` ### onCBQJobFailed This interception point is fired when the Job is marked as failed — when the Job has failed all attempts up to the configured `maxAttempts`. The data includes the `job` component and the `exception`. ```cfscript variables.interceptorService.announce( "onCBQJobFailed", { "job" : job, "exception" : e } ); ``` ## onFailure Job Method You can define an `onFailure` method on your Job component. It will be called with the `exception`. ```cfscript component extends="cbq.models.Jobs.AbstractJob" { function handle() { // ... } function onFailure( e ) { log.error( "Bad things happened: #e.message#" ); } } ``` ## Logging Failed Jobs ## Failed Jobs Table cbq includes an interceptor to log failed jobs to a database. You can enable this in your module settings: ```cfscript moduleSettings = { "cbq": { // Flag to turn on logging failed jobs to a database table. "logFailedJobs" : false, // Datasource information for loggin failed jobs. "logFailedJobsProperties" : { "tableName" : "cbq_failed_jobs", "datasource" : "", // `datasource` can also be a struct. "queryOptions" : {} // The sibling `datasource` property overrides any defined datasource in `queryOptions`. } } }; ``` Also included is a migration file to add the needed table to your database. You can find it in `resources/database/migrations/2000_01_01_000002_create_cbq_failed_jobs_table.cfc`. It is also included below. ```cfscript component { function up( schema ) { schema.create( "cbq_failed_jobs", function ( t ) { t.bigIncrements( "id" ); t.string( "connection" ); t.string( "queue" ); t.string( "mapping" ); t.longText( "memento" ); t.longText( "properties" ); t.string( "exceptionType" ).nullable(); t.string( "exceptionMessage" ); t.string( "exceptionDetail" ).nullable(); t.longText( "exceptionExtendedInfo" ).nullable(); t.longText( "exceptionStackTrace" ); t.longText( "exception" ); t.timestamp( "failedDate" ).withCurrent(); } ); } function down( schema ) { schema.dropIfExists( "cbq_failed_jobs" ); } } ``` {% code title="MySQL" %} ```sql CREATE TABLE `cbq_failed_jobs` ( `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT, `connection` NVARCHAR(255) NOT NULL, `queue` NVARCHAR(255) NOT NULL, `mapping` NVARCHAR(255) NOT NULL, `memento` LONGTEXT NOT NULL, `properties` LONGTEXT NOT NULL, `exceptionType` NVARCHAR(255), `exceptionMessage` NVARCHAR(255) NOT NULL, `exceptionDetail` NVARCHAR(255), `exceptionExtendedInfo` LONGTEXT, `exceptionStackTrace` LONGTEXT NOT NULL, `exception` LONGTEXT NOT NULL, `failedDate` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ) ``` {% endcode %} {% code title="SQL Server" %} ```sql CREATE TABLE [dbo].[cbq_failed_jobs]( [id] BIGINT NOT NULL IDENTITY, [connection] NVARCHAR(255) NOT NULL, [queue] NVARCHAR(255) NOT NULL, [mapping] NVARCHAR(255) NOT NULL, [memento] NVARCHAR(MAX) NOT NULL, [properties] NVARCHAR(MAX) NOT NULL, [exceptionType] NVARCHAR(255), [exceptionMessage] NVARCHAR(255) NOT NULL, [exceptionDetail] NVARCHAR(255), [exceptionExtendedInfo] NVARCHAR(MAX), [exceptionStackTrace] NVARCHAR(MAX) NOT NULL, [exception] NVARCHAR(MAX) NOT NULL, [failedDate] DATETIME2 NOT NULL CONSTRAINT [df_cbq_failed_jobs_failedDate] DEFAULT CURRENT_TIMESTAMP ) ``` {% endcode %} ## Retrying a Failed Job ## Configuring Job Backoff # Chained Jobs Job chains mean that after a Job completes successfully it dispatches the next Job in the chain. If a Job fails, no more Jobs in the chain are dispatched. ## Dispatching Jobs from another Job One way to make a Job chain is to dispatch a Job from inside another Job. This gives you consistency in your Job chain, and a logical code path to follow. ```cfscript // FulfillOrderJob.cfc component extends="cbq.models.Jobs.AbstractJob" { function handle() { var productId = getProperties().productId; processPayment( productId ); getInstance( "SendProductLinkEmail" ) .onConnection( "fulfillment" ) .setProperties( { "productId" : productId, "userId" : getProperties().userId } ) .dispatch(); } } ``` ## Creating a Chain Another way to make a Job chain is to create is when dispatching the Job. This gives you flexibility to compose Job chains at runtime. ```cfscript // handlers/Main.cfc component { property name="cbq" inject="cbq@cbq"; function create() { cbq.chain( [ cbq.job( "FulfillOrderJob", { "productId": rc.productId } ), cbq.job( job = "SendProductLinkEmail", properties = { "productId": rc.productId, "userId": auth().getUserId() }, connection = "fulfillment" ), ] ).dispatch(); } } ``` ## Dispatching Jobs on Failure Utilizing the `onFailure` method of a Job, we can dispatch another Job if our Job fails. ```cfscript // FulfillOrderJob.cfc component extends="cbq.models.Jobs.AbstractJob" { function handle() { // ... } function onFailure( e ) { getInstance( "SendOrderProblemEmail" ) .onConnection( "fulfillment" ) .setProperties( { "productId" : productId, "userId" : getProperties().userId, "message" : e.message } ) .dispatch(); } } ``` # Batched Jobs cbq allows for tracking jobs as part of a batch. The main use case for batched jobs is to dispatch additional jobs when a batch has completed successfully, completed with failures, or completed in any fashion. (Think `try`, `catch`, `finally`.) ## Setup To utilize batches, you first need to set up a Batch repository. Batches are tracked separate from jobs. cbq utilizes a database repository to track batches regardless of what provider your Job's connection uses. A database migration is provided in `resources/database/migrations/2000_01_01_000001_create_cbq_batches_table.cfc`. You can copy this to your own project to run via [cfmigrations](https://forgebox.io/view/cfmigrations) or you can reference the migration or SQL scripts [below](#migrations-and-sql-scripts). You can customize the batch repository using the `batchRepositoryProperties` of cbq's `moduleSettings`. This is a struct with two properties you can set: #### tableName This defaults to `cbq_batches`. You can set this to any unique table name in your datasource. #### queryOptions This is a struct of options that will be passed to `queryExecute`. The most common use case for this property is to specify a specific datasource to use for the batches table. This defaults to an empty struct (`{}`). ## Defining Batches To create a batch, you can get an instance of `PendingBatch@cbq` from WireBox or use the [`cbq.batch()`](/2.1.0/cbq-model#batch) helper method. Once you have a `PendingBatch` instance, you can add jobs to the batch using the `add` method. ### add Adds a single Job or an array of Jobs to a PendingBatch.
ArgumentsTypeRequiredDefaultDescription
jobstring | Job | array<Job>trueThe Job WireBox id, Job instance, or array of Job instances to add to the PendingBatch.
propertiesStructfalse{}A struct of properties for the Job instance. (Only used when providing a Job WireBox id.)
chainarray<Job>false[]An array of Jobs to chain after this Job. (Only used when providing a Job WireBox id.)
queuestringfalseThe queue the Job belongs to. (Only used when providing a Job WireBox id.)
connectionstringfalseThe Connection to dispatch the Job on. (Only used when providing a Job WireBox id.)
backoffnumericfalseThe amount of time, in seconds, to wait between attempting Jobs. (Only used when providing a Job WireBox id.)
timeoutnumericfalseThe amount of time, in seconds, to wait before treating a Job as erroring. (Only used when providing a Job WireBox id.)
maxAttemptsnumericfalseThe maximum amount of attempts of a Job before treating the Job as failed. (Only used when providing a Job WireBox id.)
**Return:** The PendingBatch to be dispatched. ```cfscript batch .add( cbq.job( "ImportCsvJob", { "start": 1, "end": 100 } ) ) .add( "ImportCsvJob", { "start": 101, "end": 200 } ) .add( [ cbq.job( "ImportCsvJob", { "start": 201, "end": 300 } ), cbq.job( "ImportCsvJob", { "start": 301, "end": 400 } ), cbq.job( "ImportCsvJob", { "start": 401, "end": 500 } ) ] ); ``` The primary use case of batches is to dispatch lifecycle jobs when the batch has completed and if the batch completed successfully or completed with failures. These jobs can be configured using the `then`, `catch`, and `finally` methods. ### then Defines a Job to be dispatched when all the jobs in the batch finishes successfully.
ArgumentsTypeRequiredDefaultDescription
jobstring | JobtrueThe Job WireBox id or Job instance to execute if all the jobs in the Batch complete successfully.
propertiesStructfalse{}A struct of properties for the Job instance. (Only used when providing a Job WireBox id.)
chainarray<Job>false[]An array of Jobs to chain after this Job. (Only used when providing a Job WireBox id.)
queuestringfalseThe queue the Job belongs to. (Only used when providing a Job WireBox id.)
connectionstringfalseThe Connection to dispatch the Job on. (Only used when providing a Job WireBox id.)
backoffnumericfalseThe amount of time, in seconds, to wait between attempting Jobs. (Only used when providing a Job WireBox id.)
timeoutnumericfalseThe amount of time, in seconds, to wait before treating a Job as erroring. (Only used when providing a Job WireBox id.)
maxAttemptsnumericfalseThe maximum amount of attempts of a Job before treating the Job as failed. (Only used when providing a Job WireBox id.)
**Return:** The `PendingBatch` instance. ```cfscript cbq.batch( [ cbq.job( "ImportCsvJob", { "start": 1, "end": 100 } ), cbq.job( "ImportCsvJob", { "start": 101, "end": 200 } ), cbq.job( "ImportCsvJob", { "start": 201, "end": 300 } ), cbq.job( "ImportCsvJob", { "start": 301, "end": 400 } ), cbq.job( "ImportCsvJob", { "start": 401, "end": 500 } ) ] ) .then( cbq.job( "ImportCsvSuccessfulJob" ) ); ``` ### catch Defines a Job to be dispatched the first time a job in the Batch fails.
ArgumentsTypeRequiredDefaultDescription
jobstring | JobtrueThe Job WireBox id or Job instance to execute the first time a job in the Batch fails.
propertiesStructfalse{}A struct of properties for the Job instance. (Only used when providing a Job WireBox id.)
chainarray<Job>false[]An array of Jobs to chain after this Job. (Only used when providing a Job WireBox id.)
queuestringfalseThe queue the Job belongs to. (Only used when providing a Job WireBox id.)
connectionstringfalseThe Connection to dispatch the Job on. (Only used when providing a Job WireBox id.)
backoffnumericfalseThe amount of time, in seconds, to wait between attempting Jobs. (Only used when providing a Job WireBox id.)
timeoutnumericfalseThe amount of time, in seconds, to wait before treating a Job as erroring. (Only used when providing a Job WireBox id.)
maxAttemptsnumericfalseThe maximum amount of attempts of a Job before treating the Job as failed. (Only used when providing a Job WireBox id.)
**Return:** The `PendingBatch` instance. ```cfscript cbq.batch( [ cbq.job( "ImportCsvJob", { "start": 1, "end": 100 } ), cbq.job( "ImportCsvJob", { "start": 101, "end": 200 } ), cbq.job( "ImportCsvJob", { "start": 201, "end": 300 } ), cbq.job( "ImportCsvJob", { "start": 301, "end": 400 } ), cbq.job( "ImportCsvJob", { "start": 401, "end": 500 } ) ] ) .catch( cbq.job( "ImportCsvFailedJob" ) ); ``` ### finally Defines a Job to be dispatched after all the jobs in the Batch have executed successfully or failed.
ArgumentsTypeRequiredDefaultDescription
jobstring | JobtrueThe Job WireBox id or Job instance to execute after all the jobs in the Batch have executed successfully or failed.
propertiesStructfalse{}A struct of properties for the Job instance. (Only used when providing a Job WireBox id.)
chainarray<Job>false[]An array of Jobs to chain after this Job. (Only used when providing a Job WireBox id.)
queuestringfalseThe queue the Job belongs to. (Only used when providing a Job WireBox id.)
connectionstringfalseThe Connection to dispatch the Job on. (Only used when providing a Job WireBox id.)
backoffnumericfalseThe amount of time, in seconds, to wait between attempting Jobs. (Only used when providing a Job WireBox id.)
timeoutnumericfalseThe amount of time, in seconds, to wait before treating a Job as erroring. (Only used when providing a Job WireBox id.)
maxAttemptsnumericfalseThe maximum amount of attempts of a Job before treating the Job as failed. (Only used when providing a Job WireBox id.)
**Return:** The `PendingBatch` instance. ```cfscript cbq.batch( [ cbq.job( "ImportCsvJob", { "start": 1, "end": 100 } ), cbq.job( "ImportCsvJob", { "start": 101, "end": 200 } ), cbq.job( "ImportCsvJob", { "start": 201, "end": 300 } ), cbq.job( "ImportCsvJob", { "start": 301, "end": 400 } ), cbq.job( "ImportCsvJob", { "start": 401, "end": 500 } ) ] ) .finally( cbq.job( "ImportCsvCompletedJob" ) ); ``` ## Dispatching Batches A `PendingBatch` must be dispatched before any of the jobs contained in it are dispatched. This is done using the `dispatch` method on the `PendingBatch`. ### dispatch Dispatches a PendingBatch and all the Jobs it contains.
ArgumentsTypeRequiredDefaultDescription
No arguments
**Return:** A `Batch` instance created from this `PendingBatch`. ## Interacting with Batches ## Migrations and SQL Scripts ### cfmigrations ```cfscript schema.create( "cbq_batches", function ( t ) { t.string( "id" ).primaryKey(); t.string( "name" ); t.unsignedInteger( "totalJobs" ); t.unsignedInteger( "pendingJobs" ); t.unsignedInteger( "failedJobs" ); t.text( "failedJobIds" ); // JSON column t.text( "options" ).nullable(); // JSON column t.datetime( "createdDate" ); t.datetime( "cancelledDate" ).nullable(); t.datetime( "completedDate" ).nullable(); } ); ``` ### MySQL ```sql CREATE TABLE `cbq_batches` ( `id` VARCHAR(255) NOT NULL, `name` VARCHAR(255) NOT NULL, `totalJobs` INTEGER UNSIGNED NOT NULL, `pendingJobs` INTEGER UNSIGNED NOT NULL, `failedJobs` INTEGER UNSIGNED NOT NULL, `failedJobIds` TEXT NOT NULL, `options` TEXT, `createdDate` DATETIME NOT NULL, `cancelledDate` DATETIME, `completedDate` DATETIME, CONSTRAINT `pk_cbq_batches_id` PRIMARY KEY (`id`) ) ``` ### SQL Server ```sql CREATE TABLE [cbq_batches] ( [id] VARCHAR(255) NOT NULL, [name] VARCHAR(255) NOT NULL, [totalJobs] INTEGER NOT NULL, [pendingJobs] INTEGER NOT NULL, [failedJobs] INTEGER NOT NULL, [failedJobIds] VARCHAR(MAX) NOT NULL, [options] VARCHAR(MAX), [createdDate] DATETIME2 NOT NULL, [cancelledDate] DATETIME2, [completedDate] DATETIME2, CONSTRAINT [pk_cbq_batches_id] PRIMARY KEY ([id]) ) ``` ### Postgres ```sql CREATE TABLE "cbq_batches" ( "id" VARCHAR(255) NOT NULL, "name" VARCHAR(255) NOT NULL, "totalJobs" INTEGER NOT NULL, "pendingJobs" INTEGER NOT NULL, "failedJobs" INTEGER NOT NULL, "failedJobIds" TEXT NOT NULL, "options" TEXT, "createdDate" TIMESTAMP NOT NULL, "cancelledDate" TIMESTAMP, "completedDate" TIMESTAMP, CONSTRAINT "pk_cbq_batches_id" PRIMARY KEY ("id") ) ``` ### Oracle ```sql CREATE TABLE "CBQ_BATCHES" ( "ID" VARCHAR2(255) NOT NULL, "NAME" VARCHAR2(255) NOT NULL, "TOTALJOBS" NUMBER(10, 0) NOT NULL, "PENDINGJOBS" NUMBER(10, 0) NOT NULL, "FAILEDJOBS" NUMBER(10, 0) NOT NULL, "FAILEDJOBIDS" CLOB NOT NULL, "OPTIONS" CLOB, "CREATEDDATE" DATE NOT NULL, "CANCELLEDDATE" DATE, "COMPLETEDDATE" DATE, CONSTRAINT "PK_CBQ_BATCHES_ID" PRIMARY KEY ("ID") ) ``` ### SQLite ```sql CREATE TABLE "cbq_batches" ( "id" TEXT NOT NULL, "name" TEXT NOT NULL, "totalJobs" INTEGER NOT NULL, "pendingJobs" INTEGER NOT NULL, "failedJobs" INTEGER NOT NULL, "failedJobIds" TEXT NOT NULL, "options" TEXT, "createdDate" DATETIME NOT NULL, "cancelledDate" DATETIME, "completedDate" DATETIME, PRIMARY KEY ("id") ) ``` # cbq Model This model is provided to make certain tasks easier when defining and dispatching jobs, chains, and batches. ### dispatch Dispatches a job or chain of jobs.
ArgumentsTypeRequiredDefaultDescription
jobstring | Job | array<Job>trueA Job instance or a WireBox ID of a Job instance. If the WireBox ID doesn't exist, a NonExecutableJob instance will be used instead. This allows you to dispatch a Job from a server where that Job component is not defined. If an array is passed, a Job Chain is created and dispatched.
propertiesstructfalse{}A struct of properties for the Job instance.
chainarray<Job>false[]An array of Jobs to chain after this Job.
queuestringfalseThe queue the Job belongs to.
connectionstringfalseThe Connection to dispatch the Job on.
backoffnumericfalseThe amount of time, in seconds, to wait between attempting Jobs.
timeoutnumericfalseThe amount of time, in seconds, to wait before treating a Job as erroring.
maxAttemptsnumericfalseThe maximum amount of attempts of a Job before treating the Job as failed.
**Return:** The dispatched Job instance. ```cfscript cbq.dispatch( job = "SendWelcomeEmailJob", properties = { "body": "first body" }, queue = "default" ); ``` ### job Creates a job or chain of jobs to be dispatched.
ArgumentsTypeRequiredDefaultDescription
jobstring | Job | array<Job>trueA Job instance or a WireBox ID of a Job instance. If the WireBox ID doesn't exist, a NonExecutableJob instance will be used instead. This allows you to dispatch a Job from a server where that Job component is not defined. If an array is passed, a Job Chain is created and dispatched.
propertiesstructfalse{}A struct of properties for the Job instance.
chainarray<Job>false[]An array of Jobs to chain after this Job.
queuestringfalseThe queue the Job belongs to.
connectionstringfalseThe Connection to dispatch the Job on.
backoffnumericfalseThe amount of time, in seconds, to wait between attempting Jobs.
timeoutnumericfalseThe amount of time, in seconds, to wait before treating a Job as erroring.
maxAttemptsnumericfalseThe maximum amount of attempts of a Job before treating the Job as failed.
**Return:** The new Job instance. ```cfscript cbq.job( "SendWelcomeEmailJob" ) .setProperties( { "body": "first body" } ) .onQueue( "default" ) .dispatch(); ``` ### chain Creates a chain of jobs to be ran. Alias for calling `firstJob.chain( otherJobs )`.
ArgumentsTypeRequiredDefaultDescription
jobsarray<Job>trueThe array of jobs to run in order in a chain.
**Return:** The first job of the chain with the chained jobs configured to be dispatched. ```cfscript cbq.chain( [ cbq.job( "SendWelcomeEmailJob", { "body": "One" }, [], "default" ), cbq.job( "SendWelcomeEmailJob", { "body": "Two" }, [], "default", "sync" ), cbq.job( "SendWelcomeEmailJob", { "body": "Three" }, [], "default" ) ] ) .dispatch() ``` ### batch Creates a PendingBatch from the Jobs provided. {% hint style="warning" %} To use batches, you must first configure a `BatchRepository`. Learn more in the [Batched Jobs documentation](/2.1.0/jobs/batched-jobs). {% endhint %}
ArgumentsTypeRequiredDefaultDescription
jobsarray<Job>trueAn array of jobs to batch together.
**Return:** The PendingBatch to be dispatched. ```cfscript var batch = cbq .batch( [ cbq.job( "ImportCsvJob", { "start": 1, "end": 100 } ), cbq.job( "ImportCsvJob", { "start": 101, "end": 200 } ), cbq.job( "ImportCsvJob", { "start": 201, "end": 300 } ), cbq.job( "ImportCsvJob", { "start": 301, "end": 400 } ), cbq.job( "ImportCsvJob", { "start": 401, "end": 500 } ) ] ) .then( cbq.job( "ImportCsvSuccessfulJob" ) ) .catch( cbq.job( "ImportCsvFailedJob" ) ) .finally( cbq.job( "ImportCsvCompletedJob" ) ) .dispatch(); ``` # Interceptors ## Interception Points cbq announces the following interception points: ### `onCBQJobAdded` This is called when a Job is dispatched to a Queue. ### `onCBQJobMarshalled` This is called when a Job is pulled off the Queue to work. ### `onCBQJobComplete` This is called when a Job successfully finishes executing. ### `onCBQJobException` This is called when encountering an exception when handling a Job. ### `onCBQJobFailed` This is called when a Job is considered failed, after exhausting its `maxAttempts`. ## JobPattern Annotation {% hint style="warning" %} To use the `jobPattern` annotation, you must have enabled the `registerJobInterceptorRestrictionAspect` setting in your [module settings](/2.1.0/configuration/module-settings). {% endhint %} Interceptors listening on the cbq interception points listed above can optionally restrict execution to certain Jobs using a `jobPattern` annotation (similar to the `eventPattern` annotation [in ColdBox](https://coldbox.ortusbooks.com/the-basics/interceptors/restricting-execution)). ```cfscript component { function onCBQJobMarshalled( event, data ) jobPattern="SendWelcomeEmailJob" { // check if we've hit the email send limits for the month } } ``` This annotation accepts a regex string to check against the Job full name: ```cfscript component { function onCBQJobMarshalled( event, data ) jobPattern="^.*Email.*$" { // check if we've hit the email send limits for the month } } ``` {% hint style="info" %} Jobs that do not match the interception point will send a notice to the `debug` log, if that is turned on for `JobInterceptorRestriction`. {% endhint %} # Contributing # Contributors # Prior Art Initial ideas behind cbq were derived from [Laravel](https://laravel.com). # Dedication # Home ## A protocol-based queueing system for ColdBox Queues allow you to push work to the background, schedule work to be done later, or even process work on many different machines. It runs on a provider-based system allowing a unified API to talk with many different queue backends. ### Where to go next? * [Installation](/3.0.0/getting-started/installation) * [Walkthrough](/3.0.0/getting-started/walkthrough) * CFCasts Series (Coming Soon) * API Docs # What's New? ## v3.0.2 * Add error logging around logging failed jobs. ## v3.0.1 * **DBProvider:** Fix releasing job timeouts using the wrong value ## v3.0.0 * The `failedDate` column now uses a Unix timestamp as the column type. This avoids any timezone issues and aligns more closely with the `cbq_jobs` table used by the `DBProvider`. * Allow worker pools to finish currently running jobs, up to a configurable timeout. * Add optional clean-up tasks for completed or failed jobs, failed job logs, and completed or cancelled batches. * **DBProvider:** Improve database locking to avoid duplicate runs of the same job. * Fixes unwrapping an optional in a log message. ## v2.1.0 Add back ability to [work on multiple queues](/3.0.0/configuration/config-file/worker-pool#onqueue) on a per-Provider basis. Currently only the `DBProvider` supports it. Add support for `before` and `after` [lifecycle methods](/3.0.0/jobs/defining-a-job#lifecycle-methods) on a Job instance. Add ability to [restrict interceptor execution ](/3.0.0/interceptors#jobpattern-annotation)with a `jobPattern` annotation. (This is similar to the `eventPattern` annotation [provided by ColdBox](https://coldbox.ortusbooks.com/the-basics/interceptors/restricting-execution).) ## v2.0.5 **DBProvider**: Disable `forceRun` because it is causing ColdBox Futures to lose mappings. ## v**2.0.4** Reload module mappings in an attempt to work around ColdBox Async losing them. ## **v2.0.3** **SyncProvider:** Add pool to releaseJob call ## v2.0.2 Fix moduleSettings missing a queryOptions key for failed jobs ## v2.0.1 ColdBoxAsyncProvider now correctly respects Worker Pool conifguration, including queues. ## v2.0.0 ### BREAKING CHANGES #### Worker Pools can only define a single queue to work In order to work with new Queue Providers, the Worker Pools need to be updated to only work a specific queue. This is because many future Queue Providers like RabbitMQ and Amazon SQS only support listening to a single queue in a consumer. If you previously had multiple queues defined in a Worker Pool, you will need to define multiple Worker Pool instances, one for each of the queues. ```cfscript // Old newWorkerPool( "default" ) .forConnection( "default" ) .onQueues( [ "priority", "default" ] ); // New newWorkerPool( "default" ) .forConnection( "default" ) .onQueue( "priority" ); newWorkerPool( "default" ) .forConnection( "default" ) .onQueue( "default" ); ``` Notice that the method has been renamed from `onQueues` to `onQueue`. Additionally, there are no more wildcard queues. Every queue you publish to must have a WorkerPool defined in order for that Job to be worked. Finally, queue priorities are defined by the number of workers (`quantity`) you define for the WorkerPool. WorkerPools can no longer share workers across queues. # Upgrade Guide ## v2 to v3 In v3, the `cbq_failed_jobs` table migrates the `failedDate` column type from a timestamp to a unix timestamp. A migration file is included in `resources/database/migrations/2000_01_01_000006_use_unix_timestamp_for_failed_job_log_failedDate.cfc`. To run this migration, your failed jobs table will need to be empty. Alternatively, you can write your own migration that converts the timestamp to a unix timestamp. (The logic is different for each database grammar.) # Installation ## Requirements cbq requires the following: * Adobe ColdFusion (ACF) 2018+ **OR** Lucee 5+ * ColdBox 6+ The different [Queue Providers](/3.0.0/configuration/providers) each have their own requirements that are listed on their individual pages. ## Install via ForgeBox with CommandBox cbq is installed via [ForgeBox](https://forgebox.io) with [CommandBox](https://www.ortussolutions.com/products/commandbox). You can install the latest version using the command: ```shell install cbq ``` ## Load Java Libraries When using batches, cbq utilizes additional Java libraries included in the `lib/` folder. These need to be added to your Application's `javaSettings` in `Application.cfc`. ```cfscript // Java Integration this.javaSettings = { loadPaths: [ expandPath( "./modules/cbq/lib" ) ], loadColdFusionClassPath: true, reloadOnChange: false }; ``` {% hint style="info" %} Feel free to use mappings to point to the cbq path, if needed. {% endhint %} ## Additional Provider Installation Steps The [Queue Provider ](/3.0.0/configuration/providers)you use may have additional installation steps. Check out the individual provider pages for more details. # Walkthrough ## Install cbq Refer to the [Installation](/3.0.0/getting-started/installation) section for general installation instructions as well as instructions for your specific [provider](/3.0.0/configuration/providers). ## Create a Queue Connection Open up your `config/cbq.cfc` file and create your first Queue Connection: ```cfscript component { function configure() { newConnection( "default" ) .setProvider( "ColdBoxAsyncProvider@cbq" ); } } ``` ## Create a Worker Pool Next, create a Worker Pool for your new Queue Connection. This allows our application to work the Jobs we will dispatch. ```cfscript component { function configure() { newConnection( "default" ) .setProvider( "ColdBoxAsyncProvider@cbq" ); newWorkerPool( "default" ) .forConnection( "default" ); } } ``` ## Define your first Job A Job is a CFC that extends `cbq.models.Jobs.AbstractJob`. It can live anywhere in your application. ```cfscript // models/jobs/emails/SendWelcomeEmailJob.cfc component extends="cbq.models.Jobs.AbstractJob" { property name="userId"; function handle() { log.info( "Sending a Welcome email to User ###getUserId()#" ); /* sample code var user = getInstance( "User" ).findOrFail( getUserId() ); getInstance( "MailService@cbmailservices" ) .newMail( from = "no-reply@example.com", to = user.getEmail(), subject = "Welcome!", type = "html" ) .setView( "/_emails/users/welcome" ) .setBodyTokens( { "firstName" : user.getFirstName(), "lastName" : user.getLastName() } ) .send(); */ } } ``` ## Create an instance of your Job You can create an instance of your Job anywhere in your code — handlers, services, models, etc. Populate it with the specific data needed for this instance. ```cfscript var job = getInstance( "SendWelcomeEmailJob" ); job.setUserId( newUser.getId() ); ``` ## Dispatch your Job Once your Job is created and configured, `dispatch` it to the Queue Connection. ```cfscript job.dispatch(); ``` ## Watch your job get executed Check out LogBox to see your Job being executed. Congratulations! You've dispatched your first background Job using cbq! # Module Settings ## Full Module Settings ```cfscript settings = { // The path the custom config file to register connections and worker pools "configPath" : "config.cbq", // Flag if workers should be registered. // If your application only pushes to the queues, you can set this to `false`. "registerWorkers" : getSystemSetting( "CBQ_REGISTER_WORKERS", true ), // The interval to poll for changes to the worker pool scaling. // Defaults to 0 which turns off the scheduled scaling feature. "scaleInterval" : 0, // The default amount of time, in seconds, to delay a job. // Used if the connection and job doesn't define their own. "defaultWorkerBackoff" : 0, // The default amount of time, in seconds, to wait before timing out a job. // Used if the connection and job doesn't define their own. "defaultWorkerTimeout" : 60, // The default amount of attempts to try before failing a job. // Used if the connection and job doesn't define their own. "defaultWorkerMaxAttempts" : 1, // Datasource information for tracking batches. "batchRepositoryProperties" : { "tableName" : "cbq_batches", "datasource" : "", // `datasource` can also be a struct "queryOptions" : {} // The sibling `datasource` property overrides any defined datasource in queryOptions. }, // Flag to turn on logging failed jobs to a database table. "logFailedJobs" : false, // Datasource information for loggin failed jobs. "logFailedJobsProperties" : { "tableName" : "cbq_failed_jobs", "datasource" : "", // `datasource` can also be a struct. "queryOptions" : {} // The sibling `datasource` property overrides any defined datasource in `queryOptions`. }, // Flag to allow restricting Job interceptor execution using a `jobPattern` annotation. "registerJobInterceptorRestrictionAspect" : false }; ``` ### configPath The configPath is a dot-delimited path to your cbq Config Component. By convention, this should be placed in your application's `config/` folder alongside other config files like `ColdBox.cfc` and `WireBox.cfc`. ### registerWorkers This flag is responsible for spinning up Worker Pools when the application starts. If a particular instance of your application should **not** work jobs as well as dispatch them, then this setting should be set to `false`. To make this easy, you can set the `CBQ_REGISTER_WORKERS` environment variable and it will be picked up. (If you override this setting in your own `moduleSettings` it will still take precedence over the environment variable.) ### scaleInterval {% hint style="danger" %} This feature is not implemented yet. {% endhint %} This is the interval in seconds that the scale job is ran in the background. The scale job enables you to scale Worker Pools up or down based on any other factors in your application. **Setting this value to `0` disables the job entirely.** ### defaultWorkerBackoff This setting provides a default backoff value for a Worker Pool. This can still be overridden on a configured Worker Pool in your cbq config file or on an individual Job or `dispatch` call. ### defaultWorkerTimeout This setting provides a default timeout value for a Worker Pool. This can still be overridden on a configured Worker Pool in your cbq config file or on an individual Job or `dispatch` call. ### defaultWorkerMaxAttempts This setting provides a default max attempts value for a Worker Pool. This can still be overridden on a configured Worker Pool in your cbq config file or on an individual Job or `dispatch` call. ### batchRepositoryProperties A struct of configuration properties for a Batch Repository. This is only needed if you dispatch any batches. #### tableName The name of the batch table. The default is `cbq_batches`. #### datasource The datasource to use to interact with the batch table. If no datasource is provided, it uses the default application datasource. A `struct` can also be provided if your CFML engine supports it. #### queryOptions A struct of query options to pass to the `queryExecute` call when interacting with the batch table. If a `datasource` is defined above, it will override any `datasource` key inside the `queryOptions`. ### logFailedJobs This flag will send failed jobs to a configured database table when true. In some systems, this is called a Dead Letter Queue or DLQ. ### logFailedJobsProperties #### tableName The name of the failed jobs table. The default is `cbq_failed_jobs`. #### datasource The datasource to use to interact with the failed jobs table. If no datasource is provided, it uses the default application datasource. A `struct` can also be provided if your CFML engine supports it. #### queryOptions A struct of query options to pass to the `queryExecute` call when interacting with the failed jobs table. If a `datasource` is defined above, it will override any `datasource` key inside the `queryOptions`. ### registerJobInterceptorRestrictionAspect Flag to allow [restricting Job interceptor execution](/3.0.0/interceptors#jobpattern-annotation) using a `jobPattern` annotation. # Config File The cbq config file is where you define Queue Connections as well as Worker Pools. These Connections and Worker Pools can also be added, removed, or modified based on the current environment. When you want to change where your queued Jobs are sent or how they are worked, this is the file you will modify. ## Definitions ### Queue Connections Also referred to as "Connection." A Queue Connection defines where Jobs are serialized and the default settings that apply. ## configure The `configure` method is where the production Queue Connection and Worker Pools definitions are constructed. ### newConnection This command creates a new `QueueConnectionDefinition` builder component. It requires a unique name that will define this Connection and that will be used when defining Worker Pools.
ArgumentsTypeRequiredDefaultDescription
namestringtrueThe unique name for the new Queue Connection.
```cfscript component { function configure() { newConnection( "default" ); } } ``` A `QueueConnectionDefinition` has various methods to configure the Queue Connection. {% content-ref url="/pages/pghfGWdoAmKqeFiwqKtz" %} [Queue Connection](/3.0.0/configuration/config-file/queue-connection) {% endcontent-ref %} ### newWorkerPool This command creates a new `WorkerPoolDefinition` builder component. It requires a unique `name` that will define this Worker Pool and a `connectionName` that points to an already created Connection.
ArgumentsTypeRequiredDefaultDescription
namestringtrueThe unique name for the new Worker Pool.
connectionNamestringfalseA reference to an existing Connection. If not passed in here, the forConnection method must be called to define the Connection.
quantitynumericfalse1The number of workers to spin up for this Worker Pool.
queuesarrayfalse[ * ]An array of queues that this Worker Pool will work. A queue of * refers to all queues. Queues will be worked in the order provided.
forcebooleanfalsefalseIf false, an exception will be thrown if the Worker Pool name has already been registered. If true, the new definition will override the existing definition.
component {

    function configure() {
        newConnection( "default" );
        
        newWorkerPool( "default" ).forConnection( "default" );
    }

}
A `WorkerPoolDefinition` has various methods to configure the Worker Pool. {% content-ref url="/pages/FRcimvpoy6mO5qGgTvgb" %} [Worker Pool](/3.0.0/configuration/config-file/worker-pool) {% endcontent-ref %} ### reset Removes all Connections and Worker Pools ```cfscript getInstance( "Config@cbq" ).reset(); ``` ### Environment Overrides Just as in your `config/ColdBox.cfc` file or in `ModuleConfig.cfc` files, you can add, remove, or modify Queue Connection or Worker Pool definitions per environment. You do this by defining a method on your Config component matching the environment name you want to override. ```cfscript component { function configure() { newConnection( "default" ) .provider( "DBProvider@cbq" ); newWorkerPool( "default" ) .forConnection( "default" ) .quantity( 3 ) .timeout( 15 ); } /** * This method will be called after `configure` * and only if the current environment is `development`. */ function development() { withConnection( "default" ) .provider( "SyncProvider@cbq" ); withWorkerPool( "default" ) .quantity( 1 ) .timeout( 60 ); } } ``` ### withConnection Retrieves an already defined Queue Connection Definition for overriding. All the same methods are available. {% content-ref url="/pages/pghfGWdoAmKqeFiwqKtz" %} [Queue Connection](/3.0.0/configuration/config-file/queue-connection) {% endcontent-ref %} #### delete {% hint style="danger" %} This method is not yet implemented. {% endhint %} ### withWorkerPool Retrieves an already defined Worker Pool Definition for overriding. All the same methods are available. {% content-ref url="/pages/FRcimvpoy6mO5qGgTvgb" %} [Worker Pool](/3.0.0/configuration/config-file/worker-pool) {% endcontent-ref %} #### delete {% hint style="danger" %} This method is not yet implemented. {% endhint %} # Queue Connection A Queue Connection is defined inside your cbq config file using a `QueueConnectionDefinition` builder component. You can create one of these builder components using the [`newConnection`](/3.0.0/configuration/config-file#newconnection) method. ### QueueConnectionDefinition Methods #### provider Sets the provider for the Queue Connection. This can be any valid WireBox mapping and should implement the `IQueueProvider` interface. (There is no need to use the `implements` keyword.)
ArgumentsTypeRequiredDefaultDescription
providerstringtrueA valid WireBox mapping to the desired Provider.
#### setProvider Alias for [`provider`](#provider). #### setProperties Accepts a struct of properties to configure the Queue Connection. The available properties are usually defined by the Queue Provider being used. #### setDefaultQueue Sets the default queue to use for jobs dispatched on this Queue Connection. #### markAsDefault Marks this Queue Connection as the default Queue Connection for jobs dispatched without specifying a Queue Connection. #### setMakeDefault Alias for [`markAsDefault`](#markasdefault). # Worker Pool A Queue Connection is defined inside your cbq config file using a `QueueConnectionDefinition` builder component. You can create one of these builder components using the [`newConnection`](/3.0.0/configuration/config-file#newconnection) method. ### WorkerPoolDefinition Methods #### setName Sets the name for the Worker Pool. Usually not called directly as the `newWorkerPool` method requires a `name`. ```cfscript newWorkerPool( "default" ) .setName( "not-default" ); ``` #### forConnection Sets the name of the associated Connection for the Worker Pool. This must reference an already registered Connection. ```cfscript newConnection( "db" ) .provider( "DBProvider@cbq" ); newWorkerPool( "db-worker" ) .forConnection( "db" ); ``` #### setConnectionName Alias for [forConnection](#forconnection). #### quantity Sets the quantity of workers for this Worker Pool. ```cfscript newWorkerPool( "default" ) .forConnection( "db" ) .quantity( 3 ); ``` #### setQuantity Alias for [quantity](#quantity). #### onQueue The name of a queue to work. The default queue is named `default`. ```cfscript newWorkerPool( "premium-only" ) .forConnection( "db" ) .onQueue( "premium" ); newWorkerPool( "priority" ) .forConnection( "db" ) .onQueue( "priority" ) .quantity( 4 ); newWorkerPool( "default" ) .forConnection( "db" ); // uses `default` queue ``` Some Providers allow for working a priority of queues, such as the `DBProvider`. In these cases, you can pass an array of queues, in priority order, using the asterisk (`*`) as a wildcard character. ```cfscript newWorkerPool( "premium-only" ) .forConnection( "db" ) .onQueue( "premium" ); newWorkerPool( "default" ) .forConnection( "db" ) .onQueue( [ "priority", "*" ] ); ``` {% hint style="danger" %} **Throws:** `cbq.WorkerPool.MultipleQueuesNotSupported` {% endhint %} #### setQueue Alias for [onQueue](#onqueues). #### backoff Sets the backoff time amount, in seconds. ```cfscript newWorkerPool( "default" ) .forConnection( "db" ) .backoff( 30 ); ``` #### setBackoff Alias for [backoff](#backoff). #### timeout Sets the timeout time amount, in seconds. ```cfscript newWorkerPool( "default" ) .forConnection( "db" ) .timeout( 60 ); ``` #### setTimeout Alias for [timeout](#timeout). #### maxAttempts Sets the max number of attempts. ```cfscript newWorkerPool( "default" ) .forConnection( "db" ) .maxAttempts( 5 ); ``` #### setMaxAttempts Alias for [maxAttempts](#maxattempts). # Providers A Queue Provider provides a way to serialize jobs to a Queue Connection and to work jobs off a Queue Connection. The three built-in cbq providers are: * [SyncProvider@cbq](/3.0.0/configuration/providers/syncprovider) * [ColdBoxAsyncProvider@cbq](/3.0.0/configuration/providers/coldboxasyncprovider) * [DBProvider@cbq](/3.0.0/configuration/providers/dbprovider) ### Writing your own Custom Provider All Queue Providers extend the `AbstractQueueProvider` base component. They must implement three abstract methods: ```cfscript /** * Persists a serialized job to the Queue Connection * * @queueName The queue name for the job. * @payload The serialized job string. * @delay The delay (in seconds) before working the job. * @attempts The current attempt number. * * @return AbstractQueueProvider */ public any function push( required string queueName, required string payload, numeric delay = 0, numeric attempts = 0 ); /** * Starts a worker for a Worker Pool on this Queue Connection. * * @pool The Worker Pool that is working this Queue Connection. * * @return A function that when called will stop this worker. */ public function function startWorker( required WorkerPool pool ); /** * Starts any background processes needed for the Worker Pool. * * @pool The Worker Pool that is working this Queue Connection. * * @return AbstractQueueProvider */ public any function listen( required WorkerPool pool ); ``` # SyncProvider {% hint style="danger" %} This provider is not **durable**. (Pending jobs are not saved and may be lost if there are server issues between dispatching the job and working the job.) {% endhint %} {% hint style="danger" %} This provider **does not support multiple queues**. (Only a single queue can be provided to a Worker Pool instance. Additional queues to be worked should be registered as separate Worker Pool instances.) {% endhint %} {% hint style="danger" %} Jobs dispatched from inside a Sync job are executed immediately and recursively. This means a child job failing will cause any parent jobs to also fail. {% endhint %} The SyncProvider runs any jobs dispatched during a request in the same request synchronously. This can be especially useful in development when debugging jobs. If you job would throw an exception, you will see it with any chosen error handler and tools you use to debug local exceptions. # ColdBoxAsyncProvider {% hint style="danger" %} This provider is not **durable**. (Pending jobs are not persisted and may be lost if there are server issues between dispatching the job and working the job.) {% endhint %} {% hint style="danger" %} This provider **does not support multiple queues**. (Only a single queue can be provided to a Worker Pool instance. Additional queues to be worked should be registered as separate Worker Pool instances.) {% endhint %} The ColdBoxAsyncProvider runs any jobs dispatched on a background thread using ColdBox's AsyncManager. Jobs will be worked by the same server that dispatched them. The number of workers specified translates to the number of threads dedicated to working the jobs. # DBProvider {% hint style="success" %} This provider is **durable**. (Pending jobs are persisted. If there are server issues, the jobs will be worked when the issues are resolved. Multiple different servers can dispatch to this Queue Connection and multiple different Worker Pools can work these jobs.) {% endhint %} {% hint style="success" %} This provider **supports multiple queues**. (An array of prioritized queues can be assigned to Worker Pools worked by this provider.) {% endhint %} You may use a database as the backing engine for your Queue Connection using the DBProvider. All database grammars that are [supported by qb](https://qb.ortusbooks.com/v/9.0.0/installation-and-usage) are supported. ### Configuration The DBProvider has three optional arguments. The are presented below with their default values. ```cfscript { "tableName": "cbq_jobs", "datasource": null, "queryOptions": {} } ``` #### tableName The name of the table to use when managing jobs. This table should have the structure provided by the provided database migration file. #### datasource The name of the datasource to use when managing jobs. This overrides any datasource provided in the `queryOptions`. #### queryOptions A struct of options that will be passed to [`queryExecute`](https://cfdocs.org/queryexecute) when managing jobs. ### Jobs Table Structure The `cbq_jobs` table must have a specific structure. It is provided in the form of a migration file called `2000_01_01_000000_create_cbq_jobs_table.cfc`. This migration can be ran via [CommandBox Migrations](https://forgebox.io/view/commandbox-migrations) or [CFMigrations](https://forgebox.io/view/cfmigrations). If you wish, you can generate the table in other ways, so long as it matches the structure provided in the migration file. If you choose to use the migration file in your application, copy the file out to your own migrations folder first. ```cfscript schema.create( "cbq_jobs", function ( t ) { t.bigIncrements( "id" ); t.string( "queue" ); t.longText( "payload" ); t.unsignedTinyInteger( "attempts" ); t.unsignedInteger( "reservedDate" ).nullable(); t.unsignedInteger( "availableDate" ); t.unsignedInteger( "createdDate" ); t.index( "queue" ); } ); ``` # Defining a Job A Job represents both the object to be serialized to a [Connection](/3.0.0/configuration/config-file#connection) to eventually work as well as the code to execute when working a Job. ## Extending from AbstractJob The first step to define a Job is to extend from `AbstractJob`. This provides the necessary helper methods to run a Job through the cbq pipeline. ```cfscript component name="SendWelcomeEmailJob" extends="cbq.models.Jobs.AbstractJob" { } ``` ## Properties Defining properties on your job allow you to see at a glance what data your Job expects when constructing it. These properties and their values will be serialized to a [Connection](/3.0.0/configuration/config-file#connection) when dispatching a Job. {% hint style="info" %} Defining the properties is not strictly necessary, but your future self will thank you when you try to remember what properties your job is using. {% endhint %} ```cfscript component name="SendWelcomeEmailJob" extends="cbq.models.Jobs.AbstractJob" { property name="email"; property name="greeting"; } ``` ## The handle method The `handle` method is called when a Job is worked. Before being called, a Job will be reconstructed with the serialized data from the Connection. This method is the only required method when defining a Job. ```cfscript component name="SendWelcomeEmailJob" extends="cbq.models.Jobs.AbstractJob" { property name="mailService" inject="provider:MailService@cbmailservices"; property name="email"; property name="greeting"; function handle() { variables.mailService.newMail( to = variables.email, from = "noreply@example.com", subject = "Welcome!", type = "html", bodyTokens = { "greeting": variables.greeting "link": getInstance( "coldbox:requestContext" ) .buildLink( "home" ) } ) .setView( "_emails/welcome" ) .send(); } } ``` {% hint style="info" %} Prefer using `provider:` injections or inline `getInstance` calls for logic in your `handle` method. The `Job` component is created both when dispatching and when working your job, so utilizing these tools will reduce unnecessary processing time when dispatching your job. {% endhint %} ## Job Execution Properties A job can define several execution properties on the job itself. If defined, these values override the module, connection, or worker defaults. It can still be overridden using the job methods when creating a job. 
component extends="cbq.models.Jobs.AbstractJob" {

    // Name of the connection to dispatch this job on.
    variables.connection = "db";
    
    // Name of the queue to use for this job.
    variables.queue = "priority";
    
    // Time, in seconds, between job attempts, including the initial attempt.
    variables.backoff = 15;
    
    // Time, in seconds, to let a job run before failing it.
    variables.timeout = 30;
    
    // Max number of attempts before marking a job as failed.
    // Use 0 for unlimited retries.
    variables.maxAttempts = 9;

}
## Lifecycle Methods A Job can define a `before` or `after` method that will be called as part of the Job lifecycle. 
component
    name="SendWelcomeEmailJob"
    extends="cbq.models.Jobs.AbstractJob"
{

    property name="mailService" inject="provider:MailService@cbmailservices";
    
    property name="email";
    property name="greeting";
    
    function handle() {
        variables.mailService.newMail( 
            to = variables.email,
	    from = "noreply@example.com",
	    subject = "Welcome!",
            type = "html",
	    bodyTokens = { 
		"greeting": variables.greeting
		"link": getInstance( "coldbox:requestContext" )
		    .buildLink( "home" )
	    }
        )
        .setView( "_emails/welcome" )
        .send();
    }
    
    function before() {
        log.debug( "About to execute SendWelcomeEmailJob" );
    }
    
    function after() {
        log.debug( "Finished executing SendWelcomeEmailJob" );
    }

}
# Creating a Job After you define your job, you need to create an instance of your job and set the properties before you dispatch it onto your [Queue Connection](/3.0.0/configuration/config-file#connection). You can do this is a few different ways. ## Creating a Job Instance You can create a Job instance using WireBox anywhere in your ColdBox application. ```cfscript component { function index( event, rc, prc ) { var job = getInstance( "SendWelcomeEmailJob" ); } } ``` {% hint style="warning" %} Keep in mind that Job instances are transient. Do not inject a Job instance into a Component or Scope that is not transient. For instance, do not inject a Job instance as a property in a Handler. {% endhint %} ### Setting Job Properties Once you have a Job instance, you can set the data for the particular Job by calling the setter methods. ```cfscript component { function index( event, rc, prc ) { var job = getInstance( "SendWelcomeEmailJob" ); job.setEmail( "john@example.com" ); job.setGreeting( "Welcome!" ); } } ``` ### setProperties You can also set all the properties at once using the `setProperties` method. {% hint style="warning" %} Using this method will overwrite any previously set properties. {% endhint %} ```cfscript component { function index( event, rc, prc ) { var job = getInstance( "SendWelcomeEmailJob" ); job.setProperties( { "email": "john@example.com", "greeting": "Welcome!" } ); } } ``` ### onConnection You can override the connection for a job by calling the `onConnection` method with the desired connection name. ### setConnection See [onConnection](#onconnection). ### onQueue The queue name to use for the job. Queue names can be any string you choose, but to be worked a `WorkerPool` must be defined working the same connection and queue name. ### setQueue See [onQueue](#onqueue). ### setBackoff Sets the amount of time, in seconds, to wait in-between Job attempts — including the initial attempt. ### setDelay An alias for [setBackoff](#setbackoff). ### setTimeout Sets the amount of time, in seconds, to let a Job run on a worker before marking it as a failed attempt. ### setMaxAttempts Sets the maximum number of attempts before a Job is marked as failed. ### chain Sets an array of Jobs to be executed, in order, after this Job successfully executes. ### getMemento getMemento ## [cbq.job()](/3.0.0/cbq-model#job) You can also create a Job instance using the `cbq` model. (This can be injected into your components using the `cbq@cbq` DSL.) {% hint style="success" %} The cbq model is a singleton, so feel free to inject it into any component. {% endhint %} ```cfscript component { property name="cbq" inject="cbq@cbq"; function index( event, rc, prc ) { var job = cbq.job( job = "SendWelcomeEmailJob", properties = { "email": "john@example.com", "greeting": "Welcome!" }, chain = [], queue = "priority", connection = "db", backoff = 10, timeout = 30, maxAttempts = 3 ); } } ``` # Dispatching a Job ## Dispatching Creating a Job is all well and good, but the Job will not be executed until it is dispatched on to a [Queue Connection](/3.0.0/configuration/config-file#connection). If you have a Job instance, you can dispatch it by calling the `dispatch` method. ```cfscript component { function index( event, rc, prc ) { getInstance( "SendWelcomeEmailJob" ); .setEmail( "john@example.com" ); .setGreeting( "Welcome!" ) .dispatch(); } } ``` You can also create and dispatch a job in the same call using the `cbq.dispatch()` method. ```cfscript component { property name="cbq" inject="cbq@cbq"; function index( event, rc, prc ) { cbq.dispatch( job = "SendWelcomeEmailJob", properties = { "email": "john@example.com", "greeting": "Welcome!" }, chain = [], queue = "priority", connection = "db", backoff = 10, timeout = 30, maxAttempts = 3 ); } } ``` ## Interception Points ### onCBQJobAdded This is fired before serializing a Job and sending it to a connection. The data includes the `job` to be serialized and dispatched and the connection the Job is being dispatched on. ```cfscript variables.interceptorService.announce( "onCBQJobAdded", { "job" : job, "connection" : connection } ); ``` # Working a Job Dispatched jobs will get picked up by Worker Pools set up for the same connection and queue. They are picked up in order by created date (FIFO). If you have a valid Worker Pool, then you don't need to do anything else to have your job picked up. If you want to see all the details of cbq dispatching and marshalling jobs, enable `debug` logging in LogBox for `cbq`. When a Job is worked, the Job instance is created via WireBox and the memento hydrated into the Job component. Then the `handle` method is called. When a Job is completed, the Job will be removed from the persistent storage of the Queue Provider. If the Job attempt fails, it will be sent back to the Queue Provider to be retried, up to the `maxAttempts` amount. Once a Job has failed the `maxAttempts` amount it will be removed and marked as failed. If you have `logFailedJobs` enabled, it will be sent to the [failed jobs table](/3.0.0/jobs/failed-jobs#logging-failed-jobs). ## Inside \`handle\` ### release You can choose to manually release a Job back to a queue with an optional delay (in seconds) using the `release` function. This is useful when you need to delay processing for a Job, perhaps due to rate limiting or licensing constraints. {% hint style="info" %} Calling `release` does not stop processing the `handle` method, so make sure you `return` if you don't want to keep executing your `handle` method. {% endhint %} ```cfscript component { function handle() { this.release( 60 ); // in seconds return; } } ``` ## Interception Points cbq fires a number of interception points during a Job's lifecycle. ### onCBQJobMarshalled This is fired before the `handle` method is called on a Job. The data includes the `job` to be executed. ```cfscript variables.interceptorService.announce( "onCBQJobMarshalled", { "job" : job } ); ``` ### onCBQJobComplete This is fired after a Job completes successfully. The data includes the `job` to be executed as potentially a `result` returned from the `Job`'s handle method. ```cfscript variables.interceptorService.announce( "onCBQJobComplete", { "job" : job, "result" : isNull( result ) ? javacast( "null", "" ) : result } ); ``` ## Tips ### Need up to date data? Could your data change between when you dispatched a job and when you work the job? If so, consider using the key as a property and fetching the data from inside the job. ### Need to use historical data? In some cases, you want to work with the data that existed at the time the Job was dispatched. In these cases, make sure to include all the needed data in the Job instance. ### Check if you need to work the Job still Things might have changed since the Job was dispatched. Most Queue Providers do not support querying the current Jobs queue or interacting with it in advance, so check in your `dispatch` method if the Job still needs to be worked. ### Need to try the Job later? In some cases, you need to retry your Job later. In these cases, use the `release` method to send the Job back to the queue. It also takes an optional `delay` which sets the `backoff` time for the next Job attempt. You can combine this with setting the `maxAttempts` of a Job to `0` to retry it indefinitely. Remember that a Job is only reattempted if it fails. If you decide that a Job is finished, just `return`. ### Jobs can dispatch other Jobs You can dispatch other Jobs from your Job. This can be different Jobs or even a similar instance of the same Job. ## Why isn't my Job getting picked up? To work a cbq job, you need a [defined Worker Pool in your cbq Config file.](/3.0.0/configuration/config-file#newworkerpool) In order to work a Job, the Worker Pool must: 1. Be working the same [connection](/3.0.0/configuration/config-file/worker-pool#forconnection) as the Job. 2. Be working the same [queue](/3.0.0/configuration/config-file/worker-pool#onqueues) as the Job. 3. Have at least one active worker (defined with [`setQuantity`](/3.0.0/configuration/config-file/worker-pool#setquantity)). If these requirements are not met, the job will stay in the queue storage, unable to get picked up. # Failed Jobs ## Failed Attempt vs Failed Job An important distinction to make is between failed Job attempts and failed Jobs. A Job is attempted up to the defined `maxAttempts`. If a Job fails all of its `maxAttempts` then it is marked as a failed Job and removed from the queue. Otherwise, the Job is dispatched back to the queue with the current execution count increased. ## LogBox Failed Job attempts are logged to LogBox. You will see the exception as well as the serialized Job memento in the `extraInfo`. ## Interceptors ### onCBQJobException This interception point is fired for every **Job Attempt** failure. The data includes the `job` component and the `exception`. ```cfscript variables.interceptorService.announce( "onCBQJobException", { "job" : job, "exception" : e } ); ``` ### onCBQJobFailed This interception point is fired when the Job is marked as failed — when the Job has failed all attempts up to the configured `maxAttempts`. The data includes the `job` component and the `exception`. ```cfscript variables.interceptorService.announce( "onCBQJobFailed", { "job" : job, "exception" : e } ); ``` ## onFailure Job Method You can define an `onFailure` method on your Job component. It will be called with the `exception`. ```cfscript component extends="cbq.models.Jobs.AbstractJob" { function handle() { // ... } function onFailure( e ) { log.error( "Bad things happened: #e.message#" ); } } ``` ## Logging Failed Jobs ## Failed Jobs Table cbq includes an interceptor to log failed jobs to a database. You can enable this in your module settings: ```cfscript moduleSettings = { "cbq": { // Flag to turn on logging failed jobs to a database table. "logFailedJobs" : false, // Datasource information for loggin failed jobs. "logFailedJobsProperties" : { "tableName" : "cbq_failed_jobs", "datasource" : "", // `datasource` can also be a struct. "queryOptions" : {} // The sibling `datasource` property overrides any defined datasource in `queryOptions`. } } }; ``` Also included is a migration file to add the needed table to your database. You can find it in `resources/database/migrations/2000_01_01_000002_create_cbq_failed_jobs_table.cfc`. It is also included below. ```cfscript component { function up( schema ) { schema.create( "cbq_failed_jobs", function ( t ) { t.bigIncrements( "id" ); t.string( "connection" ); t.string( "queue" ); t.string( "mapping" ); t.longText( "memento" ); t.longText( "properties" ); t.string( "exceptionType" ).nullable(); t.string( "exceptionMessage" ); t.string( "exceptionDetail" ).nullable(); t.longText( "exceptionExtendedInfo" ).nullable(); t.longText( "exceptionStackTrace" ); t.longText( "exception" ); t.unsignedInteger( "failedDate" ); } ); } function down( schema ) { schema.dropIfExists( "cbq_failed_jobs" ); } } ``` {% code title="MySQL" %} ```sql CREATE TABLE `cbq_failed_jobs` ( `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT, `connection` NVARCHAR(255) NOT NULL, `queue` NVARCHAR(255) NOT NULL, `mapping` NVARCHAR(255) NOT NULL, `memento` LONGTEXT NOT NULL, `properties` LONGTEXT NOT NULL, `exceptionType` NVARCHAR(255), `exceptionMessage` NVARCHAR(255) NOT NULL, `exceptionDetail` NVARCHAR(255), `exceptionExtendedInfo` LONGTEXT, `exceptionStackTrace` LONGTEXT NOT NULL, `exception` LONGTEXT NOT NULL, `failedDate` INT UNSIGNED NOT NULL ) ``` {% endcode %} {% code title="SQL Server" %} ```sql CREATE TABLE [dbo].[cbq_failed_jobs]( [id] BIGINT NOT NULL IDENTITY, [connection] NVARCHAR(255) NOT NULL, [queue] NVARCHAR(255) NOT NULL, [mapping] NVARCHAR(255) NOT NULL, [memento] NVARCHAR(MAX) NOT NULL, [properties] NVARCHAR(MAX) NOT NULL, [exceptionType] NVARCHAR(255), [exceptionMessage] NVARCHAR(255) NOT NULL, [exceptionDetail] NVARCHAR(255), [exceptionExtendedInfo] NVARCHAR(MAX), [exceptionStackTrace] NVARCHAR(MAX) NOT NULL, [exception] NVARCHAR(MAX) NOT NULL, [failedDate] INT NOT NULL ) ``` {% endcode %} ## Retrying a Failed Job ## Configuring Job Backoff # Chained Jobs Job chains mean that after a Job completes successfully it dispatches the next Job in the chain. If a Job fails, no more Jobs in the chain are dispatched. ## Dispatching Jobs from another Job One way to make a Job chain is to dispatch a Job from inside another Job. This gives you consistency in your Job chain, and a logical code path to follow. ```cfscript // FulfillOrderJob.cfc component extends="cbq.models.Jobs.AbstractJob" { function handle() { var productId = getProperties().productId; processPayment( productId ); getInstance( "SendProductLinkEmail" ) .onConnection( "fulfillment" ) .setProperties( { "productId" : productId, "userId" : getProperties().userId } ) .dispatch(); } } ``` ## Creating a Chain Another way to make a Job chain is to create is when dispatching the Job. This gives you flexibility to compose Job chains at runtime. ```cfscript // handlers/Main.cfc component { property name="cbq" inject="cbq@cbq"; function create() { cbq.chain( [ cbq.job( "FulfillOrderJob", { "productId": rc.productId } ), cbq.job( job = "SendProductLinkEmail", properties = { "productId": rc.productId, "userId": auth().getUserId() }, connection = "fulfillment" ), ] ).dispatch(); } } ``` ## Dispatching Jobs on Failure Utilizing the `onFailure` method of a Job, we can dispatch another Job if our Job fails. ```cfscript // FulfillOrderJob.cfc component extends="cbq.models.Jobs.AbstractJob" { function handle() { // ... } function onFailure( e ) { getInstance( "SendOrderProblemEmail" ) .onConnection( "fulfillment" ) .setProperties( { "productId" : productId, "userId" : getProperties().userId, "message" : e.message } ) .dispatch(); } } ``` # Batched Jobs cbq allows for tracking jobs as part of a batch. The main use case for batched jobs is to dispatch additional jobs when a batch has completed successfully, completed with failures, or completed in any fashion. (Think `try`, `catch`, `finally`.) ## Setup To utilize batches, you first need to set up a Batch repository. Batches are tracked separate from jobs. cbq utilizes a database repository to track batches regardless of what provider your Job's connection uses. A database migration is provided in `resources/database/migrations/2000_01_01_000001_create_cbq_batches_table.cfc`. You can copy this to your own project to run via [cfmigrations](https://forgebox.io/view/cfmigrations) or you can reference the migration or SQL scripts [below](#migrations-and-sql-scripts). You can customize the batch repository using the `batchRepositoryProperties` of cbq's `moduleSettings`. This is a struct with two properties you can set: #### tableName This defaults to `cbq_batches`. You can set this to any unique table name in your datasource. #### queryOptions This is a struct of options that will be passed to `queryExecute`. The most common use case for this property is to specify a specific datasource to use for the batches table. This defaults to an empty struct (`{}`). ## Defining Batches To create a batch, you can get an instance of `PendingBatch@cbq` from WireBox or use the [`cbq.batch()`](/3.0.0/cbq-model#batch) helper method. Once you have a `PendingBatch` instance, you can add jobs to the batch using the `add` method. ### add Adds a single Job or an array of Jobs to a PendingBatch.
ArgumentsTypeRequiredDefaultDescription
jobstring | Job | array<Job>trueThe Job WireBox id, Job instance, or array of Job instances to add to the PendingBatch.
propertiesStructfalse{}A struct of properties for the Job instance. (Only used when providing a Job WireBox id.)
chainarray<Job>false[]An array of Jobs to chain after this Job. (Only used when providing a Job WireBox id.)
queuestringfalseThe queue the Job belongs to. (Only used when providing a Job WireBox id.)
connectionstringfalseThe Connection to dispatch the Job on. (Only used when providing a Job WireBox id.)
backoffnumericfalseThe amount of time, in seconds, to wait between attempting Jobs. (Only used when providing a Job WireBox id.)
timeoutnumericfalseThe amount of time, in seconds, to wait before treating a Job as erroring. (Only used when providing a Job WireBox id.)
maxAttemptsnumericfalseThe maximum amount of attempts of a Job before treating the Job as failed. (Only used when providing a Job WireBox id.)
**Return:** The PendingBatch to be dispatched. ```cfscript batch .add( cbq.job( "ImportCsvJob", { "start": 1, "end": 100 } ) ) .add( "ImportCsvJob", { "start": 101, "end": 200 } ) .add( [ cbq.job( "ImportCsvJob", { "start": 201, "end": 300 } ), cbq.job( "ImportCsvJob", { "start": 301, "end": 400 } ), cbq.job( "ImportCsvJob", { "start": 401, "end": 500 } ) ] ); ``` The primary use case of batches is to dispatch lifecycle jobs when the batch has completed and if the batch completed successfully or completed with failures. These jobs can be configured using the `then`, `catch`, and `finally` methods. ### then Defines a Job to be dispatched when all the jobs in the batch finishes successfully.
ArgumentsTypeRequiredDefaultDescription
jobstring | JobtrueThe Job WireBox id or Job instance to execute if all the jobs in the Batch complete successfully.
propertiesStructfalse{}A struct of properties for the Job instance. (Only used when providing a Job WireBox id.)
chainarray<Job>false[]An array of Jobs to chain after this Job. (Only used when providing a Job WireBox id.)
queuestringfalseThe queue the Job belongs to. (Only used when providing a Job WireBox id.)
connectionstringfalseThe Connection to dispatch the Job on. (Only used when providing a Job WireBox id.)
backoffnumericfalseThe amount of time, in seconds, to wait between attempting Jobs. (Only used when providing a Job WireBox id.)
timeoutnumericfalseThe amount of time, in seconds, to wait before treating a Job as erroring. (Only used when providing a Job WireBox id.)
maxAttemptsnumericfalseThe maximum amount of attempts of a Job before treating the Job as failed. (Only used when providing a Job WireBox id.)
**Return:** The `PendingBatch` instance. ```cfscript cbq.batch( [ cbq.job( "ImportCsvJob", { "start": 1, "end": 100 } ), cbq.job( "ImportCsvJob", { "start": 101, "end": 200 } ), cbq.job( "ImportCsvJob", { "start": 201, "end": 300 } ), cbq.job( "ImportCsvJob", { "start": 301, "end": 400 } ), cbq.job( "ImportCsvJob", { "start": 401, "end": 500 } ) ] ) .then( cbq.job( "ImportCsvSuccessfulJob" ) ); ``` ### catch Defines a Job to be dispatched the first time a job in the Batch fails.
ArgumentsTypeRequiredDefaultDescription
jobstring | JobtrueThe Job WireBox id or Job instance to execute the first time a job in the Batch fails.
propertiesStructfalse{}A struct of properties for the Job instance. (Only used when providing a Job WireBox id.)
chainarray<Job>false[]An array of Jobs to chain after this Job. (Only used when providing a Job WireBox id.)
queuestringfalseThe queue the Job belongs to. (Only used when providing a Job WireBox id.)
connectionstringfalseThe Connection to dispatch the Job on. (Only used when providing a Job WireBox id.)
backoffnumericfalseThe amount of time, in seconds, to wait between attempting Jobs. (Only used when providing a Job WireBox id.)
timeoutnumericfalseThe amount of time, in seconds, to wait before treating a Job as erroring. (Only used when providing a Job WireBox id.)
maxAttemptsnumericfalseThe maximum amount of attempts of a Job before treating the Job as failed. (Only used when providing a Job WireBox id.)
**Return:** The `PendingBatch` instance. ```cfscript cbq.batch( [ cbq.job( "ImportCsvJob", { "start": 1, "end": 100 } ), cbq.job( "ImportCsvJob", { "start": 101, "end": 200 } ), cbq.job( "ImportCsvJob", { "start": 201, "end": 300 } ), cbq.job( "ImportCsvJob", { "start": 301, "end": 400 } ), cbq.job( "ImportCsvJob", { "start": 401, "end": 500 } ) ] ) .catch( cbq.job( "ImportCsvFailedJob" ) ); ``` ### finally Defines a Job to be dispatched after all the jobs in the Batch have executed successfully or failed.
ArgumentsTypeRequiredDefaultDescription
jobstring | JobtrueThe Job WireBox id or Job instance to execute after all the jobs in the Batch have executed successfully or failed.
propertiesStructfalse{}A struct of properties for the Job instance. (Only used when providing a Job WireBox id.)
chainarray<Job>false[]An array of Jobs to chain after this Job. (Only used when providing a Job WireBox id.)
queuestringfalseThe queue the Job belongs to. (Only used when providing a Job WireBox id.)
connectionstringfalseThe Connection to dispatch the Job on. (Only used when providing a Job WireBox id.)
backoffnumericfalseThe amount of time, in seconds, to wait between attempting Jobs. (Only used when providing a Job WireBox id.)
timeoutnumericfalseThe amount of time, in seconds, to wait before treating a Job as erroring. (Only used when providing a Job WireBox id.)
maxAttemptsnumericfalseThe maximum amount of attempts of a Job before treating the Job as failed. (Only used when providing a Job WireBox id.)
**Return:** The `PendingBatch` instance. ```cfscript cbq.batch( [ cbq.job( "ImportCsvJob", { "start": 1, "end": 100 } ), cbq.job( "ImportCsvJob", { "start": 101, "end": 200 } ), cbq.job( "ImportCsvJob", { "start": 201, "end": 300 } ), cbq.job( "ImportCsvJob", { "start": 301, "end": 400 } ), cbq.job( "ImportCsvJob", { "start": 401, "end": 500 } ) ] ) .finally( cbq.job( "ImportCsvCompletedJob" ) ); ``` ## Dispatching Batches A `PendingBatch` must be dispatched before any of the jobs contained in it are dispatched. This is done using the `dispatch` method on the `PendingBatch`. ### dispatch Dispatches a PendingBatch and all the Jobs it contains.
ArgumentsTypeRequiredDefaultDescription
No arguments
**Return:** A `Batch` instance created from this `PendingBatch`. ## Interacting with Batches ## Migrations and SQL Scripts ### cfmigrations ```cfscript schema.create( "cbq_batches", function ( t ) { t.string( "id" ).primaryKey(); t.string( "name" ); t.unsignedInteger( "totalJobs" ); t.unsignedInteger( "pendingJobs" ); t.unsignedInteger( "failedJobs" ); t.text( "failedJobIds" ); // JSON column t.text( "options" ).nullable(); // JSON column t.datetime( "createdDate" ); t.datetime( "cancelledDate" ).nullable(); t.datetime( "completedDate" ).nullable(); } ); ``` ### MySQL ```sql CREATE TABLE `cbq_batches` ( `id` VARCHAR(255) NOT NULL, `name` VARCHAR(255) NOT NULL, `totalJobs` INTEGER UNSIGNED NOT NULL, `pendingJobs` INTEGER UNSIGNED NOT NULL, `failedJobs` INTEGER UNSIGNED NOT NULL, `failedJobIds` TEXT NOT NULL, `options` TEXT, `createdDate` DATETIME NOT NULL, `cancelledDate` DATETIME, `completedDate` DATETIME, CONSTRAINT `pk_cbq_batches_id` PRIMARY KEY (`id`) ) ``` ### SQL Server ```sql CREATE TABLE [cbq_batches] ( [id] VARCHAR(255) NOT NULL, [name] VARCHAR(255) NOT NULL, [totalJobs] INTEGER NOT NULL, [pendingJobs] INTEGER NOT NULL, [failedJobs] INTEGER NOT NULL, [failedJobIds] VARCHAR(MAX) NOT NULL, [options] VARCHAR(MAX), [createdDate] DATETIME2 NOT NULL, [cancelledDate] DATETIME2, [completedDate] DATETIME2, CONSTRAINT [pk_cbq_batches_id] PRIMARY KEY ([id]) ) ``` ### Postgres ```sql CREATE TABLE "cbq_batches" ( "id" VARCHAR(255) NOT NULL, "name" VARCHAR(255) NOT NULL, "totalJobs" INTEGER NOT NULL, "pendingJobs" INTEGER NOT NULL, "failedJobs" INTEGER NOT NULL, "failedJobIds" TEXT NOT NULL, "options" TEXT, "createdDate" TIMESTAMP NOT NULL, "cancelledDate" TIMESTAMP, "completedDate" TIMESTAMP, CONSTRAINT "pk_cbq_batches_id" PRIMARY KEY ("id") ) ``` ### Oracle ```sql CREATE TABLE "CBQ_BATCHES" ( "ID" VARCHAR2(255) NOT NULL, "NAME" VARCHAR2(255) NOT NULL, "TOTALJOBS" NUMBER(10, 0) NOT NULL, "PENDINGJOBS" NUMBER(10, 0) NOT NULL, "FAILEDJOBS" NUMBER(10, 0) NOT NULL, "FAILEDJOBIDS" CLOB NOT NULL, "OPTIONS" CLOB, "CREATEDDATE" DATE NOT NULL, "CANCELLEDDATE" DATE, "COMPLETEDDATE" DATE, CONSTRAINT "PK_CBQ_BATCHES_ID" PRIMARY KEY ("ID") ) ``` ### SQLite ```sql CREATE TABLE "cbq_batches" ( "id" TEXT NOT NULL, "name" TEXT NOT NULL, "totalJobs" INTEGER NOT NULL, "pendingJobs" INTEGER NOT NULL, "failedJobs" INTEGER NOT NULL, "failedJobIds" TEXT NOT NULL, "options" TEXT, "createdDate" DATETIME NOT NULL, "cancelledDate" DATETIME, "completedDate" DATETIME, PRIMARY KEY ("id") ) ``` # cbq Model This model is provided to make certain tasks easier when defining and dispatching jobs, chains, and batches. ### dispatch Dispatches a job or chain of jobs.
ArgumentsTypeRequiredDefaultDescription
jobstring | Job | array<Job>trueA Job instance or a WireBox ID of a Job instance. If the WireBox ID doesn't exist, a NonExecutableJob instance will be used instead. This allows you to dispatch a Job from a server where that Job component is not defined. If an array is passed, a Job Chain is created and dispatched.
propertiesstructfalse{}A struct of properties for the Job instance.
chainarray<Job>false[]An array of Jobs to chain after this Job.
queuestringfalseThe queue the Job belongs to.
connectionstringfalseThe Connection to dispatch the Job on.
backoffnumericfalseThe amount of time, in seconds, to wait between attempting Jobs.
timeoutnumericfalseThe amount of time, in seconds, to wait before treating a Job as erroring.
maxAttemptsnumericfalseThe maximum amount of attempts of a Job before treating the Job as failed.
**Return:** The dispatched Job instance. ```cfscript cbq.dispatch( job = "SendWelcomeEmailJob", properties = { "body": "first body" }, queue = "default" ); ``` ### job Creates a job or chain of jobs to be dispatched.
ArgumentsTypeRequiredDefaultDescription
jobstring | Job | array<Job>trueA Job instance or a WireBox ID of a Job instance. If the WireBox ID doesn't exist, a NonExecutableJob instance will be used instead. This allows you to dispatch a Job from a server where that Job component is not defined. If an array is passed, a Job Chain is created and dispatched.
propertiesstructfalse{}A struct of properties for the Job instance.
chainarray<Job>false[]An array of Jobs to chain after this Job.
queuestringfalseThe queue the Job belongs to.
connectionstringfalseThe Connection to dispatch the Job on.
backoffnumericfalseThe amount of time, in seconds, to wait between attempting Jobs.
timeoutnumericfalseThe amount of time, in seconds, to wait before treating a Job as erroring.
maxAttemptsnumericfalseThe maximum amount of attempts of a Job before treating the Job as failed.
**Return:** The new Job instance. ```cfscript cbq.job( "SendWelcomeEmailJob" ) .setProperties( { "body": "first body" } ) .onQueue( "default" ) .dispatch(); ``` ### chain Creates a chain of jobs to be ran. Alias for calling `firstJob.chain( otherJobs )`.
ArgumentsTypeRequiredDefaultDescription
jobsarray<Job>trueThe array of jobs to run in order in a chain.
**Return:** The first job of the chain with the chained jobs configured to be dispatched. ```cfscript cbq.chain( [ cbq.job( "SendWelcomeEmailJob", { "body": "One" }, [], "default" ), cbq.job( "SendWelcomeEmailJob", { "body": "Two" }, [], "default", "sync" ), cbq.job( "SendWelcomeEmailJob", { "body": "Three" }, [], "default" ) ] ) .dispatch() ``` ### batch Creates a PendingBatch from the Jobs provided. {% hint style="warning" %} To use batches, you must first configure a `BatchRepository`. Learn more in the [Batched Jobs documentation](/3.0.0/jobs/batched-jobs). {% endhint %}
ArgumentsTypeRequiredDefaultDescription
jobsarray<Job>trueAn array of jobs to batch together.
**Return:** The PendingBatch to be dispatched. ```cfscript var batch = cbq .batch( [ cbq.job( "ImportCsvJob", { "start": 1, "end": 100 } ), cbq.job( "ImportCsvJob", { "start": 101, "end": 200 } ), cbq.job( "ImportCsvJob", { "start": 201, "end": 300 } ), cbq.job( "ImportCsvJob", { "start": 301, "end": 400 } ), cbq.job( "ImportCsvJob", { "start": 401, "end": 500 } ) ] ) .then( cbq.job( "ImportCsvSuccessfulJob" ) ) .catch( cbq.job( "ImportCsvFailedJob" ) ) .finally( cbq.job( "ImportCsvCompletedJob" ) ) .dispatch(); ``` # Interceptors ## Interception Points cbq announces the following interception points: ### `onCBQJobAdded` This is called when a Job is dispatched to a Queue. ### `onCBQJobMarshalled` This is called when a Job is pulled off the Queue to work. ### `onCBQJobComplete` This is called when a Job successfully finishes executing. ### `onCBQJobException` This is called when encountering an exception when handling a Job. ### `onCBQJobFailed` This is called when a Job is considered failed, after exhausting its `maxAttempts`. ## JobPattern Annotation {% hint style="warning" %} To use the `jobPattern` annotation, you must have enabled the `registerJobInterceptorRestrictionAspect` setting in your [module settings](/3.0.0/configuration/module-settings). {% endhint %} Interceptors listening on the cbq interception points listed above can optionally restrict execution to certain Jobs using a `jobPattern` annotation (similar to the `eventPattern` annotation [in ColdBox](https://coldbox.ortusbooks.com/the-basics/interceptors/restricting-execution)). ```cfscript component { function onCBQJobMarshalled( event, data ) jobPattern="SendWelcomeEmailJob" { // check if we've hit the email send limits for the month } } ``` This annotation accepts a regex string to check against the Job full name: ```cfscript component { function onCBQJobMarshalled( event, data ) jobPattern="^.*Email.*$" { // check if we've hit the email send limits for the month } } ``` {% hint style="info" %} Jobs that do not match the interception point will send a notice to the `debug` log, if that is turned on for `JobInterceptorRestriction`. {% endhint %} # Contributing # Contributors # Prior Art Initial ideas behind cbq were derived from [Laravel](https://laravel.com). # Dedication # Home ## A protocol-based queueing system for ColdBox Queues allow you to push work to the background, schedule work to be done later, or even process work on many different machines. It runs on a provider-based system allowing a unified API to talk with many different queue backends. ### Where to go next? * [Installation](/4.0.0/getting-started/installation) * [Walkthrough](/4.0.0/getting-started/walkthrough) * CFCasts Series (Coming Soon) * API Docs # What's New? ## v4.0.0 ### Breaking Change On Batches, the `finally` and `catch` methods have been deprecated in order to support Adobe ColdFusion. `finally` -> `onComplete` `catch` -> `onFailure` If you are running on Lucee or BoxLang, the old methods names will still work, but they may be removed in a future version. We recommend migrating to the new method names. ## v3.0.2 * Add error logging around logging failed jobs. ## v3.0.1 * **DBProvider:** Fix releasing job timeouts using the wrong value ## v3.0.0 * The `failedDate` column now uses a Unix timestamp as the column type. This avoids any timezone issues and aligns more closely with the `cbq_jobs` table used by the `DBProvider`. * Allow worker pools to finish currently running jobs, up to a configurable timeout. * Add optional clean-up tasks for completed or failed jobs, failed job logs, and completed or cancelled batches. * **DBProvider:** Improve database locking to avoid duplicate runs of the same job. * Fixes unwrapping an optional in a log message. ## v2.1.0 Add back ability to [work on multiple queues](/4.0.0/configuration/config-file/worker-pool#onqueue) on a per-Provider basis. Currently only the `DBProvider` supports it. Add support for `before` and `after` [lifecycle methods](/4.0.0/jobs/defining-a-job#lifecycle-methods) on a Job instance. Add ability to [restrict interceptor execution ](/4.0.0/interceptors#jobpattern-annotation)with a `jobPattern` annotation. (This is similar to the `eventPattern` annotation [provided by ColdBox](https://coldbox.ortusbooks.com/the-basics/interceptors/restricting-execution).) ## v2.0.5 **DBProvider**: Disable `forceRun` because it is causing ColdBox Futures to lose mappings. ## v**2.0.4** Reload module mappings in an attempt to work around ColdBox Async losing them. ## **v2.0.3** **SyncProvider:** Add pool to releaseJob call ## v2.0.2 Fix moduleSettings missing a queryOptions key for failed jobs ## v2.0.1 ColdBoxAsyncProvider now correctly respects Worker Pool conifguration, including queues. ## v2.0.0 ### BREAKING CHANGES #### Worker Pools can only define a single queue to work In order to work with new Queue Providers, the Worker Pools need to be updated to only work a specific queue. This is because many future Queue Providers like RabbitMQ and Amazon SQS only support listening to a single queue in a consumer. If you previously had multiple queues defined in a Worker Pool, you will need to define multiple Worker Pool instances, one for each of the queues. ```cfscript // Old newWorkerPool( "default" ) .forConnection( "default" ) .onQueues( [ "priority", "default" ] ); // New newWorkerPool( "default" ) .forConnection( "default" ) .onQueue( "priority" ); newWorkerPool( "default" ) .forConnection( "default" ) .onQueue( "default" ); ``` Notice that the method has been renamed from `onQueues` to `onQueue`. Additionally, there are no more wildcard queues. Every queue you publish to must have a WorkerPool defined in order for that Job to be worked. Finally, queue priorities are defined by the number of workers (`quantity`) you define for the WorkerPool. WorkerPools can no longer share workers across queues. # Upgrade Guide ## v3 to v4 On Batches, the `finally` and `catch` methods have been deprecated in order to support Adobe ColdFusion. `finally` -> `onComplete` `catch` -> `onFailure` If you are running on Lucee or BoxLang, the old methods names will still work, but they may be removed in a future version. We recommend migrating to the new method names. ## v2 to v3 In v3, the `cbq_failed_jobs` table migrates the `failedDate` column type from a timestamp to a unix timestamp. A migration file is included in `resources/database/migrations/2000_01_01_000006_use_unix_timestamp_for_failed_job_log_failedDate.cfc`. To run this migration, your failed jobs table will need to be empty. Alternatively, you can write your own migration that converts the timestamp to a unix timestamp. (The logic is different for each database grammar.) # Installation ## Requirements cbq requires the following: * Adobe ColdFusion (ACF) 2018+ **OR** Lucee 5+ * ColdBox 6+ The different [Queue Providers](/4.0.0/configuration/providers) each have their own requirements that are listed on their individual pages. ## Install via ForgeBox with CommandBox cbq is installed via [ForgeBox](https://forgebox.io) with [CommandBox](https://www.ortussolutions.com/products/commandbox). You can install the latest version using the command: ```shell install cbq ``` ## Load Java Libraries When using batches, cbq utilizes additional Java libraries included in the `lib/` folder. These need to be added to your Application's `javaSettings` in `Application.cfc`. ```cfscript // Java Integration this.javaSettings = { loadPaths: [ expandPath( "./modules/cbq/lib" ) ], loadColdFusionClassPath: true, reloadOnChange: false }; ``` {% hint style="info" %} Feel free to use mappings to point to the cbq path, if needed. {% endhint %} ## Additional Provider Installation Steps The [Queue Provider ](/4.0.0/configuration/providers)you use may have additional installation steps. Check out the individual provider pages for more details. # Walkthrough ## Install cbq Refer to the [Installation](/4.0.0/getting-started/installation) section for general installation instructions as well as instructions for your specific [provider](/4.0.0/configuration/providers). ## Create a Queue Connection Open up your `config/cbq.cfc` file and create your first Queue Connection: ```cfscript component { function configure() { newConnection( "default" ) .setProvider( "ColdBoxAsyncProvider@cbq" ); } } ``` ## Create a Worker Pool Next, create a Worker Pool for your new Queue Connection. This allows our application to work the Jobs we will dispatch. ```cfscript component { function configure() { newConnection( "default" ) .setProvider( "ColdBoxAsyncProvider@cbq" ); newWorkerPool( "default" ) .forConnection( "default" ); } } ``` ## Define your first Job A Job is a CFC that extends `cbq.models.Jobs.AbstractJob`. It can live anywhere in your application. ```cfscript // models/jobs/emails/SendWelcomeEmailJob.cfc component extends="cbq.models.Jobs.AbstractJob" { property name="userId"; function handle() { log.info( "Sending a Welcome email to User ###getUserId()#" ); /* sample code var user = getInstance( "User" ).findOrFail( getUserId() ); getInstance( "MailService@cbmailservices" ) .newMail( from = "no-reply@example.com", to = user.getEmail(), subject = "Welcome!", type = "html" ) .setView( "/_emails/users/welcome" ) .setBodyTokens( { "firstName" : user.getFirstName(), "lastName" : user.getLastName() } ) .send(); */ } } ``` ## Create an instance of your Job You can create an instance of your Job anywhere in your code — handlers, services, models, etc. Populate it with the specific data needed for this instance. ```cfscript var job = getInstance( "SendWelcomeEmailJob" ); job.setUserId( newUser.getId() ); ``` ## Dispatch your Job Once your Job is created and configured, `dispatch` it to the Queue Connection. ```cfscript job.dispatch(); ``` ## Watch your job get executed Check out LogBox to see your Job being executed. Congratulations! You've dispatched your first background Job using cbq! # Module Settings ## Full Module Settings ```cfscript settings = { // The path the custom config file to register connections and worker pools "configPath" : "config.cbq", // Flag if workers should be registered. // If your application only pushes to the queues, you can set this to `false`. "registerWorkers" : getSystemSetting( "CBQ_REGISTER_WORKERS", true ), // The interval to poll for changes to the worker pool scaling. // Defaults to 0 which turns off the scheduled scaling feature. "scaleInterval" : 0, // The default amount of time, in seconds, to delay a job. // Used if the connection and job doesn't define their own. "defaultWorkerBackoff" : 0, // The default amount of time, in seconds, to wait before timing out a job. // Used if the connection and job doesn't define their own. "defaultWorkerTimeout" : 60, // The default amount of attempts to try before failing a job. // Used if the connection and job doesn't define their own. "defaultWorkerMaxAttempts" : 1, // Datasource information for tracking batches. "batchRepositoryProperties" : { "tableName" : "cbq_batches", "datasource" : "", // `datasource` can also be a struct "queryOptions" : {} // The sibling `datasource` property overrides any defined datasource in queryOptions. }, // Flag to turn on logging failed jobs to a database table. "logFailedJobs" : false, // Datasource information for loggin failed jobs. "logFailedJobsProperties" : { "tableName" : "cbq_failed_jobs", "datasource" : "", // `datasource` can also be a struct. "queryOptions" : {} // The sibling `datasource` property overrides any defined datasource in `queryOptions`. }, // Flag to allow restricting Job interceptor execution using a `jobPattern` annotation. "registerJobInterceptorRestrictionAspect" : false }; ``` ### configPath The configPath is a dot-delimited path to your cbq Config Component. By convention, this should be placed in your application's `config/` folder alongside other config files like `ColdBox.cfc` and `WireBox.cfc`. ### registerWorkers This flag is responsible for spinning up Worker Pools when the application starts. If a particular instance of your application should **not** work jobs as well as dispatch them, then this setting should be set to `false`. To make this easy, you can set the `CBQ_REGISTER_WORKERS` environment variable and it will be picked up. (If you override this setting in your own `moduleSettings` it will still take precedence over the environment variable.) ### scaleInterval {% hint style="danger" %} This feature is not implemented yet. {% endhint %} This is the interval in seconds that the scale job is ran in the background. The scale job enables you to scale Worker Pools up or down based on any other factors in your application. **Setting this value to `0` disables the job entirely.** ### defaultWorkerBackoff This setting provides a default backoff value for a Worker Pool. This can still be overridden on a configured Worker Pool in your cbq config file or on an individual Job or `dispatch` call. ### defaultWorkerTimeout This setting provides a default timeout value for a Worker Pool. This can still be overridden on a configured Worker Pool in your cbq config file or on an individual Job or `dispatch` call. ### defaultWorkerMaxAttempts This setting provides a default max attempts value for a Worker Pool. This can still be overridden on a configured Worker Pool in your cbq config file or on an individual Job or `dispatch` call. ### batchRepositoryProperties A struct of configuration properties for a Batch Repository. This is only needed if you dispatch any batches. #### tableName The name of the batch table. The default is `cbq_batches`. #### datasource The datasource to use to interact with the batch table. If no datasource is provided, it uses the default application datasource. A `struct` can also be provided if your CFML engine supports it. #### queryOptions A struct of query options to pass to the `queryExecute` call when interacting with the batch table. If a `datasource` is defined above, it will override any `datasource` key inside the `queryOptions`. ### logFailedJobs This flag will send failed jobs to a configured database table when true. In some systems, this is called a Dead Letter Queue or DLQ. ### logFailedJobsProperties #### tableName The name of the failed jobs table. The default is `cbq_failed_jobs`. #### datasource The datasource to use to interact with the failed jobs table. If no datasource is provided, it uses the default application datasource. A `struct` can also be provided if your CFML engine supports it. #### queryOptions A struct of query options to pass to the `queryExecute` call when interacting with the failed jobs table. If a `datasource` is defined above, it will override any `datasource` key inside the `queryOptions`. ### registerJobInterceptorRestrictionAspect Flag to allow [restricting Job interceptor execution](/4.0.0/interceptors#jobpattern-annotation) using a `jobPattern` annotation. # Config File The cbq config file is where you define Queue Connections as well as Worker Pools. These Connections and Worker Pools can also be added, removed, or modified based on the current environment. When you want to change where your queued Jobs are sent or how they are worked, this is the file you will modify. ## Definitions ### Queue Connections Also referred to as "Connection." A Queue Connection defines where Jobs are serialized and the default settings that apply. ## configure The `configure` method is where the production Queue Connection and Worker Pools definitions are constructed. ### newConnection This command creates a new `QueueConnectionDefinition` builder component. It requires a unique name that will define this Connection and that will be used when defining Worker Pools.
ArgumentsTypeRequiredDefaultDescription
namestringtrueThe unique name for the new Queue Connection.
```cfscript component { function configure() { newConnection( "default" ); } } ``` A `QueueConnectionDefinition` has various methods to configure the Queue Connection. {% content-ref url="/pages/pghfGWdoAmKqeFiwqKtz" %} [Queue Connection](/4.0.0/configuration/config-file/queue-connection) {% endcontent-ref %} ### newWorkerPool This command creates a new `WorkerPoolDefinition` builder component. It requires a unique `name` that will define this Worker Pool and a `connectionName` that points to an already created Connection.
ArgumentsTypeRequiredDefaultDescription
namestringtrueThe unique name for the new Worker Pool.
connectionNamestringfalseA reference to an existing Connection. If not passed in here, the forConnection method must be called to define the Connection.
quantitynumericfalse1The number of workers to spin up for this Worker Pool.
queuesarrayfalse[ * ]An array of queues that this Worker Pool will work. A queue of * refers to all queues. Queues will be worked in the order provided.
forcebooleanfalsefalseIf false, an exception will be thrown if the Worker Pool name has already been registered. If true, the new definition will override the existing definition.
component {

    function configure() {
        newConnection( "default" );
        
        newWorkerPool( "default" ).forConnection( "default" );
    }

}
A `WorkerPoolDefinition` has various methods to configure the Worker Pool. {% content-ref url="/pages/FRcimvpoy6mO5qGgTvgb" %} [Worker Pool](/4.0.0/configuration/config-file/worker-pool) {% endcontent-ref %} ### reset Removes all Connections and Worker Pools ```cfscript getInstance( "Config@cbq" ).reset(); ``` ### Environment Overrides Just as in your `config/ColdBox.cfc` file or in `ModuleConfig.cfc` files, you can add, remove, or modify Queue Connection or Worker Pool definitions per environment. You do this by defining a method on your Config component matching the environment name you want to override. ```cfscript component { function configure() { newConnection( "default" ) .provider( "DBProvider@cbq" ); newWorkerPool( "default" ) .forConnection( "default" ) .quantity( 3 ) .timeout( 15 ); } /** * This method will be called after `configure` * and only if the current environment is `development`. */ function development() { withConnection( "default" ) .provider( "SyncProvider@cbq" ); withWorkerPool( "default" ) .quantity( 1 ) .timeout( 60 ); } } ``` ### withConnection Retrieves an already defined Queue Connection Definition for overriding. All the same methods are available. {% content-ref url="/pages/pghfGWdoAmKqeFiwqKtz" %} [Queue Connection](/4.0.0/configuration/config-file/queue-connection) {% endcontent-ref %} #### delete {% hint style="danger" %} This method is not yet implemented. {% endhint %} ### withWorkerPool Retrieves an already defined Worker Pool Definition for overriding. All the same methods are available. {% content-ref url="/pages/FRcimvpoy6mO5qGgTvgb" %} [Worker Pool](/4.0.0/configuration/config-file/worker-pool) {% endcontent-ref %} #### delete {% hint style="danger" %} This method is not yet implemented. {% endhint %} # Queue Connection A Queue Connection is defined inside your cbq config file using a `QueueConnectionDefinition` builder component. You can create one of these builder components using the [`newConnection`](/4.0.0/configuration/config-file#newconnection) method. ### QueueConnectionDefinition Methods #### provider Sets the provider for the Queue Connection. This can be any valid WireBox mapping and should implement the `IQueueProvider` interface. (There is no need to use the `implements` keyword.)
ArgumentsTypeRequiredDefaultDescription
providerstringtrueA valid WireBox mapping to the desired Provider.
#### setProvider Alias for [`provider`](#provider). #### setProperties Accepts a struct of properties to configure the Queue Connection. The available properties are usually defined by the Queue Provider being used. #### setDefaultQueue Sets the default queue to use for jobs dispatched on this Queue Connection. #### markAsDefault Marks this Queue Connection as the default Queue Connection for jobs dispatched without specifying a Queue Connection. #### setMakeDefault Alias for [`markAsDefault`](#markasdefault). # Worker Pool A Queue Connection is defined inside your cbq config file using a `QueueConnectionDefinition` builder component. You can create one of these builder components using the [`newConnection`](/4.0.0/configuration/config-file#newconnection) method. ### WorkerPoolDefinition Methods #### setName Sets the name for the Worker Pool. Usually not called directly as the `newWorkerPool` method requires a `name`. ```cfscript newWorkerPool( "default" ) .setName( "not-default" ); ``` #### forConnection Sets the name of the associated Connection for the Worker Pool. This must reference an already registered Connection. ```cfscript newConnection( "db" ) .provider( "DBProvider@cbq" ); newWorkerPool( "db-worker" ) .forConnection( "db" ); ``` #### setConnectionName Alias for [forConnection](#forconnection). #### quantity Sets the quantity of workers for this Worker Pool. ```cfscript newWorkerPool( "default" ) .forConnection( "db" ) .quantity( 3 ); ``` #### setQuantity Alias for [quantity](#quantity). #### onQueue The name of a queue to work. The default queue is named `default`. ```cfscript newWorkerPool( "premium-only" ) .forConnection( "db" ) .onQueue( "premium" ); newWorkerPool( "priority" ) .forConnection( "db" ) .onQueue( "priority" ) .quantity( 4 ); newWorkerPool( "default" ) .forConnection( "db" ); // uses `default` queue ``` Some Providers allow for working a priority of queues, such as the `DBProvider`. In these cases, you can pass an array of queues, in priority order, using the asterisk (`*`) as a wildcard character. ```cfscript newWorkerPool( "premium-only" ) .forConnection( "db" ) .onQueue( "premium" ); newWorkerPool( "default" ) .forConnection( "db" ) .onQueue( [ "priority", "*" ] ); ``` {% hint style="danger" %} **Throws:** `cbq.WorkerPool.MultipleQueuesNotSupported` {% endhint %} #### setQueue Alias for [onQueue](#onqueues). #### backoff Sets the backoff time amount, in seconds. ```cfscript newWorkerPool( "default" ) .forConnection( "db" ) .backoff( 30 ); ``` #### setBackoff Alias for [backoff](#backoff). #### timeout Sets the timeout time amount, in seconds. ```cfscript newWorkerPool( "default" ) .forConnection( "db" ) .timeout( 60 ); ``` #### setTimeout Alias for [timeout](#timeout). #### maxAttempts Sets the max number of attempts. ```cfscript newWorkerPool( "default" ) .forConnection( "db" ) .maxAttempts( 5 ); ``` #### setMaxAttempts Alias for [maxAttempts](#maxattempts). # Providers A Queue Provider provides a way to serialize jobs to a Queue Connection and to work jobs off a Queue Connection. The three built-in cbq providers are: * [SyncProvider@cbq](/4.0.0/configuration/providers/syncprovider) * [ColdBoxAsyncProvider@cbq](/4.0.0/configuration/providers/coldboxasyncprovider) * [DBProvider@cbq](/4.0.0/configuration/providers/dbprovider) ### Writing your own Custom Provider All Queue Providers extend the `AbstractQueueProvider` base component. They must implement three abstract methods: ```cfscript /** * Persists a serialized job to the Queue Connection * * @queueName The queue name for the job. * @payload The serialized job string. * @delay The delay (in seconds) before working the job. * @attempts The current attempt number. * * @return AbstractQueueProvider */ public any function push( required string queueName, required string payload, numeric delay = 0, numeric attempts = 0 ); /** * Starts a worker for a Worker Pool on this Queue Connection. * * @pool The Worker Pool that is working this Queue Connection. * * @return A function that when called will stop this worker. */ public function function startWorker( required WorkerPool pool ); /** * Starts any background processes needed for the Worker Pool. * * @pool The Worker Pool that is working this Queue Connection. * * @return AbstractQueueProvider */ public any function listen( required WorkerPool pool ); ``` # SyncProvider {% hint style="danger" %} This provider is not **durable**. (Pending jobs are not saved and may be lost if there are server issues between dispatching the job and working the job.) {% endhint %} {% hint style="danger" %} This provider **does not support multiple queues**. (Only a single queue can be provided to a Worker Pool instance. Additional queues to be worked should be registered as separate Worker Pool instances.) {% endhint %} {% hint style="danger" %} Jobs dispatched from inside a Sync job are executed immediately and recursively. This means a child job failing will cause any parent jobs to also fail. {% endhint %} The SyncProvider runs any jobs dispatched during a request in the same request synchronously. This can be especially useful in development when debugging jobs. If you job would throw an exception, you will see it with any chosen error handler and tools you use to debug local exceptions. # ColdBoxAsyncProvider {% hint style="danger" %} This provider is not **durable**. (Pending jobs are not persisted and may be lost if there are server issues between dispatching the job and working the job.) {% endhint %} {% hint style="danger" %} This provider **does not support multiple queues**. (Only a single queue can be provided to a Worker Pool instance. Additional queues to be worked should be registered as separate Worker Pool instances.) {% endhint %} The ColdBoxAsyncProvider runs any jobs dispatched on a background thread using ColdBox's AsyncManager. Jobs will be worked by the same server that dispatched them. The number of workers specified translates to the number of threads dedicated to working the jobs. # DBProvider {% hint style="success" %} This provider is **durable**. (Pending jobs are persisted. If there are server issues, the jobs will be worked when the issues are resolved. Multiple different servers can dispatch to this Queue Connection and multiple different Worker Pools can work these jobs.) {% endhint %} {% hint style="success" %} This provider **supports multiple queues**. (An array of prioritized queues can be assigned to Worker Pools worked by this provider.) {% endhint %} You may use a database as the backing engine for your Queue Connection using the DBProvider. All database grammars that are [supported by qb](https://qb.ortusbooks.com/v/9.0.0/installation-and-usage) are supported. ### Configuration The DBProvider has three optional arguments. The are presented below with their default values. ```cfscript { "tableName": "cbq_jobs", "datasource": null, "queryOptions": {} } ``` #### tableName The name of the table to use when managing jobs. This table should have the structure provided by the provided database migration file. #### datasource The name of the datasource to use when managing jobs. This overrides any datasource provided in the `queryOptions`. #### queryOptions A struct of options that will be passed to [`queryExecute`](https://cfdocs.org/queryexecute) when managing jobs. ### Jobs Table Structure The `cbq_jobs` table must have a specific structure. It is provided in the form of a migration file called `2000_01_01_000000_create_cbq_jobs_table.cfc`. This migration can be ran via [CommandBox Migrations](https://forgebox.io/view/commandbox-migrations) or [CFMigrations](https://forgebox.io/view/cfmigrations). If you wish, you can generate the table in other ways, so long as it matches the structure provided in the migration file. If you choose to use the migration file in your application, copy the file out to your own migrations folder first. ```cfscript schema.create( "cbq_jobs", function ( t ) { t.bigIncrements( "id" ); t.string( "queue" ); t.longText( "payload" ); t.unsignedTinyInteger( "attempts" ); t.unsignedInteger( "reservedDate" ).nullable(); t.unsignedInteger( "availableDate" ); t.unsignedInteger( "createdDate" ); t.index( "queue" ); } ); ``` # Defining a Job A Job represents both the object to be serialized to a [Connection](/4.0.0/configuration/config-file#connection) to eventually work as well as the code to execute when working a Job. ## Extending from AbstractJob The first step to define a Job is to extend from `AbstractJob`. This provides the necessary helper methods to run a Job through the cbq pipeline. ```cfscript component name="SendWelcomeEmailJob" extends="cbq.models.Jobs.AbstractJob" { } ``` ## Properties Defining properties on your job allow you to see at a glance what data your Job expects when constructing it. These properties and their values will be serialized to a [Connection](/4.0.0/configuration/config-file#connection) when dispatching a Job. {% hint style="info" %} Defining the properties is not strictly necessary, but your future self will thank you when you try to remember what properties your job is using. {% endhint %} ```cfscript component name="SendWelcomeEmailJob" extends="cbq.models.Jobs.AbstractJob" { property name="email"; property name="greeting"; } ``` ## The handle method The `handle` method is called when a Job is worked. Before being called, a Job will be reconstructed with the serialized data from the Connection. This method is the only required method when defining a Job. ```cfscript component name="SendWelcomeEmailJob" extends="cbq.models.Jobs.AbstractJob" { property name="mailService" inject="provider:MailService@cbmailservices"; property name="email"; property name="greeting"; function handle() { variables.mailService.newMail( to = variables.email, from = "noreply@example.com", subject = "Welcome!", type = "html", bodyTokens = { "greeting": variables.greeting "link": getInstance( "coldbox:requestContext" ) .buildLink( "home" ) } ) .setView( "_emails/welcome" ) .send(); } } ``` {% hint style="info" %} Prefer using `provider:` injections or inline `getInstance` calls for logic in your `handle` method. The `Job` component is created both when dispatching and when working your job, so utilizing these tools will reduce unnecessary processing time when dispatching your job. {% endhint %} ## Job Execution Properties A job can define several execution properties on the job itself. If defined, these values override the module, connection, or worker defaults. It can still be overridden using the job methods when creating a job. 
component extends="cbq.models.Jobs.AbstractJob" {

    // Name of the connection to dispatch this job on.
    variables.connection = "db";
    
    // Name of the queue to use for this job.
    variables.queue = "priority";
    
    // Time, in seconds, between job attempts, including the initial attempt.
    variables.backoff = 15;
    
    // Time, in seconds, to let a job run before failing it.
    variables.timeout = 30;
    
    // Max number of attempts before marking a job as failed.
    // Use 0 for unlimited retries.
    variables.maxAttempts = 9;

}
## Lifecycle Methods A Job can define a `before` or `after` method that will be called as part of the Job lifecycle. 
component
    name="SendWelcomeEmailJob"
    extends="cbq.models.Jobs.AbstractJob"
{

    property name="mailService" inject="provider:MailService@cbmailservices";
    
    property name="email";
    property name="greeting";
    
    function handle() {
        variables.mailService.newMail( 
            to = variables.email,
	    from = "noreply@example.com",
	    subject = "Welcome!",
            type = "html",
	    bodyTokens = { 
		"greeting": variables.greeting
		"link": getInstance( "coldbox:requestContext" )
		    .buildLink( "home" )
	    }
        )
        .setView( "_emails/welcome" )
        .send();
    }
    
    function before() {
        log.debug( "About to execute SendWelcomeEmailJob" );
    }
    
    function after() {
        log.debug( "Finished executing SendWelcomeEmailJob" );
    }

}
# Creating a Job After you define your job, you need to create an instance of your job and set the properties before you dispatch it onto your [Queue Connection](/4.0.0/configuration/config-file#connection). You can do this is a few different ways. ## Creating a Job Instance You can create a Job instance using WireBox anywhere in your ColdBox application. ```cfscript component { function index( event, rc, prc ) { var job = getInstance( "SendWelcomeEmailJob" ); } } ``` {% hint style="warning" %} Keep in mind that Job instances are transient. Do not inject a Job instance into a Component or Scope that is not transient. For instance, do not inject a Job instance as a property in a Handler. {% endhint %} ### Setting Job Properties Once you have a Job instance, you can set the data for the particular Job by calling the setter methods. ```cfscript component { function index( event, rc, prc ) { var job = getInstance( "SendWelcomeEmailJob" ); job.setEmail( "john@example.com" ); job.setGreeting( "Welcome!" ); } } ``` ### setProperties You can also set all the properties at once using the `setProperties` method. {% hint style="warning" %} Using this method will overwrite any previously set properties. {% endhint %} ```cfscript component { function index( event, rc, prc ) { var job = getInstance( "SendWelcomeEmailJob" ); job.setProperties( { "email": "john@example.com", "greeting": "Welcome!" } ); } } ``` ### onConnection You can override the connection for a job by calling the `onConnection` method with the desired connection name. ### setConnection See [onConnection](#onconnection). ### onQueue The queue name to use for the job. Queue names can be any string you choose, but to be worked a `WorkerPool` must be defined working the same connection and queue name. ### setQueue See [onQueue](#onqueue). ### setBackoff Sets the amount of time, in seconds, to wait in-between Job attempts — including the initial attempt. ### setDelay An alias for [setBackoff](#setbackoff). ### setTimeout Sets the amount of time, in seconds, to let a Job run on a worker before marking it as a failed attempt. ### setMaxAttempts Sets the maximum number of attempts before a Job is marked as failed. ### chain Sets an array of Jobs to be executed, in order, after this Job successfully executes. ### getMemento getMemento ## [cbq.job()](/4.0.0/cbq-model#job) You can also create a Job instance using the `cbq` model. (This can be injected into your components using the `cbq@cbq` DSL.) {% hint style="success" %} The cbq model is a singleton, so feel free to inject it into any component. {% endhint %} ```cfscript component { property name="cbq" inject="cbq@cbq"; function index( event, rc, prc ) { var job = cbq.job( job = "SendWelcomeEmailJob", properties = { "email": "john@example.com", "greeting": "Welcome!" }, chain = [], queue = "priority", connection = "db", backoff = 10, timeout = 30, maxAttempts = 3 ); } } ``` # Dispatching a Job ## Dispatching Creating a Job is all well and good, but the Job will not be executed until it is dispatched on to a [Queue Connection](/4.0.0/configuration/config-file#connection). If you have a Job instance, you can dispatch it by calling the `dispatch` method. ```cfscript component { function index( event, rc, prc ) { getInstance( "SendWelcomeEmailJob" ); .setEmail( "john@example.com" ); .setGreeting( "Welcome!" ) .dispatch(); } } ``` You can also create and dispatch a job in the same call using the `cbq.dispatch()` method. ```cfscript component { property name="cbq" inject="cbq@cbq"; function index( event, rc, prc ) { cbq.dispatch( job = "SendWelcomeEmailJob", properties = { "email": "john@example.com", "greeting": "Welcome!" }, chain = [], queue = "priority", connection = "db", backoff = 10, timeout = 30, maxAttempts = 3 ); } } ``` ## Interception Points ### onCBQJobAdded This is fired before serializing a Job and sending it to a connection. The data includes the `job` to be serialized and dispatched and the connection the Job is being dispatched on. ```cfscript variables.interceptorService.announce( "onCBQJobAdded", { "job" : job, "connection" : connection } ); ``` # Working a Job Dispatched jobs will get picked up by Worker Pools set up for the same connection and queue. They are picked up in order by created date (FIFO). If you have a valid Worker Pool, then you don't need to do anything else to have your job picked up. If you want to see all the details of cbq dispatching and marshalling jobs, enable `debug` logging in LogBox for `cbq`. When a Job is worked, the Job instance is created via WireBox and the memento hydrated into the Job component. Then the `handle` method is called. When a Job is completed, the Job will be removed from the persistent storage of the Queue Provider. If the Job attempt fails, it will be sent back to the Queue Provider to be retried, up to the `maxAttempts` amount. Once a Job has failed the `maxAttempts` amount it will be removed and marked as failed. If you have `logFailedJobs` enabled, it will be sent to the [failed jobs table](/4.0.0/jobs/failed-jobs#logging-failed-jobs). ## Inside \`handle\` ### release You can choose to manually release a Job back to a queue with an optional delay (in seconds) using the `release` function. This is useful when you need to delay processing for a Job, perhaps due to rate limiting or licensing constraints. {% hint style="info" %} Calling `release` does not stop processing the `handle` method, so make sure you `return` if you don't want to keep executing your `handle` method. {% endhint %} ```cfscript component { function handle() { this.release( 60 ); // in seconds return; } } ``` ## Interception Points cbq fires a number of interception points during a Job's lifecycle. ### onCBQJobMarshalled This is fired before the `handle` method is called on a Job. The data includes the `job` to be executed. ```cfscript variables.interceptorService.announce( "onCBQJobMarshalled", { "job" : job } ); ``` ### onCBQJobComplete This is fired after a Job completes successfully. The data includes the `job` to be executed as potentially a `result` returned from the `Job`'s handle method. ```cfscript variables.interceptorService.announce( "onCBQJobComplete", { "job" : job, "result" : isNull( result ) ? javacast( "null", "" ) : result } ); ``` ## Tips ### Need up to date data? Could your data change between when you dispatched a job and when you work the job? If so, consider using the key as a property and fetching the data from inside the job. ### Need to use historical data? In some cases, you want to work with the data that existed at the time the Job was dispatched. In these cases, make sure to include all the needed data in the Job instance. ### Check if you need to work the Job still Things might have changed since the Job was dispatched. Most Queue Providers do not support querying the current Jobs queue or interacting with it in advance, so check in your `dispatch` method if the Job still needs to be worked. ### Need to try the Job later? In some cases, you need to retry your Job later. In these cases, use the `release` method to send the Job back to the queue. It also takes an optional `delay` which sets the `backoff` time for the next Job attempt. You can combine this with setting the `maxAttempts` of a Job to `0` to retry it indefinitely. Remember that a Job is only reattempted if it fails. If you decide that a Job is finished, just `return`. ### Jobs can dispatch other Jobs You can dispatch other Jobs from your Job. This can be different Jobs or even a similar instance of the same Job. ## Why isn't my Job getting picked up? To work a cbq job, you need a [defined Worker Pool in your cbq Config file.](/4.0.0/configuration/config-file#newworkerpool) In order to work a Job, the Worker Pool must: 1. Be working the same [connection](/4.0.0/configuration/config-file/worker-pool#forconnection) as the Job. 2. Be working the same [queue](/4.0.0/configuration/config-file/worker-pool#onqueues) as the Job. 3. Have at least one active worker (defined with [`setQuantity`](/4.0.0/configuration/config-file/worker-pool#setquantity)). If these requirements are not met, the job will stay in the queue storage, unable to get picked up. # Failed Jobs ## Failed Attempt vs Failed Job An important distinction to make is between failed Job attempts and failed Jobs. A Job is attempted up to the defined `maxAttempts`. If a Job fails all of its `maxAttempts` then it is marked as a failed Job and removed from the queue. Otherwise, the Job is dispatched back to the queue with the current execution count increased. ## LogBox Failed Job attempts are logged to LogBox. You will see the exception as well as the serialized Job memento in the `extraInfo`. ## Interceptors ### onCBQJobException This interception point is fired for every **Job Attempt** failure. The data includes the `job` component and the `exception`. ```cfscript variables.interceptorService.announce( "onCBQJobException", { "job" : job, "exception" : e } ); ``` ### onCBQJobFailed This interception point is fired when the Job is marked as failed — when the Job has failed all attempts up to the configured `maxAttempts`. The data includes the `job` component and the `exception`. ```cfscript variables.interceptorService.announce( "onCBQJobFailed", { "job" : job, "exception" : e } ); ``` ## onFailure Job Method You can define an `onFailure` method on your Job component. It will be called with the `exception`. ```cfscript component extends="cbq.models.Jobs.AbstractJob" { function handle() { // ... } function onFailure( e ) { log.error( "Bad things happened: #e.message#" ); } } ``` ## Logging Failed Jobs ## Failed Jobs Table cbq includes an interceptor to log failed jobs to a database. You can enable this in your module settings: ```cfscript moduleSettings = { "cbq": { // Flag to turn on logging failed jobs to a database table. "logFailedJobs" : false, // Datasource information for loggin failed jobs. "logFailedJobsProperties" : { "tableName" : "cbq_failed_jobs", "datasource" : "", // `datasource` can also be a struct. "queryOptions" : {} // The sibling `datasource` property overrides any defined datasource in `queryOptions`. } } }; ``` Also included is a migration file to add the needed table to your database. You can find it in `resources/database/migrations/2000_01_01_000002_create_cbq_failed_jobs_table.cfc`. It is also included below. ```cfscript component { function up( schema ) { schema.create( "cbq_failed_jobs", function ( t ) { t.bigIncrements( "id" ); t.string( "connection" ); t.string( "queue" ); t.string( "mapping" ); t.longText( "memento" ); t.longText( "properties" ); t.string( "exceptionType" ).nullable(); t.string( "exceptionMessage" ); t.string( "exceptionDetail" ).nullable(); t.longText( "exceptionExtendedInfo" ).nullable(); t.longText( "exceptionStackTrace" ); t.longText( "exception" ); t.unsignedInteger( "failedDate" ); } ); } function down( schema ) { schema.dropIfExists( "cbq_failed_jobs" ); } } ``` {% code title="MySQL" %} ```sql CREATE TABLE `cbq_failed_jobs` ( `id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT, `connection` NVARCHAR(255) NOT NULL, `queue` NVARCHAR(255) NOT NULL, `mapping` NVARCHAR(255) NOT NULL, `memento` LONGTEXT NOT NULL, `properties` LONGTEXT NOT NULL, `exceptionType` NVARCHAR(255), `exceptionMessage` NVARCHAR(255) NOT NULL, `exceptionDetail` NVARCHAR(255), `exceptionExtendedInfo` LONGTEXT, `exceptionStackTrace` LONGTEXT NOT NULL, `exception` LONGTEXT NOT NULL, `failedDate` INT UNSIGNED NOT NULL ) ``` {% endcode %} {% code title="SQL Server" %} ```sql CREATE TABLE [dbo].[cbq_failed_jobs]( [id] BIGINT NOT NULL IDENTITY, [connection] NVARCHAR(255) NOT NULL, [queue] NVARCHAR(255) NOT NULL, [mapping] NVARCHAR(255) NOT NULL, [memento] NVARCHAR(MAX) NOT NULL, [properties] NVARCHAR(MAX) NOT NULL, [exceptionType] NVARCHAR(255), [exceptionMessage] NVARCHAR(255) NOT NULL, [exceptionDetail] NVARCHAR(255), [exceptionExtendedInfo] NVARCHAR(MAX), [exceptionStackTrace] NVARCHAR(MAX) NOT NULL, [exception] NVARCHAR(MAX) NOT NULL, [failedDate] INT NOT NULL ) ``` {% endcode %} ## Retrying a Failed Job ## Configuring Job Backoff # Chained Jobs Job chains mean that after a Job completes successfully it dispatches the next Job in the chain. If a Job fails, no more Jobs in the chain are dispatched. ## Dispatching Jobs from another Job One way to make a Job chain is to dispatch a Job from inside another Job. This gives you consistency in your Job chain, and a logical code path to follow. ```cfscript // FulfillOrderJob.cfc component extends="cbq.models.Jobs.AbstractJob" { function handle() { var productId = getProperties().productId; processPayment( productId ); getInstance( "SendProductLinkEmail" ) .onConnection( "fulfillment" ) .setProperties( { "productId" : productId, "userId" : getProperties().userId } ) .dispatch(); } } ``` ## Creating a Chain Another way to make a Job chain is to create is when dispatching the Job. This gives you flexibility to compose Job chains at runtime. ```cfscript // handlers/Main.cfc component { property name="cbq" inject="cbq@cbq"; function create() { cbq.chain( [ cbq.job( "FulfillOrderJob", { "productId": rc.productId } ), cbq.job( job = "SendProductLinkEmail", properties = { "productId": rc.productId, "userId": auth().getUserId() }, connection = "fulfillment" ), ] ).dispatch(); } } ``` ## Dispatching Jobs on Failure Utilizing the `onFailure` method of a Job, we can dispatch another Job if our Job fails. ```cfscript // FulfillOrderJob.cfc component extends="cbq.models.Jobs.AbstractJob" { function handle() { // ... } function onFailure( e ) { getInstance( "SendOrderProblemEmail" ) .onConnection( "fulfillment" ) .setProperties( { "productId" : productId, "userId" : getProperties().userId, "message" : e.message } ) .dispatch(); } } ``` # Batched Jobs cbq allows for tracking jobs as part of a batch. The main use case for batched jobs is to dispatch additional jobs when a batch has completed successfully, completed with failures, or completed in any fashion. (Think `try`, `catch`, `finally`.) ## Setup To utilize batches, you first need to set up a Batch repository. Batches are tracked separate from jobs. cbq utilizes a database repository to track batches regardless of what provider your Job's connection uses. A database migration is provided in `resources/database/migrations/2000_01_01_000001_create_cbq_batches_table.cfc`. You can copy this to your own project to run via [cfmigrations](https://forgebox.io/view/cfmigrations) or you can reference the migration or SQL scripts [below](#migrations-and-sql-scripts). You can customize the batch repository using the `batchRepositoryProperties` of cbq's `moduleSettings`. This is a struct with two properties you can set: #### tableName This defaults to `cbq_batches`. You can set this to any unique table name in your datasource. #### queryOptions This is a struct of options that will be passed to `queryExecute`. The most common use case for this property is to specify a specific datasource to use for the batches table. This defaults to an empty struct (`{}`). ## Defining Batches To create a batch, you can get an instance of `PendingBatch@cbq` from WireBox or use the [`cbq.batch()`](/4.0.0/cbq-model#batch) helper method. Once you have a `PendingBatch` instance, you can add jobs to the batch using the `add` method. ### add Adds a single Job or an array of Jobs to a PendingBatch.
ArgumentsTypeRequiredDefaultDescription
jobstring | Job | array<Job>trueThe Job WireBox id, Job instance, or array of Job instances to add to the PendingBatch.
propertiesStructfalse{}A struct of properties for the Job instance. (Only used when providing a Job WireBox id.)
chainarray<Job>false[]An array of Jobs to chain after this Job. (Only used when providing a Job WireBox id.)
queuestringfalseThe queue the Job belongs to. (Only used when providing a Job WireBox id.)
connectionstringfalseThe Connection to dispatch the Job on. (Only used when providing a Job WireBox id.)
backoffnumericfalseThe amount of time, in seconds, to wait between attempting Jobs. (Only used when providing a Job WireBox id.)
timeoutnumericfalseThe amount of time, in seconds, to wait before treating a Job as erroring. (Only used when providing a Job WireBox id.)
maxAttemptsnumericfalseThe maximum amount of attempts of a Job before treating the Job as failed. (Only used when providing a Job WireBox id.)
**Return:** The PendingBatch to be dispatched. ```cfscript batch .add( cbq.job( "ImportCsvJob", { "start": 1, "end": 100 } ) ) .add( "ImportCsvJob", { "start": 101, "end": 200 } ) .add( [ cbq.job( "ImportCsvJob", { "start": 201, "end": 300 } ), cbq.job( "ImportCsvJob", { "start": 301, "end": 400 } ), cbq.job( "ImportCsvJob", { "start": 401, "end": 500 } ) ] ); ``` The primary use case of batches is to dispatch lifecycle jobs when the batch has completed and if the batch completed successfully or completed with failures. These jobs can be configured using the `then`, `catch`, and `finally` methods. ### then Defines a Job to be dispatched when all the jobs in the batch finishes successfully.
ArgumentsTypeRequiredDefaultDescription
jobstring | JobtrueThe Job WireBox id or Job instance to execute if all the jobs in the Batch complete successfully.
propertiesStructfalse{}A struct of properties for the Job instance. (Only used when providing a Job WireBox id.)
chainarray<Job>false[]An array of Jobs to chain after this Job. (Only used when providing a Job WireBox id.)
queuestringfalseThe queue the Job belongs to. (Only used when providing a Job WireBox id.)
connectionstringfalseThe Connection to dispatch the Job on. (Only used when providing a Job WireBox id.)
backoffnumericfalseThe amount of time, in seconds, to wait between attempting Jobs. (Only used when providing a Job WireBox id.)
timeoutnumericfalseThe amount of time, in seconds, to wait before treating a Job as erroring. (Only used when providing a Job WireBox id.)
maxAttemptsnumericfalseThe maximum amount of attempts of a Job before treating the Job as failed. (Only used when providing a Job WireBox id.)
**Return:** The `PendingBatch` instance. ```cfscript cbq.batch( [ cbq.job( "ImportCsvJob", { "start": 1, "end": 100 } ), cbq.job( "ImportCsvJob", { "start": 101, "end": 200 } ), cbq.job( "ImportCsvJob", { "start": 201, "end": 300 } ), cbq.job( "ImportCsvJob", { "start": 301, "end": 400 } ), cbq.job( "ImportCsvJob", { "start": 401, "end": 500 } ) ] ) .then( cbq.job( "ImportCsvSuccessfulJob" ) ); ``` ### onFailure Defines a Job to be dispatched the first time a job in the Batch fails.
ArgumentsTypeRequiredDefaultDescription
jobstring | JobtrueThe Job WireBox id or Job instance to execute the first time a job in the Batch fails.
propertiesStructfalse{}A struct of properties for the Job instance. (Only used when providing a Job WireBox id.)
chainarray<Job>false[]An array of Jobs to chain after this Job. (Only used when providing a Job WireBox id.)
queuestringfalseThe queue the Job belongs to. (Only used when providing a Job WireBox id.)
connectionstringfalseThe Connection to dispatch the Job on. (Only used when providing a Job WireBox id.)
backoffnumericfalseThe amount of time, in seconds, to wait between attempting Jobs. (Only used when providing a Job WireBox id.)
timeoutnumericfalseThe amount of time, in seconds, to wait before treating a Job as erroring. (Only used when providing a Job WireBox id.)
maxAttemptsnumericfalseThe maximum amount of attempts of a Job before treating the Job as failed. (Only used when providing a Job WireBox id.)
**Return:** The `PendingBatch` instance. ```cfscript cbq.batch( [ cbq.job( "ImportCsvJob", { "start": 1, "end": 100 } ), cbq.job( "ImportCsvJob", { "start": 101, "end": 200 } ), cbq.job( "ImportCsvJob", { "start": 201, "end": 300 } ), cbq.job( "ImportCsvJob", { "start": 301, "end": 400 } ), cbq.job( "ImportCsvJob", { "start": 401, "end": 500 } ) ] ) .onFailure( cbq.job( "ImportCsvFailedJob" ) ); ``` ### onComplete Defines a Job to be dispatched after all the jobs in the Batch have executed successfully or failed.
ArgumentsTypeRequiredDefaultDescription
jobstring | JobtrueThe Job WireBox id or Job instance to execute after all the jobs in the Batch have executed successfully or failed.
propertiesStructfalse{}A struct of properties for the Job instance. (Only used when providing a Job WireBox id.)
chainarray<Job>false[]An array of Jobs to chain after this Job. (Only used when providing a Job WireBox id.)
queuestringfalseThe queue the Job belongs to. (Only used when providing a Job WireBox id.)
connectionstringfalseThe Connection to dispatch the Job on. (Only used when providing a Job WireBox id.)
backoffnumericfalseThe amount of time, in seconds, to wait between attempting Jobs. (Only used when providing a Job WireBox id.)
timeoutnumericfalseThe amount of time, in seconds, to wait before treating a Job as erroring. (Only used when providing a Job WireBox id.)
maxAttemptsnumericfalseThe maximum amount of attempts of a Job before treating the Job as failed. (Only used when providing a Job WireBox id.)
**Return:** The `PendingBatch` instance. ```cfscript cbq.batch( [ cbq.job( "ImportCsvJob", { "start": 1, "end": 100 } ), cbq.job( "ImportCsvJob", { "start": 101, "end": 200 } ), cbq.job( "ImportCsvJob", { "start": 201, "end": 300 } ), cbq.job( "ImportCsvJob", { "start": 301, "end": 400 } ), cbq.job( "ImportCsvJob", { "start": 401, "end": 500 } ) ] ) .onComplete( cbq.job( "ImportCsvCompletedJob" ) ); ``` ## Dispatching Batches A `PendingBatch` must be dispatched before any of the jobs contained in it are dispatched. This is done using the `dispatch` method on the `PendingBatch`. ### dispatch Dispatches a PendingBatch and all the Jobs it contains.
ArgumentsTypeRequiredDefaultDescription
No arguments
**Return:** A `Batch` instance created from this `PendingBatch`. ## Interacting with Batches ## Migrations and SQL Scripts ### cfmigrations ```cfscript schema.create( "cbq_batches", function ( t ) { t.string( "id" ).primaryKey(); t.string( "name" ); t.unsignedInteger( "totalJobs" ); t.unsignedInteger( "pendingJobs" ); t.unsignedInteger( "failedJobs" ); t.text( "failedJobIds" ); // JSON column t.text( "options" ).nullable(); // JSON column t.datetime( "createdDate" ); t.datetime( "cancelledDate" ).nullable(); t.datetime( "completedDate" ).nullable(); } ); ``` ### MySQL ```sql CREATE TABLE `cbq_batches` ( `id` VARCHAR(255) NOT NULL, `name` VARCHAR(255) NOT NULL, `totalJobs` INTEGER UNSIGNED NOT NULL, `pendingJobs` INTEGER UNSIGNED NOT NULL, `failedJobs` INTEGER UNSIGNED NOT NULL, `failedJobIds` TEXT NOT NULL, `options` TEXT, `createdDate` DATETIME NOT NULL, `cancelledDate` DATETIME, `completedDate` DATETIME, CONSTRAINT `pk_cbq_batches_id` PRIMARY KEY (`id`) ) ``` ### SQL Server ```sql CREATE TABLE [cbq_batches] ( [id] VARCHAR(255) NOT NULL, [name] VARCHAR(255) NOT NULL, [totalJobs] INTEGER NOT NULL, [pendingJobs] INTEGER NOT NULL, [failedJobs] INTEGER NOT NULL, [failedJobIds] VARCHAR(MAX) NOT NULL, [options] VARCHAR(MAX), [createdDate] DATETIME2 NOT NULL, [cancelledDate] DATETIME2, [completedDate] DATETIME2, CONSTRAINT [pk_cbq_batches_id] PRIMARY KEY ([id]) ) ``` ### Postgres ```sql CREATE TABLE "cbq_batches" ( "id" VARCHAR(255) NOT NULL, "name" VARCHAR(255) NOT NULL, "totalJobs" INTEGER NOT NULL, "pendingJobs" INTEGER NOT NULL, "failedJobs" INTEGER NOT NULL, "failedJobIds" TEXT NOT NULL, "options" TEXT, "createdDate" TIMESTAMP NOT NULL, "cancelledDate" TIMESTAMP, "completedDate" TIMESTAMP, CONSTRAINT "pk_cbq_batches_id" PRIMARY KEY ("id") ) ``` ### Oracle ```sql CREATE TABLE "CBQ_BATCHES" ( "ID" VARCHAR2(255) NOT NULL, "NAME" VARCHAR2(255) NOT NULL, "TOTALJOBS" NUMBER(10, 0) NOT NULL, "PENDINGJOBS" NUMBER(10, 0) NOT NULL, "FAILEDJOBS" NUMBER(10, 0) NOT NULL, "FAILEDJOBIDS" CLOB NOT NULL, "OPTIONS" CLOB, "CREATEDDATE" DATE NOT NULL, "CANCELLEDDATE" DATE, "COMPLETEDDATE" DATE, CONSTRAINT "PK_CBQ_BATCHES_ID" PRIMARY KEY ("ID") ) ``` ### SQLite ```sql CREATE TABLE "cbq_batches" ( "id" TEXT NOT NULL, "name" TEXT NOT NULL, "totalJobs" INTEGER NOT NULL, "pendingJobs" INTEGER NOT NULL, "failedJobs" INTEGER NOT NULL, "failedJobIds" TEXT NOT NULL, "options" TEXT, "createdDate" DATETIME NOT NULL, "cancelledDate" DATETIME, "completedDate" DATETIME, PRIMARY KEY ("id") ) ``` # cbq Model This model is provided to make certain tasks easier when defining and dispatching jobs, chains, and batches. ### dispatch Dispatches a job or chain of jobs.
ArgumentsTypeRequiredDefaultDescription
jobstring | Job | array<Job>trueA Job instance or a WireBox ID of a Job instance. If the WireBox ID doesn't exist, a NonExecutableJob instance will be used instead. This allows you to dispatch a Job from a server where that Job component is not defined. If an array is passed, a Job Chain is created and dispatched.
propertiesstructfalse{}A struct of properties for the Job instance.
chainarray<Job>false[]An array of Jobs to chain after this Job.
queuestringfalseThe queue the Job belongs to.
connectionstringfalseThe Connection to dispatch the Job on.
backoffnumericfalseThe amount of time, in seconds, to wait between attempting Jobs.
timeoutnumericfalseThe amount of time, in seconds, to wait before treating a Job as erroring.
maxAttemptsnumericfalseThe maximum amount of attempts of a Job before treating the Job as failed.
**Return:** The dispatched Job instance. ```cfscript cbq.dispatch( job = "SendWelcomeEmailJob", properties = { "body": "first body" }, queue = "default" ); ``` ### job Creates a job or chain of jobs to be dispatched.
ArgumentsTypeRequiredDefaultDescription
jobstring | Job | array<Job>trueA Job instance or a WireBox ID of a Job instance. If the WireBox ID doesn't exist, a NonExecutableJob instance will be used instead. This allows you to dispatch a Job from a server where that Job component is not defined. If an array is passed, a Job Chain is created and dispatched.
propertiesstructfalse{}A struct of properties for the Job instance.
chainarray<Job>false[]An array of Jobs to chain after this Job.
queuestringfalseThe queue the Job belongs to.
connectionstringfalseThe Connection to dispatch the Job on.
backoffnumericfalseThe amount of time, in seconds, to wait between attempting Jobs.
timeoutnumericfalseThe amount of time, in seconds, to wait before treating a Job as erroring.
maxAttemptsnumericfalseThe maximum amount of attempts of a Job before treating the Job as failed.
**Return:** The new Job instance. ```cfscript cbq.job( "SendWelcomeEmailJob" ) .setProperties( { "body": "first body" } ) .onQueue( "default" ) .dispatch(); ``` ### chain Creates a chain of jobs to be ran. Alias for calling `firstJob.chain( otherJobs )`.
ArgumentsTypeRequiredDefaultDescription
jobsarray<Job>trueThe array of jobs to run in order in a chain.
**Return:** The first job of the chain with the chained jobs configured to be dispatched. ```cfscript cbq.chain( [ cbq.job( "SendWelcomeEmailJob", { "body": "One" }, [], "default" ), cbq.job( "SendWelcomeEmailJob", { "body": "Two" }, [], "default", "sync" ), cbq.job( "SendWelcomeEmailJob", { "body": "Three" }, [], "default" ) ] ) .dispatch() ``` ### batch Creates a PendingBatch from the Jobs provided. {% hint style="warning" %} To use batches, you must first configure a `BatchRepository`. Learn more in the [Batched Jobs documentation](/4.0.0/jobs/batched-jobs). {% endhint %}
ArgumentsTypeRequiredDefaultDescription
jobsarray<Job>trueAn array of jobs to batch together.
**Return:** The PendingBatch to be dispatched. ```cfscript var batch = cbq .batch( [ cbq.job( "ImportCsvJob", { "start": 1, "end": 100 } ), cbq.job( "ImportCsvJob", { "start": 101, "end": 200 } ), cbq.job( "ImportCsvJob", { "start": 201, "end": 300 } ), cbq.job( "ImportCsvJob", { "start": 301, "end": 400 } ), cbq.job( "ImportCsvJob", { "start": 401, "end": 500 } ) ] ) .then( cbq.job( "ImportCsvSuccessfulJob" ) ) .catch( cbq.job( "ImportCsvFailedJob" ) ) .finally( cbq.job( "ImportCsvCompletedJob" ) ) .dispatch(); ``` # Interceptors ## Interception Points cbq announces the following interception points: ### `onCBQJobAdded` This is called when a Job is dispatched to a Queue. ### `onCBQJobMarshalled` This is called when a Job is pulled off the Queue to work. ### `onCBQJobComplete` This is called when a Job successfully finishes executing. ### `onCBQJobException` This is called when encountering an exception when handling a Job. ### `onCBQJobFailed` This is called when a Job is considered failed, after exhausting its `maxAttempts`. ## JobPattern Annotation {% hint style="warning" %} To use the `jobPattern` annotation, you must have enabled the `registerJobInterceptorRestrictionAspect` setting in your [module settings](/4.0.0/configuration/module-settings). {% endhint %} Interceptors listening on the cbq interception points listed above can optionally restrict execution to certain Jobs using a `jobPattern` annotation (similar to the `eventPattern` annotation [in ColdBox](https://coldbox.ortusbooks.com/the-basics/interceptors/restricting-execution)). ```cfscript component { function onCBQJobMarshalled( event, data ) jobPattern="SendWelcomeEmailJob" { // check if we've hit the email send limits for the month } } ``` This annotation accepts a regex string to check against the Job full name: ```cfscript component { function onCBQJobMarshalled( event, data ) jobPattern="^.*Email.*$" { // check if we've hit the email send limits for the month } } ``` {% hint style="info" %} Jobs that do not match the interception point will send a notice to the `debug` log, if that is turned on for `JobInterceptorRestriction`. {% endhint %} # Contributing # Contributors # Prior Art Initial ideas behind cbq were derived from [Laravel](https://laravel.com). --- [Next Page](/llms-full.txt/1)