implements a mosquito api for reading queues and fetching job runs fr…

…om them

spec/mosquito/api/queue_spec.cr

@@ -0,0 +1,57 @@
+require "../../spec_helper"
+
+describe Mosquito::Api::JobRun do
+  let(job_classes) {
+    [QueuedTestJob, PassingJob, FailingJob, QueueHookedTestJob]
+  }
+  let(queued_test_job) { QueuedTestJob.new }
+  let(passing_job) { PassingJob.new }
+
+  it "can fetch a list of current queues" do
+    clean_slate do
+      queued_test_job.enqueue
+      passing_job.enqueue
+      expected_queues = ["queued_test_job", "passing_job"].sort
+      queues = Mosquito::Api::Queue.all
+      assert_equal 2, queues.size
+      assert_equal expected_queues, queues.map(&.name).sort
+    end
+  end
+
+  it "can fetch the size of a queue" do
+    clean_slate do
+      job_classes.map(&.new).each(&.enqueue)
+      queues = Mosquito::Api::Queue.all
+      queues.each do |queue|
+        assert_equal 1, queue.size
+      end
+    end
+  end
+
+  it "can fetch the size details of a queue" do
+    clean_slate do
+      job_classes.map(&.new).each(&.enqueue)
+      queues = Mosquito::Api::Queue.all
+      sizes = queues.map(&.size_details)
+      sizes.each do |size|
+        assert_equal 1, size["waiting"]
+        assert_equal 0, size["scheduled"]
+        assert_equal 0, size["pending"]
+        assert_equal 0, size["dead"]
+      end
+    end
+  end
+
+  it "can fetch job runs from a queue" do
+    clean_slate do
+      job_classes.each do |job_class|
+        job = job_class.new
+        job.enqueue
+        api = Mosquito::Api::Queue.new job_class.queue.name
+        job_runs = api.waiting_job_runs
+        assert_equal 1, job_runs.size
+        assert_equal job.class.name.underscore, job_runs.first.type
+      end
+    end
+  end
+end

src/mosquito/api.cr

@@ -13,6 +13,12 @@ module Mosquito::Api
   def self.job_run(id : String) : JobRun
     JobRun.new id
   end
+
+  def self.list_queues : Array(Observability::Queue)
+    Mosquito.backend.list_queues
+      .map { |name| Observability::Queue.new name }
+  end
+
   def self.list_overseers : Array(Overseer)
     Mosquito.backend.list_overseers
       .map { |name| Overseer.new name }

src/mosquito/api/queue.cr

@@ -0,0 +1,38 @@
+module Mosquito::Api
+  class Queue
+    getter name : String
+
+    private property backend : Mosquito::Backend
+
+    def self.all : Array(Queue)
+      Mosquito.backend.list_queues.map { |name| new name }
+    end
+
+    def initialize(@name)
+      @backend = Mosquito.backend.named name
+    end
+
+    {% for name in Mosquito::Backend::QUEUES %}
+      def {{name.id}}_job_runs : Array(JobRun)
+        backend.dump_{{name.id}}_q
+          .map { |task_id| JobRun.new task_id }
+      end
+    {% end %}
+
+    def size : Int64
+      backend.size(include_dead: false)
+    end
+
+    def size_details : Hash(String, Int64)
+      sizes = {} of String => Int64
+      {% for name in Mosquito::Backend::QUEUES %}
+        sizes["{{name.id}}"] = backend.{{name.id}}_size
+      {% end %}
+      sizes
+    end
+
+    def <=>(other)
+      name <=> other.name
+    end
+  end
+end

src/mosquito/redis_backend.cr

@@ -241,6 +241,22 @@ module Mosquito
           raise "don't know how to dump a #{type} for {{name.id}}"
         end
       end
+
+      def {{name.id}}_size : Int64
+        key = {{name.id}}_q
+        type = redis.type key
+
+        case type
+        when "list"
+          redis.llen(key).as(Int64)
+        when "zset"
+          redis.zcount(key, "0", "+inf").as(Int64)
+        when "none"
+          0_i64
+        else
+          raise "don't know how to {{name.id}}_size (redis type is a #{type})."
+        end
+      end
     {% end %}
 
     def scheduled_job_run_time(job_run : JobRun) : String?

src/mosquito/test_backend.cr

@@ -156,6 +156,10 @@ module Mosquito
       def dump_{{name.id}}_q : Array(String)
         [] of String
       end
+
+      def {{name.id}}_size : Int64
+        0_i64
+      end
     {% end %}
 
     def scheduled_job_run_time(job_run : JobRun) : String?