A PGMQ library written in Clojure
com.thirstysink.pgmq-clj.corearchive-messages- Archives messagesmsg-idsin a queue namedqueue-name.create-queue- Create a queue namedqueue-name.delete-message- Permanently deletes message with idmsg-idin the queue namedqueue-name.delete-message-batch- Deletes allmsg-idsmessages in queuequeue-name.drop-queue- Drop queue namedqueue-name.list-queues- List all queues.pop-message- Pops one message from the queue namedqueue-name.read-message- Read aquantityof messages fromqueue-namemarking them invisible forvisible_timeseconds.send-message- Send one message to a queuequeue-namewith apayloadthat will not be read fordelayseconds.send-message-batch- Sendspayloadto the queue namedqueue-nameas a collection of messages that cannot be read fordelayseconds.
com.thirstysink.pgmq-clj.db.adapterAdapterclose- Performs database connection cleanup.execute!- Execute a SQL statement with 0 or more return values.execute-one!- Execute a SQL statement with 0 or 1 return values.query- Query the database and return results.with-transaction- Wrap a function in a database transaction.
com.thirstysink.pgmq-clj.db.adapters.hikari-adapter->pgobject- Transforms Clojure data to a PGobject that contains the data as JSON.<-pgobject- Transform PGobject containingjsonorjsonbvalue to Clojure data.ensure-pgmq-extension- Checks the database to verify that thepgmqextension is installed.make-hikari-adapter- Create a new HikariAdapter instance.
com.thirstysink.pgmq-clj.instrumentationdisable-instrumentation- Disablesclojure.specs.alphaspecs instrumentation.enable-instrumentation- Enablesclojure.specs.alphaspecs instrumentation.instrumentation-enabled?
com.thirstysink.pgmq-clj.jsoncom.thirstysink.pgmq-clj.specs
(archive-messages adapter queue-name msg-ids)Function.
Archives messages msg-ids in a queue named queue-name. This will remove the message from
queue-name and place it in a archive table which is named a_{queue-name}.
(create-queue adapter queue-name)Function.
Create a queue named queue-name.
(delete-message adapter queue-name msg-id)Function.
Permanently deletes message with id msg-id in the queue named queue-name.
(delete-message-batch adapter queue-name msg-ids)Function.
Deletes all msg-ids messages in queue queue-name.
(drop-queue adapter queue-name)Function.
Drop queue named queue-name.
(list-queues adapter)Function.
List all queues.
(pop-message adapter queue-name)Function.
Pops one message from the queue named queue-name. The side-effect of
this function is equivalent to reading and deleting a message. See also
[[read-message]] and [[delete-message]].
(read-message adapter queue-name visible_time quantity filter)Function.
Read a quantity of messages from queue-name marking them invisible for
visible_time seconds. This function supports the ability to filter messages
received when making a read request.
Here are some examples of how this conditional works: If conditional is an empty JSON object ('{}'::jsonb), the condition always evaluates to TRUE, and all messages are considered matching.
If conditional is a JSON object with a single key-value pair, such as {'key': 'value'}, the condition checks if the message column contains a JSON object with the same key-value pair. For example:
message = {'key': 'value', 'other_key': 'other_value'}: // matches
message = {'other_key': 'other_value'}: // does not match
If conditional is a JSON object with multiple key-value pairs, such as {'key1': 'value1', 'key2': 'value2'}, the condition checks if the message column contains a JSON object with all the specified key-value pairs. For example:
message = {'key1': 'value1', 'key2': 'value2', 'other_key': 'other_value'}: // matches
message = {'key1': 'value1', 'other_key': 'other_value'}: // does not match
Some examples of conditional JSONB values and their effects on the query:
{}: matches all messages{'type': 'error'}: matches messages with a type key equal to 'error'{'type': 'error', 'severity': 'high'}: matches messages with both type equal to 'error' and severity equal to 'high'{'user_id': 123}: matches messages with a user_id key equal to 123
(send-message adapter queue-name payload delay)Function.
Send one message to a queue queue-name with a payload
that will not be read for delay seconds. A delay of 0
indicates it may be read immediately.
Example Payloads:
[{:data {:foo "bad"} :headers {:x-data "baz"}}][{:data 10022 :headers {}} {:data "feed" :headers {:version "3"}}]
(send-message-batch adapter queue-name payload delay)Function.
Sends payload to the queue named queue-name as a collection of messages
that cannot be read for delay seconds. The payload should be a sequence
of valid JSON objects. See also [[send-message]].
Example Payloads:
[{:data {:foo "bar"} :headers {:x-data "bat"}}][{:data 10002 :headers {}} {:data "feed" :headers {:version "2"}} ]
(close this)Function.
Performs database connection cleanup.
(execute! this sql params)Function.
Execute a SQL statement with 0 or more return values.
(execute-one! this sql params)Function.
Execute a SQL statement with 0 or 1 return values.
(query this sql params)Function.
Query the database and return results.
(with-transaction this f)Function.
Wrap a function in a database transaction.
(->pgobject x)Function.
Transforms Clojure data to a PGobject that contains the data as
JSON. PGObject type defaults to jsonb but can be changed via
metadata key :pgtype
(<-pgobject v)Function.
Transform PGobject containing json or jsonb value to Clojure data.
(ensure-pgmq-extension adapter)Function.
Checks the database to verify that the pgmq extension is installed.
If it is not then it will throw an exception.
(make-hikari-adapter config)Function.
Create a new HikariAdapter instance. The argument config provides database connection values. See https://github.com/tomekw/hikari-cp for additional details on the configuration options.
| Setting | Description |
|---|---|
| JdbcUrl | This property sets the JDBC connection URL. |
| Username | This property sets the default authentication username used when obtaining Connections from the underlying driver. |
| Password | This property sets the default authentication password used when obtaining Connections from the underlying driver. |
| MaximumPoolSize | This property controls the maximum size that the pool is allowed to reach, including both idle and in-use connections. |
| MinimumIdle | This property controls the minimum number of idle connections that HikariCP tries to maintain in the pool. |
(disable-instrumentation)
(disable-instrumentation ns)Function.
Disables clojure.specs.alpha specs instrumentation.
Learn more. If
no namespace is provided it will disable instrumentation
for com.thirstysink.pgmq-clj.core.
(enable-instrumentation)
(enable-instrumentation ns)Function.
Enables clojure.specs.alpha specs instrumentation.
Learn more. If
no namespace is provided it will instrument com.thirstysink.pgmq-clj.core.
(coll-of :com.thirstysink.pgmq-clj.specs/mesage-record)(and int? (> % 0))(keys :req-un [:com.thirstysink.pgmq-clj.specs/queue-name :com.thirstysink.pgmq-clj.specs/is-partitioned :com.thirstysink.pgmq-clj.specs/is-unlogged :com.thirstysink.pgmq-clj.specs/created-at])(fspec :args (cat :adapter :com.thirstysink.pgmq-clj.specs/adapter :queue-name :com.thirstysink.pgmq-clj.specs/queue-name :payload :com.thirstysink.pgmq-clj.specs/payload-object :delay :com.thirstysink.pgmq-clj.specs/delay) :ret :com.thirstysink.pgmq-clj.specs/msg-id :fn nil)(fn [x] (or (map? x) (vector? x) (string? x) (number? x) (boolean? x) (nil? x)))(instance? java.time.Instant %)(or :string string? :number number? :list (coll-of (or :string string? :number number?)))(fspec :args (cat :adapter :com.thirstysink.pgmq-clj.specs/adapter) :ret :com.thirstysink.pgmq-clj.specs/queue-result :fn nil)(keys :req-un [:com.thirstysink.pgmq-clj.specs/msg-id :com.thirstysink.pgmq-clj.specs/read-ct :com.thirstysink.pgmq-clj.specs/enqueued-at :com.thirstysink.pgmq-clj.specs/vt :com.thirstysink.pgmq-clj.specs/message] :opt-un [:com.thirstysink.pgmq-clj.specs/headers])(fspec :args (cat :adapter :com.thirstysink.pgmq-clj.specs/adapter :queue-name :com.thirstysink.pgmq-clj.specs/queue-name) :ret boolean? :fn nil)int?(satisfies? Adapter %)(keys :req-un [:com.thirstysink.pgmq-clj.specs/data :com.thirstysink.pgmq-clj.specs/headers])(coll-of :com.thirstysink.pgmq-clj.specs/payload-object)(fspec :args (cat :adapter :com.thirstysink.pgmq-clj.specs/adapter :queue-name :com.thirstysink.pgmq-clj.specs/queue-name :msg-ids :com.thirstysink.pgmq-clj.specs/non-empty-msg-ids) :ret :com.thirstysink.pgmq-clj.specs/msg-ids :fn nil)(and int? (>= % 0))(fspec :args (cat :adapter :com.thirstysink.pgmq-clj.specs/adapter :queue-name :com.thirstysink.pgmq-clj.specs/queue-name) :ret :com.thirstysink.pgmq-clj.specs/message-record :fn nil)(coll-of :com.thirstysink.pgmq-clj.specs/queue-record)(and :com.thirstysink.pgmq-clj.specs/msg-ids (complement empty?))(fspec :args (cat :adapter :com.thirstysink.pgmq-clj.specs/adapter :queue-name :com.thirstysink.pgmq-clj.specs/queue-name) :ret nil :fn nil)boolean?valid-queue-name?(fspec :args (cat :adapter :com.thirstysink.pgmq-clj.specs/adapter :queue-name :com.thirstysink.pgmq-clj.specs/queue-name :msg-ids :com.thirstysink.pgmq-clj.specs/msg-ids) :ret :com.thirstysink.pgmq-clj.specs/msg-ids :fn nil)(nilable (map-of :com.thirstysink.pgmq-clj.specs/header-key :com.thirstysink.pgmq-clj.specs/header-value :min-count 0))(fn [x] (fn* [] (instance? java.time.Instant x)))(fspec :args (cat :adapter :com.thirstysink.pgmq-clj.specs/adapter :queue-name :com.thirstysink.pgmq-clj.specs/queue-name :payload :com.thirstysink.pgmq-clj.specs/payload-objects :delay :com.thirstysink.pgmq-clj.specs/delay) :ret :com.thirstysink.pgmq-clj.specs/msg-ids :fn nil)(coll-of :com.thirstysink.pgmq-clj.specs/msg-id)(fn [x] (or (map? x) (vector? x) (string? x) (number? x) (boolean? x) (nil? x)))(fn [x] (or (map? x) (vector? x) (string? x) (number? x) (boolean? x) (nil? x)))(conformer (zipmap (map :clojure.spec.alpha/k %) (map :clojure.spec.alpha/v %)) (map (fn [[k v]] #:clojure.spec.alpha{:k k, :v v}) %))(fspec :args (cat :adapter :com.thirstysink.pgmq-clj.specs/adapter :queue-name :com.thirstysink.pgmq-clj.specs/queue-name :visibility_time :com.thirstysink.pgmq-clj.specs/visibility_time :quantity :com.thirstysink.pgmq-clj.specs/quantity :filter :com.thirstysink.pgmq-clj.specs/json) :ret :com.thirstysink.pgmq-clj.specs/message-records :fn nil)int?(instance? java.time.Instant %)(and number? pos?)(instance? java.time.Instant %)boolean?(or :string string? :keyword keyword?)(fspec :args (cat :adapter :com.thirstysink.pgmq-clj.specs/adapter :queue-name :com.thirstysink.pgmq-clj.specs/queue-name :msg-id :com.thirstysink.pgmq-clj.specs/msg-id) :ret boolean? :fn nil)
