Carmine Redis Client

When calling multiple commands inside a single wcar block, Carmine automatically uses Redis pipelining for efficiency. The result is returned as a vector containing the replies for each command.

Handling variable command counts

If the number of commands might vary, use the :as-pipeline keyword to ensure Carmine always returns a vector (pipeline-style reply), even for a single command.

Error handling in pipelines

• If a single command fails in a standard wcar block, an exception is thrown.
• If a command fails within a pipeline, the exception is captured and returned as an element within the result vector, allowing the rest of the pipeline to complete.

;; Standard pipelining (returns vector of replies)
(wcar* 
  (car/ping) 
  (car/set "foo" "bar") 
  (car/get "foo")) 
;; => ["PONG" "OK" "bar"]

;; Force pipeline-style reply for single commands
(wcar* :as-pipeline (car/ping)) 
;; => ["PONG"]

;; Error handling in a pipeline
(wcar*
  (car/set "foo" "bar")
  (car/spop "foo") ; This might fail if 'foo' is a string, not a set
  (car/get "foo"))
;; => ["OK" #<Exception ERR Operation against ...> "bar"]

Overview

Note: Tundra is deprecated and will only be supported until Carmine 4.

Tundra is a mechanism to relax Redis memory limitations by offloading "cold" data to a secondary datastore (like S3, Disk, or DynamoDB) while keeping "hot" data in memory.

Core Workflow

1. Mark data as dirty: Call tundra/dirty whenever you modify or create keys that should be evictable. This queues them for replication.
2. Run a worker: Use tundra/worker to start a threaded worker that automatically replicates dirty keys to your secondary datastore.
3. Access data: Use tundra/ensure-ks when accessing keys. This will extend their TTL in Redis or fetch them from the secondary datastore if they have been evicted.

Supported Datastores

• Disk
• Amazon S3
• Amazon DynamoDB (via Faraday)

(:require [taoensso.carmine.tundra :as tundra :refer (ensure-ks dirty)]
          [taoensso.carmine.tundra.s3])

(def my-tundra-store
  (tundra/tundra-store
    (taoensso.carmine.tundra.s3/s3-datastore {:access-key "" :secret-key ""}
      "my-bucket/my-folder")))

;; Usage:
(tundra/worker my-tundra-store {} {})
(tundra/dirty my-tundra-store "foo:bar1")
(tundra/ensure-ks my-tundra-store "foo:bar1")

Carmine provides a distributed message queue built on top of Redis. It allows you to define workers that process messages from named queues.

Core API

• worker: Returns a worker for a named queue. The returned worker can be dereferenced to view detailed status and statistics.
• enqueue: Adds a message to a specific named queue for processing by active workers.
• queue-status: Returns a detailed status map for a specific named queue.

Message Semantics

• Persistence: Messages are durable based on your Redis configuration.
• Delivery: Messages are handled once and only once using a lock-based mechanism.
• Ordering: Messages are handled in loose order. Exact ordering is not guaranteed due to concurrent handler threads, retries, and backoff logic.
• Fault Tolerance: Messages are preserved in Redis until they are explicitly acknowledged as handled.
• De-duplication: Supports optional per-message de-duplication to prevent the same message from being queued multiple times within a configurable backoff period.
• Serialization: Messages are serialized using Nippy and stored as byte strings in Redis hashes. The maximum size per message is 512MiB, though small maps or pointers to larger data stores are recommended.

;; Example of creating a worker and enqueuing a message
(def my-conn-opts {:pool {<opts>} :spec {<opts>}})

(def my-worker
  (car-mq/worker my-conn-opts "my-queue"
    {:handler
     (fn [{:keys [message attempt]}]
       (try
         (println "Received" message)
         {:status :success}
         (catch Throwable _
           (println "Handler error!")
           {:status :retry}))}))

(car-mq/enqueue "my-queue" "my message!")

To use Carmine, you typically define a connection pool and a connection spec for your Redis server. The connection spec uses a URI format.

Once configured, you pass these options to the wcar (with Carmine) API, which is the primary entry point for executing commands.

To simplify repeated calls, you can define a wcar* macro that captures your configuration.

(ns my-app (:require [taoensso.carmine :as car :refer [wcar]]))

;; 1. Create a stateful connection pool
(defonce my-conn-pool (car/connection-pool {}))

;; 2. Define the connection spec (URI)
(def my-conn-spec {:uri "redis://redistogo:pass@panga.redistogo.com:9475/"})

;; 3. Combine them into options
(def my-wcar-opts {:pool my-conn-pool, :spec my-conn-spec})

;; 4. Use the wcar API
(wcar my-wcar-opts (car/ping)) ; => "PONG"

;; Optional: Create a convenience macro
(defmacro wcar* [& body] `(car/wcar my-wcar-opts ~@body))

Carmine features a flexible listener API for Redis Pub/Sub and monitoring. Listeners are connection-local and allow you to map specific channels or patterns to handler functions.

Workflow:

1. Create a listener: Use car/with-new-pubsub-listener with a map of channel/pattern strings to handler functions. Initial subscriptions are ready immediately upon return.
2. Manage subscriptions: Use car/with-open-listener to perform asynchronous updates like car/subscribe, car/psubscribe, or car/unsubscribe.
3. Update handlers: You can dynamically update handlers by swapping the :state map in the listener object.
4. Cleanup: Always call car/close-listener when finished to prevent resource leaks. Closure is silent and does not trigger handlers.

Handler Behavior:

When a message is published, exactly one handler will trigger for every active subscription that matches the message (either via direct channel match or pattern match).

(def my-listener
  (car/with-new-pubsub-listener (:spec server1-conn)
    {"channel1" (fn f1 [msg] (println "f1:" msg))
     "channel*" (fn f2 [msg] (println "f2:" msg))}
   (car/subscribe "channel1")
   (car/psubscribe "channel*" "ch*")))

;; To use and update:
(car/with-open-listener my-listener
  (car/unsubscribe)
  (car/subscribe "channel3"))

;; To close:
(car/close-listener my-listener)