com.climate.squeedo.sqs-consumer
Functions for using Amazon Simple Queueing Service to request and perform computation.
Functions for using Amazon Simple Queueing Service to request and perform computation.
start-consumerclj
(start-consumer queue-name compute & opts)Creates a consumer that reads messages as quickly as possible into a local buffer up to the configured buffer size.
Work is done asynchronously by workers controlled by the size of the work buffer (currently hardcoded to number of cpus minus 1) Bad things happen if you have workers > number of cpus.
The client has responsibility for calling stop-consumer when you no longer want to process messages. Additionally, the client MUST send the message back on the done-channel when processing is complete or if an uncaught exception occurs the message will be auto-nack'd in SQS.
Failed messages will currently not be acked, but rather go back on for redelivery after the timeout.
This code is atom free :)
Input: queue-name - the name of an sqs queue compute - a compute function that takes two args: a 'message' containing the body of the sqs message and a channel on which to ack/nack when done.
Optional arguments: :message-channel-size - the number of messages to prefetch from sqs; default 20 * num-listeners :num-workers - the number of workers processing messages concurrently :num-listeners - the number of listeners polling from sqs. default is (num-workers / 10) since each listener dequeues up to 10 messages at a time :listener-threads? - run listeners in dedicated threads. if true, will create one thread per listener. :dequeue-limit - the number of messages to dequeue at a time; default 10 :max-concurrent-work - the maximum number of total messages processed. This is mainly for asynch workflows; default num-workers :client - the sqs client to use (if missing, sqs/mk-connection will create the client) Output: a map with keys, :done-channel - the channel to send messages to be acked :message-channel - unused by the client.
Creates a consumer that reads messages as quickly as possible into a local buffer up
to the configured buffer size.
Work is done asynchronously by workers controlled by the size of the work buffer (currently
hardcoded to number of cpus minus 1) Bad things happen if you have workers > number of cpus.
The client has responsibility for calling stop-consumer when you no longer want to process messages.
Additionally, the client MUST send the message back on the done-channel when processing is complete
or if an uncaught exception occurs the message will be auto-nack'd in SQS.
Failed messages will currently not be acked, but rather go back on for redelivery after the timeout.
This code is atom free :)
Input:
queue-name - the name of an sqs queue
compute - a compute function that takes two args: a 'message' containing the body of the sqs
message and a channel on which to ack/nack when done.
Optional arguments:
:message-channel-size - the number of messages to prefetch from sqs; default 20 * num-listeners
:num-workers - the number of workers processing messages concurrently
:num-listeners - the number of listeners polling from sqs. default is (num-workers / 10)
since each listener dequeues up to 10 messages at a time
:listener-threads? - run listeners in dedicated threads. if true, will create
one thread per listener.
:dequeue-limit - the number of messages to dequeue at a time; default 10
:max-concurrent-work - the maximum number of total messages processed. This is mainly for
asynch workflows; default num-workers
:client - the sqs client to use (if missing, sqs/mk-connection will create
the client)
Output:
a map with keys, :done-channel - the channel to send messages to be acked
:message-channel - unused by the client.stop-consumerclj
(stop-consumer {:keys [done-channel message-channel]})Takes a consumer created by start-consumer and closes the channels. This should be called to stopped consuming messages.
Takes a consumer created by start-consumer and closes the channels. This should be called to stopped consuming messages.
cljdoc is a website building & hosting documentation for Clojure/Script libraries
× close