ITB 2023 cbq - Jobs And Tasks In the Background - Eric Peterson.pdf

Learn how to manage background tasks with cbq. Scale from simple tasks to database or message queue providers. Ideal for developers wanting to run background processes efficiently, like sending emails without blocking requests. Some familiarity with asynchronous code helpful but not required.

2. https://slides.com/elpete/itb2023-cbq

3. What this talk is
An overview of cbq
Examples of problems that queues and
jobs can solve
Overview of how to create new cbq
protocols

4. What this talk isn't
How to use the ColdBox Async
Manager
Overview of ColdBox Scheduled Tasks

6. What is cbq?

7. A protocol-based queueing
system for ColdBox

8. A protocol-based queueing
system for ColdBox
Can interact with different providers, like
the ColdBox Async Engine, a database,
Redis, or Rabbit MQ
A protocol-based queueing
system for ColdBox

9. A protocol-based queueing
system for ColdBox
Sends a message to be consumed later.
It can be consumed by the same
application or a completely different
application, language, or service.
A protocol-based queueing
system for ColdBox

10. Defines a Job as the unit of work on a queue
Push a Job onto a queue connection to be worked later
Define multiple queues or named piles of work
Register worker pools to work the jobs passed to queues
Ability to switch between Queue Providers easily
What does cbq give me?

11. Why not just use...?
cfthread, runAsync, AsyncManager
Redis, Rabbit MQ, etc.
Homegrown queue framework

12. Why would I use cbq?

13. Sending
Email
Preparing Large
Spreadsheets
Video
Processing
Background
Uploads
Sequenced
Jobs
Scheduled SMS
Messages
Email
Verification
Processing
Payments
Cancelling
Abandoned Orders
Send Monthly
Invoices

14. Other reasons
Easier testing making sure the right jobs were
queued
Sync in development, Rabbit in production
Adjust the worker pool scale dynamically
Retry, timeout, and backoff built in.

15. Definitions

16. Job

17. Does its work in the `handle` method
Serializes and deserializes itself to the queue protocol
Set instance data in the `properties`
Exist in the context of your application
Job

18. Queue

19. A named stack of jobs or messages to be delivered
A queue connection must have at least one queue which is
usually `default`
A queue connection can have as many queues as desired
Queue

20. Queue Provider

21. How a queue connection connects to a backend like Redis,
RabbitMQ, or a database
Can be used multiple times in a single application to define
multiple queue connections with different configuration
options
Queue Provider

22. Queue Connection

23. A named combination of Queue Provider and properties
Allows you to connect with multiple Database providers or
multiple Redis providers
Queue Connection

24. Worker Pool

25. A group of workers for a Queue Connection
Can optionally work a subset of queues
Can optionally work queues in a specific order
Can be scaled up or down as needed
Worker Pool

26. Installation

27. box install cbq

28. component {


this.javaSettings = {
loadPaths : [ expandPath( "./modules/cbq/lib" ) ],
loadColdFusionClassPath : true,
reloadOnChange : false
};

}1 2 3 4 5 6 7 8 9 10
Batch-specific
Setup

29. component {


this.javaSettings = {
loadPaths : [ expandPath( "./modules/cbq/lib" ) ],
loadColdFusionClassPath : true,
reloadOnChange : false
};

}1 2 3 4 5 6 7 8 9 10
this.javaSettings = {component {1 2 3 4 loadPaths : [ expandPath( "./modules/cbq/lib" ) ],5 loadColdFusionClassPath : true,6 reloadOnChange : false7 };8 9 }10
Batch-specific
Setup

30. component {


this.javaSettings = {
loadPaths : [ expandPath( "./modules/cbq/lib" ) ],
loadColdFusionClassPath : true,
reloadOnChange : false
};

}1 2 3 4 5 6 7 8 9 10
this.javaSettings = {component {1 2 3 4 loadPaths : [ expandPath( "./modules/cbq/lib" ) ],5 loadColdFusionClassPath : true,6 reloadOnChange : false7 };8 9 }10 loadColdFusionClassPath : true,
reloadOnChange : false
};component {1 2 3 this.javaSettings = {4 loadPaths : [ expandPath( "./modules/cbq/lib" ) ],5 6 7 8 9 }10
Batch-specific
Setup

31. component {


this.javaSettings = {
loadPaths : [ expandPath( "./modules/cbq/lib" ) ],
loadColdFusionClassPath : true,
reloadOnChange : false
};

}1 2 3 4 5 6 7 8 9 10
this.javaSettings = {component {1 2 3 4 loadPaths : [ expandPath( "./modules/cbq/lib" ) ],5 loadColdFusionClassPath : true,6 reloadOnChange : false7 };8 9 }10 loadColdFusionClassPath : true,
reloadOnChange : false
};component {1 2 3 this.javaSettings = {4 loadPaths : [ expandPath( "./modules/cbq/lib" ) ],5 6 7 8 9 }10 }component {1 2 3 this.javaSettings = {4 loadPaths : [ expandPath( "./modules/cbq/lib" ) ],5 loadColdFusionClassPath : true,6 reloadOnChange : false7 };8 9 10
Batch-specific
Setup

32. Configuration

33. Module Settings

34. moduleSettings = {
"cbq" : {
// 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,

// continued...
}
};1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18

35. moduleSettings = {
"cbq" : {
// 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,

// continued...
}
};1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 // The path the custom config file to
// register connections and worker pools
"configPath" : "config.cbq",moduleSettings = {1 "cbq" : {2 3 4 5 6 // Flag if workers should be registered.7 // If your application only pushes to the queues,8 // you can set this to false.9 "registerWorkers" : getSystemSetting( "CBQ_REGISTER_WORKERS" , true ),10 11 // The interval to poll for changes to the worker pool scaling.12 // Defaults to 0 which turns off the scheduled scaling feature.13 "scaleInterval" : 0,14 15 // continued...16 }17 };18

36. moduleSettings = {
"cbq" : {
// 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,

// continued...
}
};1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 // The path the custom config file to
// register connections and worker pools
"configPath" : "config.cbq",moduleSettings = {1 "cbq" : {2 3 4 5 6 // Flag if workers should be registered.7 // If your application only pushes to the queues,8 // you can set this to false.9 "registerWorkers" : getSystemSetting( "CBQ_REGISTER_WORKERS" , true ),10 11 // The interval to poll for changes to the worker pool scaling.12 // Defaults to 0 which turns off the scheduled scaling feature.13 "scaleInterval" : 0,14 15 // continued...16 }17 };18 // 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 ),moduleSettings = {1 "cbq" : {2 // The path the custom config file to3 // register connections and worker pools4 "configPath" : "config.cbq",5 6 7 8 9 10 11 // The interval to poll for changes to the worker pool scaling.12 // Defaults to 0 which turns off the scheduled scaling feature.13 "scaleInterval" : 0,14 15 // continued...16 }17 };18

37. moduleSettings = {
"cbq" : {
// 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,

// continued...
}
};1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 // The path the custom config file to
// register connections and worker pools
"configPath" : "config.cbq",moduleSettings = {1 "cbq" : {2 3 4 5 6 // Flag if workers should be registered.7 // If your application only pushes to the queues,8 // you can set this to false.9 "registerWorkers" : getSystemSetting( "CBQ_REGISTER_WORKERS" , true ),10 11 // The interval to poll for changes to the worker pool scaling.12 // Defaults to 0 which turns off the scheduled scaling feature.13 "scaleInterval" : 0,14 15 // continued...16 }17 };18 // 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 ),moduleSettings = {1 "cbq" : {2 // The path the custom config file to3 // register connections and worker pools4 "configPath" : "config.cbq",5 6 7 8 9 10 11 // The interval to poll for changes to the worker pool scaling.12 // Defaults to 0 which turns off the scheduled scaling feature.13 "scaleInterval" : 0,14 15 // continued...16 }17 };18 // The interval to poll for changes to the worker pool scaling.
// Defaults to 0 which turns off the scheduled scaling feature.
"scaleInterval" : 0,moduleSettings = {1 "cbq" : {2 // The path the custom config file to3 // register connections and worker pools4 "configPath" : "config.cbq",5 6 // Flag if workers should be registered.7 // If your application only pushes to the queues,8 // you can set this to false.9 "registerWorkers" : getSystemSetting( "CBQ_REGISTER_WORKERS" , true ),10 11 12 13 14 15 // continued...16 }17 };18

38. moduleSettings = {
"cbq" : {
// 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 j
// 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,

// continued...
}
};1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17

39. moduleSettings = {
"cbq" : {
// 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 j
// 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,

// continued...
}
};1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 // The default amount of time, in seconds, to delay a job.
// Used if the connection and job doesn't define their own.
"defaultWorkerBackoff" : 0,moduleSettings = {1 "cbq" : {2 3 4 5 6 // The default amount of time, in seconds, to wait before timing out a j7 // Used if the connection and job doesn't define their own.8 "defaultWorkerTimeout" : 60,9 10 // The default amount of attempts to try before failing a job.11 // Used if the connection and job doesn't define their own.12 "defaultWorkerMaxAttempts" : 1,13 14 // continued...15 }16 };17

40. moduleSettings = {
"cbq" : {
// 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 j
// 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,

// continued...
}
};1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 // The default amount of time, in seconds, to delay a job.
// Used if the connection and job doesn't define their own.
"defaultWorkerBackoff" : 0,moduleSettings = {1 "cbq" : {2 3 4 5 6 // The default amount of time, in seconds, to wait before timing out a j7 // Used if the connection and job doesn't define their own.8 "defaultWorkerTimeout" : 60,9 10 // The default amount of attempts to try before failing a job.11 // Used if the connection and job doesn't define their own.12 "defaultWorkerMaxAttempts" : 1,13 14 // continued...15 }16 };17 // The default amount of time, in seconds, to wait before timing out a j
// Used if the connection and job doesn't define their own.
"defaultWorkerTimeout" : 60,moduleSettings = {1 "cbq" : {2 // The default amount of time, in seconds, to delay a job.3 // Used if the connection and job doesn't define their own.4 "defaultWorkerBackoff" : 0,5 6 7 8 9 10 // The default amount of attempts to try before failing a job.11 // Used if the connection and job doesn't define their own.12 "defaultWorkerMaxAttempts" : 1,13 14 // continued...15 }16 };17

41. moduleSettings = {
"cbq" : {
// 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 j
// 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,

// continued...
}
};1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 // The default amount of time, in seconds, to delay a job.
// Used if the connection and job doesn't define their own.
"defaultWorkerBackoff" : 0,moduleSettings = {1 "cbq" : {2 3 4 5 6 // The default amount of time, in seconds, to wait before timing out a j7 // Used if the connection and job doesn't define their own.8 "defaultWorkerTimeout" : 60,9 10 // The default amount of attempts to try before failing a job.11 // Used if the connection and job doesn't define their own.12 "defaultWorkerMaxAttempts" : 1,13 14 // continued...15 }16 };17 // The default amount of time, in seconds, to wait before timing out a j
// Used if the connection and job doesn't define their own.
"defaultWorkerTimeout" : 60,moduleSettings = {1 "cbq" : {2 // The default amount of time, in seconds, to delay a job.3 // Used if the connection and job doesn't define their own.4 "defaultWorkerBackoff" : 0,5 6 7 8 9 10 // The default amount of attempts to try before failing a job.11 // Used if the connection and job doesn't define their own.12 "defaultWorkerMaxAttempts" : 1,13 14 // continued...15 }16 };17 // The default amount of attempts to try before failing a job.
// Used if the connection and job doesn't define their own.
"defaultWorkerMaxAttempts" : 1,moduleSettings = {1 "cbq" : {2 // The default amount of time, in seconds, to delay a job.3 // Used if the connection and job doesn't define their own.4 "defaultWorkerBackoff" : 0,5 6 // The default amount of time, in seconds, to wait before timing out a j7 // Used if the connection and job doesn't define their own.8 "defaultWorkerTimeout" : 60,9 10 11 12 13 14 // continued...15 }16 };17

42. moduleSettings = {
"cbq" : {
// 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.
}
}
};1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22

43. moduleSettings = {
"cbq" : {
// 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.
}
}
};1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 // 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.
},moduleSettings = {1 "cbq" : {2 3 4 5 6 7 8 9 10 // Flag to turn on logging failed jobs to a database table.11 "logFailedJobs" : false,12 13 // Datasource information for loggin failed jobs.14 "logFailedJobsProperties" : {15 "tableName" : "cbq_failed_jobs",16 "datasource" : "", // `datasource` can also be a struct.17 "queryOptions" : {} // The sibling `datasource` property overrides18 // any defined datasource in queryOptions.19 }20 }21 };22

44. moduleSettings = {
"cbq" : {
// 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.
}
}
};1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 // 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.
},moduleSettings = {1 "cbq" : {2 3 4 5 6 7 8 9 10 // Flag to turn on logging failed jobs to a database table.11 "logFailedJobs" : false,12 13 // Datasource information for loggin failed jobs.14 "logFailedJobsProperties" : {15 "tableName" : "cbq_failed_jobs",16 "datasource" : "", // `datasource` can also be a struct.17 "queryOptions" : {} // The sibling `datasource` property overrides18 // any defined datasource in queryOptions.19 }20 }21 };22 // Flag to turn on logging failed jobs to a database table.
"logFailedJobs" : false,moduleSettings = {1 "cbq" : {2 // Datasource information for tracking batches.3 "batchRepositoryProperties" : {4 "tableName" : "cbq_batches",5 "datasource" : "", // `datasource` can also be a struct6 "queryOptions" : {} // The sibling `datasource` property overrides7 // any defined datasource in queryOptions.8 },9 10 11 12 13 // Datasource information for loggin failed jobs.14 "logFailedJobsProperties" : {15 "tableName" : "cbq_failed_jobs",16 "datasource" : "", // `datasource` can also be a struct.17 "queryOptions" : {} // The sibling `datasource` property overrides18 // any defined datasource in queryOptions.19 }20 }21 };22

45. moduleSettings = {
"cbq" : {
// 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.
}
}
};1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 // 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.
},moduleSettings = {1 "cbq" : {2 3 4 5 6 7 8 9 10 // Flag to turn on logging failed jobs to a database table.11 "logFailedJobs" : false,12 13 // Datasource information for loggin failed jobs.14 "logFailedJobsProperties" : {15 "tableName" : "cbq_failed_jobs",16 "datasource" : "", // `datasource` can also be a struct.17 "queryOptions" : {} // The sibling `datasource` property overrides18 // any defined datasource in queryOptions.19 }20 }21 };22 // Flag to turn on logging failed jobs to a database table.
"logFailedJobs" : false,moduleSettings = {1 "cbq" : {2 // Datasource information for tracking batches.3 "batchRepositoryProperties" : {4 "tableName" : "cbq_batches",5 "datasource" : "", // `datasource` can also be a struct6 "queryOptions" : {} // The sibling `datasource` property overrides7 // any defined datasource in queryOptions.8 },9 10 11 12 13 // Datasource information for loggin failed jobs.14 "logFailedJobsProperties" : {15 "tableName" : "cbq_failed_jobs",16 "datasource" : "", // `datasource` can also be a struct.17 "queryOptions" : {} // The sibling `datasource` property overrides18 // any defined datasource in queryOptions.19 }20 }21 };22 // 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.
}moduleSettings = {1 "cbq" : {2 // Datasource information for tracking batches.3 "batchRepositoryProperties" : {4 "tableName" : "cbq_batches",5 "datasource" : "", // `datasource` can also be a struct6 "queryOptions" : {} // The sibling `datasource` property overrides7 // any defined datasource in queryOptions.8 },9 10 // Flag to turn on logging failed jobs to a database table.11 "logFailedJobs" : false,12 13 14 15 16 17 18 19 20 }21 };22

46. cbq Config File

47. component {
function configure() {
newConnection( "default" )
.setProvider( "ColdBoxAsyncProvider@cbq" );

newConnection( "database" )
.setProvider( "DBProvider@cbq" )
.setProperties( { "tableName": "custom_jobs_table" } );

newConnection( "external" )
.setProvider( "DBProvider@cbq" )
.setProperties( { "datasource": "external" } );

newWorkerPool( "default" )
.forConnection( "default" )
.setTimeout( 5 )
.setMaxAttempts( 5 );

newWorkerPool( "priority" )
.forConnection( "database" )
.onQueues( [ "priority", "*" ] )
.setQuantity( 3 );
}
}1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24

48. component {
function configure() {
newConnection( "default" )
.setProvider( "ColdBoxAsyncProvider@cbq" );

newConnection( "database" )
.setProvider( "DBProvider@cbq" )
.setProperties( { "tableName": "custom_jobs_table" } );

newConnection( "external" )
.setProvider( "DBProvider@cbq" )
.setProperties( { "datasource": "external" } );

newWorkerPool( "default" )
.forConnection( "default" )
.setTimeout( 5 )
.setMaxAttempts( 5 );

newWorkerPool( "priority" )
.forConnection( "database" )
.onQueues( [ "priority", "*" ] )
.setQuantity( 3 );
}
}1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 newConnection( "default" )
.setProvider( "ColdBoxAsyncProvider@cbq" );
newWorkerPool( "default" )
.forConnection( "default" )
.setTimeout( 5 )
.setMaxAttempts( 5 );component {1 function configure() {2 3 4 5 newConnection( "database" )6 .setProvider( "DBProvider@cbq" )7 .setProperties( { "tableName": "custom_jobs_table" } );8 9 newConnection( "external" )10 .setProvider( "DBProvider@cbq" )11 .setProperties( { "datasource": "external" } );12 13 14 15 16 17 18 newWorkerPool( "priority" )19 .forConnection( "database" )20 .onQueues( [ "priority", "*" ] )21 .setQuantity( 3 );22 }23 }24

49. component {
function configure() {
newConnection( "default" )
.setProvider( "ColdBoxAsyncProvider@cbq" );

newConnection( "database" )
.setProvider( "DBProvider@cbq" )
.setProperties( { "tableName": "custom_jobs_table" } );

newConnection( "external" )
.setProvider( "DBProvider@cbq" )
.setProperties( { "datasource": "external" } );

newWorkerPool( "default" )
.forConnection( "default" )
.setTimeout( 5 )
.setMaxAttempts( 5 );

newWorkerPool( "priority" )
.forConnection( "database" )
.onQueues( [ "priority", "*" ] )
.setQuantity( 3 );
}
}1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 newConnection( "default" )
.setProvider( "ColdBoxAsyncProvider@cbq" );
newWorkerPool( "default" )
.forConnection( "default" )
.setTimeout( 5 )
.setMaxAttempts( 5 );component {1 function configure() {2 3 4 5 newConnection( "database" )6 .setProvider( "DBProvider@cbq" )7 .setProperties( { "tableName": "custom_jobs_table" } );8 9 newConnection( "external" )10 .setProvider( "DBProvider@cbq" )11 .setProperties( { "datasource": "external" } );12 13 14 15 16 17 18 newWorkerPool( "priority" )19 .forConnection( "database" )20 .onQueues( [ "priority", "*" ] )21 .setQuantity( 3 );22 }23 }24 newConnection( "database" )
.setProvider( "DBProvider@cbq" )
.setProperties( { "tableName": "custom_jobs_table" } );
newWorkerPool( "priority" )
.forConnection( "database" )
.onQueues( [ "priority", "*" ] )
.setQuantity( 3 );component {1 function configure() {2 newConnection( "default" )3 .setProvider( "ColdBoxAsyncProvider@cbq" );4 5 6 7 8 9 newConnection( "external" )10 .setProvider( "DBProvider@cbq" )11 .setProperties( { "datasource": "external" } );12 13 newWorkerPool( "default" )14 .forConnection( "default" )15 .setTimeout( 5 )16 .setMaxAttempts( 5 );17 18 19 20 21 22 }23 }24

50. component {
function configure() {
newConnection( "default" )
.setProvider( "ColdBoxAsyncProvider@cbq" );

newConnection( "database" )
.setProvider( "DBProvider@cbq" )
.setProperties( { "tableName": "custom_jobs_table" } );

newConnection( "external" )
.setProvider( "DBProvider@cbq" )
.setProperties( { "datasource": "external" } );

newWorkerPool( "default" )
.forConnection( "default" )
.setTimeout( 5 )
.setMaxAttempts( 5 );

newWorkerPool( "priority" )
.forConnection( "database" )
.onQueues( [ "priority", "*" ] )
.setQuantity( 3 );
}
}1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 newConnection( "default" )
.setProvider( "ColdBoxAsyncProvider@cbq" );
newWorkerPool( "default" )
.forConnection( "default" )
.setTimeout( 5 )
.setMaxAttempts( 5 );component {1 function configure() {2 3 4 5 newConnection( "database" )6 .setProvider( "DBProvider@cbq" )7 .setProperties( { "tableName": "custom_jobs_table" } );8 9 newConnection( "external" )10 .setProvider( "DBProvider@cbq" )11 .setProperties( { "datasource": "external" } );12 13 14 15 16 17 18 newWorkerPool( "priority" )19 .forConnection( "database" )20 .onQueues( [ "priority", "*" ] )21 .setQuantity( 3 );22 }23 }24 newConnection( "database" )
.setProvider( "DBProvider@cbq" )
.setProperties( { "tableName": "custom_jobs_table" } );
newWorkerPool( "priority" )
.forConnection( "database" )
.onQueues( [ "priority", "*" ] )
.setQuantity( 3 );component {1 function configure() {2 newConnection( "default" )3 .setProvider( "ColdBoxAsyncProvider@cbq" );4 5 6 7 8 9 newConnection( "external" )10 .setProvider( "DBProvider@cbq" )11 .setProperties( { "datasource": "external" } );12 13 newWorkerPool( "default" )14 .forConnection( "default" )15 .setTimeout( 5 )16 .setMaxAttempts( 5 );17 18 19 20 21 22 }23 }24 newConnection( "external" )
.setProvider( "DBProvider@cbq" )
.setProperties( { "datasource": "external" } );component {1 function configure() {2 newConnection( "default" )3 .setProvider( "ColdBoxAsyncProvider@cbq" );4 5 newConnection( "database" )6 .setProvider( "DBProvider@cbq" )7 .setProperties( { "tableName": "custom_jobs_table" } );8 9 10 11 12 13 newWorkerPool( "default" )14 .forConnection( "default" )15 .setTimeout( 5 )16 .setMaxAttempts( 5 );17 18 newWorkerPool( "priority" )19 .forConnection( "database" )20 .onQueues( [ "priority", "*" ] )21 .setQuantity( 3 );22 }23 }24

51. Usage

52. Create a
Job

53. // SendWelcomeEmailJob.cfc
component extends="cbq.models.Jobs.AbstractJob" {

property name="mailService" inject="provider:MailService@cbmailservices" ;

property name="userId";

function handle() {
var user = getInstance( "User" ).findOrFail( getUserId() );

variables.mailService
.newMail(
from = "[email protected]" ,
to = user.getEmail(),
subject = "Welcome!",
type = "html"
)
.setView( "/_emails/users/welcome" )
.setBodyTokens( {
"firstName" : user.getFirstName(),
"lastName" : user.getLastName()
} )
.send();
}

}1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26

54. Per-Job
Configuration

55. // SendWelcomeEmailJob.cfc
component extends="cbq.models.Jobs.AbstractJob" {

variables.connection = "emails";
variables.maxAttempts = 3;
variables.timeout = 10;

function handle() {
// ...
}

}1 2 3 4 5 6 7 8 9 10 11 12

56. // SendWelcomeEmailJob.cfc
component extends="cbq.models.Jobs.AbstractJob" {

variables.connection = "emails";
variables.maxAttempts = 3;
variables.timeout = 10;

function handle() {
// ...
}

}1 2 3 4 5 6 7 8 9 10 11 12 variables.connection = "emails";
variables.maxAttempts = 3;
variables.timeout = 10;// SendWelcomeEmailJob.cfc1 component extends="cbq.models.Jobs.AbstractJob" {2 3 4 5 6 7 function handle() {8 // ...9 }10 11 }12

57. Dispatching
Jobs

58. // handlers/Main.cfc
component {

function create() {
var user = createUser( rc );

getInstance( "SendWelcomeEmailJob" )
.setProperties( { "userId": user.getId() } )
.dispatch();
}

}1 2 3 4 5 6 7 8 9 10 11 12

59. // handlers/Main.cfc
component {

function create() {
var user = createUser( rc );

getInstance( "SendWelcomeEmailJob" )
.setProperties( { "userId": user.getId() } )
.dispatch();
}

}1 2 3 4 5 6 7 8 9 10 11 12 getInstance( "SendWelcomeEmailJob" )// handlers/Main.cfc1 component {2 3 function create() {4 var user = createUser( rc );5 6 7 .setProperties( { "userId": user.getId() } )8 .dispatch();9 }10 11 }12

60. // handlers/Main.cfc
component {

function create() {
var user = createUser( rc );

getInstance( "SendWelcomeEmailJob" )
.setProperties( { "userId": user.getId() } )
.dispatch();
}

}1 2 3 4 5 6 7 8 9 10 11 12 getInstance( "SendWelcomeEmailJob" )// handlers/Main.cfc1 component {2 3 function create() {4 var user = createUser( rc );5 6 7 .setProperties( { "userId": user.getId() } )8 .dispatch();9 }10 11 }12 getInstance( "SendWelcomeEmailJob" )
.setProperties( { "userId": user.getId() } )// handlers/Main.cfc1 component {2 3 function create() {4 var user = createUser( rc );5 6 7 8 .dispatch();9 }10 11 }12

61. // handlers/Main.cfc
component {

function create() {
var user = createUser( rc );

getInstance( "SendWelcomeEmailJob" )
.setProperties( { "userId": user.getId() } )
.dispatch();
}

}1 2 3 4 5 6 7 8 9 10 11 12 getInstance( "SendWelcomeEmailJob" )// handlers/Main.cfc1 component {2 3 function create() {4 var user = createUser( rc );5 6 7 .setProperties( { "userId": user.getId() } )8 .dispatch();9 }10 11 }12 getInstance( "SendWelcomeEmailJob" )
.setProperties( { "userId": user.getId() } )// handlers/Main.cfc1 component {2 3 function create() {4 var user = createUser( rc );5 6 7 8 .dispatch();9 }10 11 }12 getInstance( "SendWelcomeEmailJob" )
.setProperties( { "userId": user.getId() } )
.dispatch();// handlers/Main.cfc1 component {2 3 function create() {4 var user = createUser( rc );5 6 7 8 9 }10 11 }12

62. Setting up
Worker Pools

63. // config/cbq.cfc
component {
function configure() {
newConnection( "default" ).setProvider( "DBProvider@cbq" );

newWorkerPool( "default" )
.forConnection( "default" )
.setTimeout( 30 )
.setMaxAttempts( 5 )
.setQuantity( 3 );

newWorkerPool( "default" )
.forConnection( "default" )
.onQueues( "emails" )
.setTimeout( 60 )
.setMaxAttempts( 10 )
.setQuantity( 10 );

}
}1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20
moduleSettings = {
"cbq" : {
"registerWorkers" : true // this is also the default
}
};1 2 3 4 5

64. cbq Model
Helper

65. // handlers/Main.cfc
component {

property name="cbq" inject="cbq@cbq";

function create() {
var user = createUser( rc );

cbq.job( "SendWelcomeEmailJob" , {
"userId": user.getId()
} ).dispatch();
}

}1 2 3 4 5 6 7 8 9 10 11 12 13 14

66. Job Chains

67. // 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();
}

}1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18

68. // 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();
}

}1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 getInstance( "SendProductLinkEmail" )
.onConnection( "fulfillment" )
.setProperties( {
"productId" : productId,
"userId" : getProperties().userId
} )
.dispatch();// FulfillOrderJob.cfc1 component extends="cbq.models.Jobs.AbstractJob" {2 3 function handle() {4 var productId = getProperties().productId;5 6 processPayment( productId );7 8 9 10 11 12 13 14 15 }16 17 }18

69. // handlers/Main.cfc
component {

function create() {
getInstance( "FulfillOrderJob" )
.setProperties( { "productId": rc.productId } )
.chain( [
getInstance( "SendProductLinkEmail" )
.onConnection( "fulfillment" )
.setProperties( {
"productId" : rc.productId,
"userId" : auth().getUserId()
} )
] )
}

}1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17

70. // 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();
}

}1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16

71. Job Batches

72. // handlers/Main.cfc
component {

property name="cbq" inject="cbq@cbq";

function create() {
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();
}

}1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19

73. // handlers/Main.cfc
component {

property name="cbq" inject="cbq@cbq";

function create() {
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();
}

}1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 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 } )// handlers/Main.cfc1 component {2 3 property name="cbq" inject="cbq@cbq";4 5 function create() {6 cbq.batch( [7 8 9 10 11 12 ] ).then( cbq.job( "ImportCsvSuccessfulJob" ) )13 .catch( cbq.job( "ImportCsvFailedJob" ) )14 .finally( cbq.job( "ImportCsvCompletedJob" ) )15 .dispatch();16 }17 18 }19

74. // handlers/Main.cfc
component {

property name="cbq" inject="cbq@cbq";

function create() {
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();
}

}1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 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 } )// handlers/Main.cfc1 component {2 3 property name="cbq" inject="cbq@cbq";4 5 function create() {6 cbq.batch( [7 8 9 10 11 12 ] ).then( cbq.job( "ImportCsvSuccessfulJob" ) )13 .catch( cbq.job( "ImportCsvFailedJob" ) )14 .finally( cbq.job( "ImportCsvCompletedJob" ) )15 .dispatch();16 }17 18 }19 ] ).then( cbq.job( "ImportCsvSuccessfulJob" ) )
.catch( cbq.job( "ImportCsvFailedJob" ) )
.finally( cbq.job( "ImportCsvCompletedJob" ) )// handlers/Main.cfc1 component {2 3 property name="cbq" inject="cbq@cbq";4 5 function create() {6 cbq.batch( [7 cbq.job( "ImportCsvJob", { "start": 1, "end": 100 } ),8 cbq.job( "ImportCsvJob", { "start": 101, "end": 200 } ),9 cbq.job( "ImportCsvJob", { "start": 201, "end": 300 } ),10 cbq.job( "ImportCsvJob", { "start": 301, "end": 400 } ),11 cbq.job( "ImportCsvJob", { "start": 401, "end": 500 } )12 13 14 15 .dispatch();16 }17 18 }19

75. Demo

76. Bonus

77. Delay the
Current Job

78. // MonitorPendingCartJob.cfc
component extends="cbq.models.Jobs.AbstractJob" {
function handle() {
var cartId = getProperties().cartId;
var cart = getInstance( "Cart" ).findOrFail( cartId );

if ( cart.isConfirmed() || cart.isCancelled() ) {
return;
}

if ( abs( dateDiff( "n", cart.getModifiedDate(), now() ) ) > 60 ) {
cart.cancel();
return;
}

getInstance( "MonitorPendingCartJob" )
.setProperties( { "cartId" : cartId } )
.delay( 15 * 1000 )
.dispatch();
}
}1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21

79. // MonitorPendingCartJob.cfc
component extends="cbq.models.Jobs.AbstractJob" {
function handle() {
var cartId = getProperties().cartId;
var cart = getInstance( "Cart" ).findOrFail( cartId );

if ( cart.isConfirmed() || cart.isCancelled() ) {
return;
}

if ( abs( dateDiff( "n", cart.getModifiedDate(), now() ) ) > 60 ) {
cart.cancel();
return;
}

getInstance( "MonitorPendingCartJob" )
.setProperties( { "cartId" : cartId } )
.delay( 15 * 1000 )
.dispatch();
}
}1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 .setProperties( { "cartId" : cartId } )
.delay( 15 * 1000 )
.dispatch();
}// MonitorPendingCartJob.cfc1 component extends="cbq.models.Jobs.AbstractJob" {2 function handle() {3 var cartId = getProperties().cartId;4 var cart = getInstance( "Cart" ).findOrFail( cartId );5 6 if ( cart.isConfirmed() || cart.isCancelled() ) {7 return;8 }9 10 if ( abs( dateDiff( "n", cart.getModifiedDate(), now() ) ) > 60 ) {11 cart.cancel();12 return;13 }14 15 getInstance( "MonitorPendingCartJob" )16 17 18 19 20 }21

80. // MonitorPendingCartJob.cfc
component extends="cbq.models.Jobs.AbstractJob" {
function handle() {
var cartId = getProperties().cartId;
var cart = getInstance( "Cart" ).findOrFail( cartId );

if ( cart.isConfirmed() || cart.isCancelled() ) {
return;
}

if ( abs( dateDiff( "n", cart.getModifiedDate(), now() ) ) > 60 ) {
cart.cancel();
return;
}

this.release( 15 * 1000 );
}
}1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18
Using `release`

81. // MonitorPendingCartJob.cfc
component extends="cbq.models.Jobs.AbstractJob" {
function handle() {
var cartId = getProperties().cartId;
var cart = getInstance( "Cart" ).findOrFail( cartId );

if ( cart.isConfirmed() || cart.isCancelled() ) {
return;
}

if ( abs( dateDiff( "n", cart.getModifiedDate(), now() ) ) > 60 ) {
cart.cancel();
return;
}

this.release( 15 * 1000 );
}
}1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 }
}// MonitorPendingCartJob.cfc1 component extends="cbq.models.Jobs.AbstractJob" {2 function handle() {3 var cartId = getProperties().cartId;4 var cart = getInstance( "Cart" ).findOrFail( cartId );5 6 if ( cart.isConfirmed() || cart.isCancelled() ) {7 return;8 }9 10 if ( abs( dateDiff( "n", cart.getModifiedDate(), now() ) ) > 60 ) {11 cart.cancel();12 return;13 }14 15 this.release( 15 * 1000 );16 17 18
Using `release`

82. // MonitorPendingCartJob.cfc
component extends="cbq.models.Jobs.AbstractJob" {
function handle() {
var cartId = getProperties().cartId;
var cart = getInstance( "Cart" ).findOrFail( cartId );

if ( cart.isConfirmed() || cart.isCancelled() ) {
return;
}

if ( abs( dateDiff( "n", cart.getModifiedDate(), now() ) ) > 60 ) {
cart.cancel();
return;
}

var delay = 2 ^ this.getCurrentAttempt() * 1000;

this.release( delay );
}
}1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20
Exponential Backoff

83. // MonitorPendingCartJob.cfc
component extends="cbq.models.Jobs.AbstractJob" {
function handle() {
var cartId = getProperties().cartId;
var cart = getInstance( "Cart" ).findOrFail( cartId );

if ( cart.isConfirmed() || cart.isCancelled() ) {
return;
}

if ( abs( dateDiff( "n", cart.getModifiedDate(), now() ) ) > 60 ) {
cart.cancel();
return;
}

var delay = 2 ^ this.getCurrentAttempt() * 1000;

this.release( delay );
}
}1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20
this.release( delay );
}
}// MonitorPendingCartJob.cfc1 component extends="cbq.models.Jobs.AbstractJob" {2 function handle() {3 var cartId = getProperties().cartId;4 var cart = getInstance( "Cart" ).findOrFail( cartId );5 6 if ( cart.isConfirmed() || cart.isCancelled() ) {7 return;8 }9 10 if ( abs( dateDiff( "n", cart.getModifiedDate(), now() ) ) > 60 ) {11 cart.cancel();12 return;13 }14 15 var delay = 2 ^ this.getCurrentAttempt() * 1000;16 17 18 19 20
Exponential Backoff